java - PublishSubject doesn't do anything when calling onNext -

here code using publishsubject.

public class rxpager {   private int startpage; //starting page number private final int initialpagecount; //number of pages provide when requestnext() called private final publishsubject<integer> requests = publishsubject.create(); private int requestedcount; //basically keeps track of last page requested  public rxpager() {     this(1, 1); // default start @ page #1 }  public rxpager(int startpage, int initialpagecount) {     this.initialpagecount = initialpagecount;     this.startpage = startpage; }  public void decrementrequestedcount(){     requestedcount--; }  public void requestnext(int page) {     requests.onnext(page); }  public observable<integer> pages() {     return requests             .concatmap(new func1<integer, observable<integer>>() {                 @override                 public observable<integer> call(integer targetpage) {                     //check if have provided page -- if targetpager less requestedcount means page has been provided                     if (targetpage <= requestedcount) {                         return observable.empty();                     }                     else {                         //returns each page 1 one , adds offset them                         return observable.range(requestedcount, targetpage - requestedcount)                                 .flatmap(new func1<integer, observable<integer>>() {                                     @override                                     public observable<integer> call(integer page) {                                         return observable.just(page + startpage); // offset page number                                     }                                 });                     }                 }             })             .startwith(                     observable.range(0, initialpagecount)                             .flatmap(new func1<integer, observable<integer>>() {                                 @override                                 public observable<integer> call(integer page) {                                     return observable.just(page + startpage); // offset page number                                 }                             })             )             .doonnext(new action1<integer>() {                 @override                 public void call(integer it) {                     //make sure increment requested count same page can't requested                     requestedcount = - startpage + 1; //need offset here                 }             }); }  } 

i trying create paging mechanism can call rxpager.requestnext(page) , using value returned, use paging api. however, if api returns error call rxpager.decrementrequestedcount() , rerequest same page using this: rxpager.requestnext(samepage). however, seems when nothing happens. never hits if (targetpage <= requestedcount)