rx java - Android RxJava: I am not unsubscribing after onComplete -

i have class called pendingrequests. class counter volley requests. have 2 volley methods - 1 called checkopen , 1 called checkcompleted. each methods selects either open or completed records on sqlite tablet database , checks them against server.

in checkopen , checkclose, use 3 calls pendingrequests. @ start call initialize(), add volley request queue call add() , begin receive responses call subtract(). both checkopen or checkclose add() 50 , countdown (subtract()) zero.

once countdown zero, call oncomplete.

here code pendingrequests

import rx.observable; import rx.subjects.publishsubject; import rx.subjects.subject;  /**  * created admin on 6/17/15.  */ public class pendingrequests {     private static integer pendingreq;     private static subject<integer, integer> pendingobser = publishsubject.create();      public static void initialize(integer myint) {         pendingreq = myint;         pendingobser.onnext(pendingreq);     }      public static void add() {         pendingreq ++;         pendingobser.onnext(pendingreq);     }      public static void subtract() {         if (pendingreq == 0) {             return;         }         else         {             pendingreq --;             if (pendingreq == 0) {                 pendingobser.oncompleted();             }             else {                 pendingobser.onnext(pendingreq);             }          }     }      public static integer current() {         // return current value of pendingreq testing purposes         return pendingreq;     }      public static observable<integer> valuechanged() {         return pendingobser;     } } 

so above pendingobser subject , valuechanged observer. (hopefully understand correctly).

in mainactivity have 2 buttons. beside each button textview shows value coming pendingobser.

here code 2 buttons

public void checkcomplete(view view) {     //intent intent = new intent(this, resendactivity.class);     //startactivity(intent);     pendingrequests.valuechanged().subscribe(new subscriber<integer>() {         @override         public void oncompleted() {             tvpendingtotalclosed.settext("done");             system.gc();         }          @override         public void onerror(throwable e) {             tvpendingtotalclosed.settext(e.getmessage());         }          @override         public void onnext(integer integer) {             tvpendingtotalclosed.settext(string.valueof(integer));         }     });     string myresult = sewebserviceutils.checkcompletedrecord();     toast.maketext(applicationcontroller.getcontext(), myresult, toast.length_long).show(); }  public void checkopen(view view){     // called directly button btncheckopen     pendingrequests.valuechanged().subscribe(new subscriber<integer>() {         @override         public void oncompleted() {            tvpendingtotalopen.settext("done");             system.gc();         }          @override         public void onerror(throwable e) {             tvpendingtotalopen.settext(e.getmessage());         }          @override         public void onnext(integer integer) {             tvpendingtotalopen.settext(string.valueof(integer));         }     });     string myresult = sewebserviceutils.checkopenrecord();     toast.maketext(applicationcontroller.getcontext(), myresult, toast.length_long).show(); } 

i have 2 problems:

1. when call checkcomplete performs expected , end word done in tvpendingtotalclosed. however, when hit checkopen both tvpendingtotalclosed , tvpendingtotalopen count down. checkcomplete not unsubscribing

2. after hitting buttons first time, if try hit button checkopenrecords second time nothing happens. can't re-subscribe pendingrequests second time.

thanks, john