rxjava + facebook

FacebookHomeProvider.java

public class FacebookHomeProvider {
	protected final PublishSubject<FbPost> behaviorSubject;
	private Request request;

	public FacebookHomeProvider() {
		behaviorSubject = PublishSubject.create();
		behaviorSubject.subscribeOn(Schedulers.io());
	}

    /**
     * Observable for this subject
     * */
	public Observable<FbPost> getObservable() {
		return behaviorSubject.asObservable();
	}


    /**
     * Call nextHomePage() every time a new page is needed.
     * */
	public void nextHomePage() {
		if (request == null) {
			Bundle params = new Bundle();
			params.putString("fields", "id, from, message, picture, link, type, created_time");
			request = new Request(Session.getActiveSession(), "/me/home", params, HttpMethod.GET);
		}
		request.setCallback(new Request.Callback() {
			@Override
			public void onCompleted(Response response) {
				FbHome fbPosts = new Gson().fromJson(response.getGraphObject().getInnerJSONObject().toString(), FbHome.class);
				List<FbPost> data = fbPosts.getData();
				for (FbPost fbPost : data) {
					behaviorSubject.onNext(fbPost);
				}
				behaviorSubject.onCompleted();
				
				// save cursor/request to next page
				request = response.getRequestForPagedResults(Response.PagingDirection.NEXT);
			}
		});
		request.executeAsync();
	}
}