rxjava + twitter4j

gistfile1.java

public class HomePresenterImpl implements HomePresenter {
	private static final String TAG = HomePresenterImpl.class.getSimpleName();
	private final HomeScreen screen;
	private final DataProvider provider;
	private Subscription subscription1;

	public HomePresenterImpl(HomeScreenImpl screen, DataProvider dataProvider) {
		this.screen = screen;
		this.provider = dataProvider;
	}

	@Override
	public void needTimeline() {
		if (provider.isLoggedIn()) {
			Log.d(TAG, "[HomePresenterImpl - needTimeline] - (line 40): " + "is logged in");
			Observable<Status> observable = timelineObservable();
			subscription1 = observable
					.observeOn(AndroidSchedulers.mainThread())
					.subscribeOn(Schedulers.newThread())
					.subscribe(new Observer<Status>() {
						@Override
						public void onCompleted() {
							Log.d(TAG, "[HomePresenterImpl - onCompleted] - (line 48): ");
						}

						@Override
						public void onError(Throwable e) {
							Log.e(TAG, "[HomePresenterImpl - onError] - (line 53): ", e);
						}

						@Override
						public void onNext(Status status) {
							Log.d(TAG, "[HomePresenterImpl - onNext] - (line 59): " + "Status: " + status.getText());
						}
					});
		} else {
			screen.navigateToLogin();
		}
	}

	private Observable<Status> timelineObservable() {
		return Observable.create(new Observable.OnSubscribe<Status>() {
			@Override
			public void call(Subscriber<? super Status> subscriber) {
				try {
					ResponseList<Status> timeLine = provider.timeline();
					for (Status status : timeLine) {
						subscriber.onNext(status);
					}
					subscriber.onCompleted();
				} catch (TwitterException e) {
					subscriber.onError(e);
				}
			}
		});
	}
	
	@Override
	public void dispose() {
		subscription1.unsubscribe();
	}
	
	
}