RxJava2 Flowable.create blockingSubscribeセマンティクスとBackpressureStrategy

Question

Flowable BackpressureStrategyがblockingSubscribeメソッドとどのように機能するかを理解するのが難しいです。誰かが私にそれを説明することができたらうれしいです。

私はこのコードを現在のトランクのFlowableTestsファイルでテストしていました。

@Test 
public void testCreateBackpressureDrop() { 
    Flowable.create(new FlowableOnSubscribe<Integer>() { 
     @Override 
     public void subscribe(FlowableEmitter<Integer> e) throws Exception { 
      e.onNext(1); 
      e.onNext(3); 
      e.onNext(4); 
      e.onComplete(); 
     } 
    }, BackpressureStrategy.DROP).blockingSubscribe(w); 

    verify(w, times(1)).onNext(1); 
    verify(w, times(1)).onNext(3); 
    verify(w, times(1)).onNext(4); 
    verify(w, times(1)).onComplete(); 
} 

私はsubscribe(w)とBackpressureStragegy.DROPまたはBackpressure.BUFFERテストパスを使用している場合。しかし、私がblockingSubscribe(w)を使用した場合、が呼び出されなかったと言うと、Backpressure.DROPが失敗して、Backpressure.BUFFERが失敗します。

ありがとうございます！

Answer 1

これはSubscriberを模擬するためにMockitoを使った典型的な問題です：あなたはそのonSubscribeでrequest(N)を呼び出す必要があり：

@SuppressWarnings("unchecked") 
public static <T> Subscriber<T> mockSubscriber() { 
    Subscriber<T> w = mock(Subscriber.class); 

    Mockito.doAnswer(new Answer<Object>() { 
     @Override 
     public Object answer(InvocationOnMock a) throws Throwable { 
      Subscription s = a.getArgument(0); 
      s.request(Long.MAX_VALUE); 
      return null; 
     } 
    }).when(w).onSubscribe((Subscription)any()); 

    return w; 
} 

blockingSubscribeとの合併症がFlowableOnSubscribeが実行された後に、それは上w.onSubscribeを実行していることです。