RxJava Тема переизлучит некорректный планировщик

Question

У меня есть следующий класс, который я держу в одноточечном:

public class SessionStore { 
    Subject<Session, Session> subject; 

    public SessionStore() { 
     subject = new SerializedSubject<>(BehaviorSubject.create(new Session()); 
    } 

    public void set(Session session) { 
     subject.onNext(session); 
    } 

    public Observable<UserSession> observe() { 
     return subject.distinctUntilChanged(); 
    } 
} 

В деятельности наблюдает сессию и выполнять работу сети при каждом изменении:

private Subscription init() { 
    return sessionStore 
      .observe() 
      .subscribeOn(Schedulers.io()) 
      .observeOn(AndroidSchedulers.mainThread()) 
      .flatMap(new Func1<Session, Observable<Object>>() { 
       @Override 
       public Observable<Object> call(Session session) { 
        return retrofitService.getAThing(); 
       } 
      }) 
      .subscribe(...); 
} 

Когда Я подписываюсь на хранилище сеансов, субъект испускает на io() сразу же, как это BehaviourSubject, а абонент выполняет по mainThread().

Проблема возникает, когда я звоню sessionStore.set(new AnotherSession()), но уже подписался на нее. IMO, это должно выполнить поток, определенный в init() в планировщике io(). Однако вместо этого происходит то, что поток выполняется в том же потоке, на который был вызван subject.onNext(). Результатом является NetworkOnMainThreadException, поскольку я выполняю сетевую операцию в flatMap().

Насколько я понял предметы? Я неправильно их использую? Какое правильное решение, пожалуйста?

Я также попытался заменить весь предметный подход на Observable.fromEmitter() в методе observe(), но удивительно, что выход был тем же самым.

Answer 1

заявление se, посмотрите на следующую часть из книги 'Reactive Programming with RxJava'

По умолчанию вызов onNext() в теме напрямую передается всем методам обратного вызова onNext() наблюдателя. Неудивительно, что эти методы имеют одно и то же имя. В некотором смысле вызов onNext() на Subject косвенно вызывает onNext() для каждого Подписчика.

Рекоммендовать: Если вы вызываете onNext на тему из Thread-1, он будет вызывать onNext для абонента из Thread-1. onSubscribe будет удалено.

Так первые вещи сначала: На какой нить подписки происходят на:

retrofitService.getAThing() 

Я просто предполагаю, и говорят, что это ссылающееся нить. Каким будет поток, описанный в watchOn, который является Android-UI-Loop.

Каждое значение под наблюдениемOn будет смещено от Thread-a к Thread-b, как указано планировщиком. Наблюдение должно быть на UI-Loop должно происходить прямо перед подпиской. Каждое значение, которое будет получено в подписке, будет на UI-Loop, которое не будет блокировать поток пользовательского интерфейса или конец в исключении.

Пиз взглянуть на пример кода и вывод:

class SessionStore { 
    private Subject<String, String> subject; 

    public SessionStore() { 
     subject = BehaviorSubject.create("wurst").toSerialized(); 
    } 

    public void set(String session) { 
     subject.onNext(session); 
    } 

    public Observable<String> observe() { 
     return subject 
       .asObservable() 
       .doOnNext(s -> System.out.println("Receiving value on Thread:: " + Thread.currentThread())) 
       .distinctUntilChanged(); 
    } 
} 

@Test 
public void name() throws Exception { 
    // init 
    SessionStore sessionStore = new SessionStore(); 

    TestSubscriber testSubscriber = new TestSubscriber(); 
    Subscription subscribe = sessionStore 
      .observe() 
      .flatMap(s -> { 
       return Observable.fromCallable(() -> { 
        System.out.println("flatMap Thread:: " + Thread.currentThread()); 
        return s; 
       }).subscribeOn(Schedulers.io()); 
      }) 
      .doOnNext(s -> System.out.println("After flatMap Thread:: " + Thread.currentThread())) 
      .observeOn(Schedulers.newThread()) // imagine AndroidScheduler here 
      .subscribe(testSubscriber); // Do UI-Stuff in subscribe 

    new Thread(() -> { 
     System.out.println("set on Thread:: " + Thread.currentThread()); 
     sessionStore.set("123"); 
    }).start(); 

    new Thread(() -> { 
     System.out.println("set on Thread:: " + Thread.currentThread()); 
     sessionStore.set("345"); 
    }).start(); 

    boolean b = testSubscriber.awaitValueCount(3, 3_000, TimeUnit.MILLISECONDS); 

    Assert.assertTrue(b); 
} 

Выход ::

Receiving value on Thread:: Thread[main,5,main] 
flatMap Thread:: Thread[RxIoScheduler-2,5,main] 
After flatMap Thread:: Thread[RxIoScheduler-2,5,main] 
set on Thread:: Thread[Thread-1,5,main] 
set on Thread:: Thread[Thread-0,5,main] 
Receiving value on Thread:: Thread[Thread-1,5,main] 
flatMap Thread:: Thread[RxIoScheduler-2,5,main] 
After flatMap Thread:: Thread[RxIoScheduler-2,5,main] 
Receiving value on Thread:: Thread[Thread-1,5,main] 
flatMap Thread:: Thread[RxIoScheduler-2,5,main] 
After flatMap Thread:: Thread[RxIoScheduler-2,5,main] 

Answer 2

Когда вы вызываете оператора, он влияет на весь нисходящий поток. Если вы звоните:

.observeOn(AndroidSchedulers.mainThread()) 

в неправильном месте, остальная часть потока выполняется по указанной теме.

Я предлагаю вам всегда добавить:

.subscribeOn(Schedulers.io()) 
.observeOn(AndroidSchedulers.mainThread()) 

в самом конце потока:

private Subscription init() { 
    return sessionStore 
      .observe() 
      .flatMap(new Func1<Session, Observable<Object>>() { 
       @Override 
       public Observable<Object> call(Session session) { 
        return retrofitService.getAThing(); 
       } 
      }) 
      .subscribeOn(Schedulers.io()) 
      .observeOn(AndroidSchedulers.mainThread()) 
      .subscribe(...); 
} 

Answer 3

Я думаю, что вы забываете, что ваш Subject также наблюдатель так, чтобы получить onNext для запуска на резьбе io try

public class SessionStore { 
    Subject<Session, Session> subject; 

    public UserSessionStore() { 
     subject = new SerializedSubject<>(BehaviorSubject.create(new Session())).observeOn(Schedulers.io()); 
    } 

    public void set(Session session) { 
     subject.onNext(session); 
    } 

    public Observable<UserSession> observe() { 
     return subject.distinctUntilChanged(); 
    } 
}