RxJava i Retrofit2: NetworkOnMainThreadException

Question

Zdaję sobie sprawę, że używam funkcji subscribeOn()/observeOn() na MainThread. Jaki jest zestaw opcji, które można przekazać do subscribeOn()? Jaki jest zestaw opcji, które można przekazać do funkcji observeOn()?

12-17 21:36:09.154 20550-20550/rx.test D/MainActivity2: [onCreate] 
12-17 21:36:09.231 20550-20550/rx.test D/MainActivity2: starting up observable... 
12-17 21:36:09.256 20550-20550/rx.test D/MainActivity2: [onError] 
12-17 21:36:09.256 20550-20550/rx.test W/System.err: android.os.NetworkOnMainThreadException 

GovService.java

import java.util.List; 
import retrofit.Call; 
import retrofit.http.GET; 
import rx.Observable; 

public interface GovService { 
    @GET("/txt2lrn/sat/index_1.json") 
    Observable<MyTest> getOneTestRx(); 
} 

MyTest.java

public class MyTest { 
    private String name, url; 
    private int num; 

    public String getName() { 
     return name; 
    } 

    public void setName(String name) { 
     this.name = name; 
    } 

    public String getUrl() { 
     return url; 
    } 

    public void setUrl(String url) { 
     this.url = url; 
    } 

    public int getNum() { 
     return num; 
    } 

    public void setNum(int num) { 
     this.num = num; 
    } 

    @Override 
    public String toString() { 
     return "Name: " + this.name + ", num: " + this.num + ", url: " + this.url; 
    } 
} 

MainActivity2.java

import android.os.Bundle; 
import android.support.v7.app.AppCompatActivity; 
import android.support.v7.widget.DefaultItemAnimator; 
import android.support.v7.widget.LinearLayoutManager; 
import android.support.v7.widget.RecyclerView; 
import android.support.v7.widget.Toolbar; 
import android.util.Log; 

import retrofit.GsonConverterFactory; 
import retrofit.Retrofit; 
import retrofit.RxJavaCallAdapterFactory; 
import rx.Observable; 
import rx.Subscriber; 
import rx.android.schedulers.AndroidSchedulers; 
import rx.schedulers.Schedulers; 

public class MainActivity2 extends AppCompatActivity { 
    private final String TAG = getClass().getSimpleName(); 

    @Override 
    protected void onCreate(Bundle savedInstanceState) { 
     super.onCreate(savedInstanceState); 
     Log.d(TAG, "[onCreate]"); 
     setContentView(R.layout.activity_main); 
     Toolbar toolbar = (Toolbar) findViewById(R.id.toolbar); 
     setSupportActionBar(toolbar); 
     RecyclerView mRV = (RecyclerView) findViewById(R.id.rv); 
     mRV.setLayoutManager(new LinearLayoutManager(this));// setup LayoutManager 
     mRV.setItemAnimator(new DefaultItemAnimator());// setup ItemAnimator 

     // setup retrofit 
     Retrofit retrofit = new Retrofit.Builder() 
       .baseUrl("http://goanuj.freeshell.org") 
       .addConverterFactory(GsonConverterFactory.create()) 
       .addCallAdapterFactory(RxJavaCallAdapterFactory.create()) 
       .build(); 
     GovService service = retrofit.create(GovService.class); 

     Log.d(TAG, "starting up observable..."); 
     Observable<MyTest> o = service.getOneTestRx(); 
     o.subscribeOn(Schedulers.io()); 
     o.observeOn(AndroidSchedulers.mainThread()); 
     o.subscribe(new Subscriber<MyTest>() { 
      @Override 
      public void onCompleted() { 
       Log.d(TAG, "[onCompleted] "); 
      } 

      @Override 
      public void onError(Throwable t) { 
       Log.d(TAG, "[onError] "); 
       t.printStackTrace(); 
      } 

      @Override 
      public void onNext(MyTest m) { 
       Log.d(TAG, "[onNext] " + m.toString()); 
      } 
     }); 
    } 
} 

Answer 1

Przepisz ostatnią część swojego kodu:

service.getOneTestRx() 
    .subscribeOn(Schedulers.io()) 
    .observeOn(AndroidSchedulers.mainThread()) 
    .subscribe(new Subscriber<MyTest>() { 
     @Override 
     public void onCompleted() { 
      Log.d(TAG, "[onCompleted] "); 
     } 

     @Override 
     public void onError(Throwable t) { 
      Log.d(TAG, "[onError] "); 
      t.printStackTrace(); 
     } 

     @Override 
     public void onNext(MyTest m) { 
      Log.d(TAG, "[onNext] " + m.toString()); 
     } 
    }); 

Ważna uwaga z @akarnokd:

Warto wspomnieć, że jeden musi łańcuchu połączenia jak tu, bo Obserwowalne nie jest wzorzec konstruktora (w którym modyfikuje się ustawienia istniejącego obiektu)

Answer 2

Należy zadzwonić pod numer Observable.unsubscribeOn(Schedulers.io()), retrofit, który zrezygnuje z subskrypcji na końcu żądania http.

W RxJavaCallAdapterFactory z retrofit-rxjava-adapter

nim działań, takich jak ten.

if (!subscriber.isUnsubscribed()) { 
    subscriber.onCompleted(); 
} 

Ale kiedy subscriber jest to SafeSubscriber, będzie to nazwać unsubscribe wreszcie.

Mam ten problem w mojej aplikacji.

Pełny kod:

o.subscribeOn(Schedulers.io()) 
      .observeOn(AndroidSchedulers.mainThread()) 
      .unsubscribeOn(Schedulers.io());