RxJava Наблюдаемые

Question

Я пытаюсь работать на моем первом примере RxJava

У меня есть основной вид деятельности с текстового поля и три кнопки. Первая кнопка инициализирует целое число в отдельном классе. Вторая кнопка подписывается на наблюдаемое, которое, как предполагается, должно наблюдать целое число. Третья кнопка уменьшает значение целого на единицу.

Вот мой код

package com.someurl.www.myobservable; 

import android.support.v7.app.ActionBarActivity; 
import android.os.Bundle; 
import android.view.Menu; 
import android.view.MenuItem; 
import android.view.View; 
import android.widget.Button; 
import android.widget.TextView; 

import rx.Observable; 
import rx.Subscriber; 
import rx.android.schedulers.AndroidSchedulers; 
import rx.functions.Action1; 
import rx.schedulers.Schedulers; 


public class MainActivity extends ActionBarActivity { 

    TextView tvDisplay; 

    Button btnInitialze; 
    Button btnSubscribeClass; 
    Button btnChangeInt; 

    @Override 
    protected void onCreate(Bundle savedInstanceState) { 
     super.onCreate(savedInstanceState); 
     setContentView(R.layout.activity_main); 

     tvDisplay = (TextView)findViewById(R.id.textView); 

     btnInitialze = (Button)findViewById(R.id.btnInitialize); 
     btnSubscribeClass = (Button)findViewById(R.id.btnSubscribeClass); 
     btnChangeInt = (Button)findViewById(R.id.btnChangeInt); 



     btnInitialze.setOnClickListener(new View.OnClickListener() { 
      @Override 
      public void onClick(View v) { 
       myObserve.InitializeBigInt(6); 
      } 
     }); 

     btnSubscribeClass.setOnClickListener(new View.OnClickListener() { 
      @Override 
      public void onClick(View v) { 
       //subClassNow(); 
       subJust(); 
      } 
     }); 

     btnChangeInt.setOnClickListener(new View.OnClickListener() { 
      @Override 
      public void onClick(View v) { 
       int myNewInt = myObserve.DecreaseBigInt(); 
       tvDisplay.append("\nFrom Button " + String.valueOf(myNewInt)); 
      } 
     }); 

    } 



    @Override 
    public boolean onCreateOptionsMenu(Menu menu) { 
     // Inflate the menu; this adds items to the action bar if it is present. 
     getMenuInflater().inflate(R.menu.menu_main, menu); 
     return true; 
    } 

    @Override 
    public boolean onOptionsItemSelected(MenuItem item) { 
     // Handle action bar item clicks here. The action bar will 
     // automatically handle clicks on the Home/Up button, so long 
     // as you specify a parent activity in AndroidManifest.xml. 
     int id = item.getItemId(); 

     //noinspection SimplifiableIfStatement 
     if (id == R.id.action_settings) { 
      return true; 
     } 

     return super.onOptionsItemSelected(item); 
    } 

    Observable<String> mObservable = Observable.create(new Observable.OnSubscribe<String>() { 
     @Override 
     public void call(Subscriber<? super String> subscriber) { 
      subscriber.onNext("Hello, World!"); 
      subscriber.onCompleted(); 
     } 
    }); 

    public void subNow() { 
     mObservable.subscribe(new Subscriber<String>() { 

      @Override 
      public void onCompleted() { 
       tvDisplay.append("\nDone!"); 

      } 

      @Override 
      public void onError(Throwable e) { 
       tvDisplay.append("Error! \n" + e.getMessage()); 
      } 

      @Override 
      public void onNext(String s) { 
       tvDisplay.append(s); 
      } 
     }); 
    } 

    private void subClassNow() { 

     myObserve.mObservableClass.subscribe(new Subscriber<Integer>() { 
      @Override 
      public void onCompleted() { 
       tvDisplay.append("\nClass Done!"); 
      } 

      @Override 
      public void onError(Throwable e) { 
       tvDisplay.append("Class Error! \n" + e.getMessage()); 
      } 

      @Override 
      public void onNext(Integer myInt) { 
       tvDisplay.append("\nClass " + String.valueOf(myInt)); 
      } 

     }); 
    } 

    private void subJust() { 
     myObserve.newObservableClass.subscribe(new Subscriber<Integer>() { 
      @Override 
      public void onCompleted() { 
       tvDisplay.append("\nClass Done!"); 
      } 

      @Override 
      public void onError(Throwable e) { 
       tvDisplay.append("Class Error! \n" + e.getMessage()); 
      } 

      @Override 
      public void onNext(Integer myInt) { 
       tvDisplay.append("\nClass " + String.valueOf(myInt)); 
      } 
     }); 
    } 
} 

А класс

package com.someurl.www.myobservable; 

import rx.Observable; 
import rx.Subscriber; 

/** 
* Created by Admin on 6/21/15. 
*/ 
public class myObserve { 
    static Integer myBigInt; 

    public static Observable<Integer> mObservableClass = Observable.create(new Observable.OnSubscribe<Integer>() { 
     @Override 
     public void call(Subscriber<? super Integer> subscriber) { 
      subscriber.onNext(myBigInt); 
      //subscriber.onCompleted(); 
     } 

    }); 

    public static void InitializeBigInt(Integer myInt){ 
     myBigInt = myInt; 
    } 

    public static Integer DecreaseBigInt(){ 
     myBigInt --; 
     return myBigInt; 
    } 

    public static Observable<Integer> newObservableClass = Observable.just(myBigInt); 

} 

Когда я попытался подписаться, используя mObservableClass он просто дает мне значение myBigInt (который был 6), а затем он дает мне onComplete Сделано!

Затем я попытался с помощью newObservableClass, думая, что мне нужно использовать, только вместо .create, но затем я получаю возвращение нуля для myBigInt, а затем снова onComplete Готово!

Может кто-нибудь помочь мне начать работу в правильном направлении, как наблюдать за изменением стоимости myBigInt. В идеале я хотел бы посмотреть значение myBigInt, пока оно не уменьшится до нуля (0), а затем при нулевом вызове onComplete Done!

Спасибо, Джон

Answer 1

Как правило, вы хотели бы использовать BehaviorSubject для хранения значения и передать в него изменения. Тем не менее, вам, похоже, нужна также способность к уменьшению количества атомов. Попробуйте следующее:

public class AtomicBehaviorSubject { 
    private static final AtomicInteger value = new AtomicInteger(); 
    private static final Subject<Integer, Integer> setter = 
      BehaviorSubject.<Integer>create().toSerialized(); 

    public static void setValue(int newValue) { 
     value.set(newValue); 
     setter.onNext(newValue); 
    } 

    public static void decrementValue() { 
     for (;;) { 
      int curr = value.get(); 
      if (curr == 0) { 
       return; 
      } 
      int u = curr - 1; 
      if (value.compareAndSet(curr, u)) { 
       if (u == 0) { 
        setter.onCompleted(); 
       } else { 
        setter.onNext(u); 
       } 
       return; 
      } 
     } 
    } 

    public static Observable<Integer> valueChanged() { 
     return setter; 
    } 
}