RxJava अंतराल ऑपरेटर

Question

मैं RxJava ऑपरेटर के बारे में सीख रहा हूँ, और मैंने पाया नीचे इन कोड कुछ भी प्रिंट नहीं था उपयोग कैसे करें:

public static void main(String[] args) { 

    Observable 
    .interval(1, TimeUnit.SECONDS) 
    .subscribe(new Subscriber<Long>() { 
     @Override 
     public void onCompleted() { 
      System.out.println("onCompleted"); 
     } 

     @Override 
     public void onError(Throwable e) { 
      System.out.println("onError -> " + e.getMessage()); 
     } 

     @Override 
     public void onNext(Long l) { 
      System.out.println("onNext -> " + l); 
     } 
    }); 
} 

ReactiveX के रूप में, interval

एक नमूदार कि उत्सर्जन करता है बनाने किसी विशेष समय अंतराल से घिरे पूर्णांक का अनुक्रम

क्या मैंने कोई गलती की या कुछ भूल गए?

Answer 1

आप जब तक नमूदार सेवन किया जाता है ब्लॉक करने के लिए है:

public static void main(String[] args) throws Exception { 

    CountDownLatch latch = new CountDownLatch(1); 

    Observable 
    .interval(1, TimeUnit.SECONDS) 
    .subscribe(new Subscriber<Long>() { 
     @Override 
     public void onCompleted() { 
      System.out.println("onCompleted"); 
      // make sure to complete only when observable is done 
      latch.countDown(); 
     } 

     @Override 
     public void onError(Throwable e) { 
      System.out.println("onError -> " + e.getMessage()); 
     } 

     @Override 
     public void onNext(Long l) { 
      System.out.println("onNext -> " + l); 
     } 
    }); 

    // wait for observable to complete (never in this case...) 
    latch.await(); 
} 

आप नमूदार पूरा देखने के लिए उदाहरण के लिए .take(10) जोड़ सकते हैं।

Answer 2

सदस्यता लेने के बाद Thread.sleep(1000000) रखो और आप इसे काम देख पाएंगे। Observable.interval डिफ़ॉल्ट रूप से Schedulers.computation() पर संचालित होता है ताकि आपकी स्ट्रीम मुख्य थ्रेड के अलावा किसी थ्रेड पर चल रही हो।

Answer 3

जैसा कि वे आपको बताते हैं कि पहले से ही अंतराल अतुल्यकालिक काम करता है, इसलिए आपको सभी घटनाओं को समाप्त करने की प्रतीक्षा करनी होगी।

सदस्यता लेने के बाद आप सदस्यता प्राप्त कर सकते हैं और फिर टेस्टसब्रिबर का उपयोग करें जो कि रेएक्टिवएक्स प्लेटफॉर्म का हिस्सा है, और जो आपको सभी घटनाओं को समाप्त करने की प्रतीक्षा करने के लिए सुविधा देगा।

 @Test 
public void testObservableInterval() throws InterruptedException { 
    Subscription subscription = Observable.interval(1, TimeUnit.SECONDS) 
       .map(time-> "item emitted") 
       .subscribe(System.out::print, 
         item -> System.out.print("final:" + item)); 
    new TestSubscriber((Observer) subscription) 
      .awaitTerminalEvent(100, TimeUnit.MILLISECONDS); 
} 

मैं अगर तुम https://github.com/politrons/reactive/blob/master/src/test/java/rx/observables/scheduler/ObservableAsynchronous.java