Observable im Zeitintervall erstellt

Question

Ich wurde an den RxScala Observablen suchen, die in einem bestimmten Zeitintervall erstellt werden:

val periodic: Observable[Long] = Observable.interval(100 millis) 

periodic.foreach(x => println(x)) 

Wenn ich das in einem Arbeitsblatt setzen, bekomme ich dieses Ergebnis:

periodic: rx.lang.scala.Observable[Long] = rx.lang.scala.JavaConversions$$anon$2@2cce3493 

res0: Unit =() 

Das lässt mich verwirrt: Was enthalten die Elemente von periodic eigentlich?

Haben sie einen Index? Enthalten sie das Zeitintervall, in dem sie erstellt wurden?

Answer 1

Wie Sie hier lesen können http://reactivex.io/documentation/operators/interval.html produzierten Elemente sind Werte inkrementieren von 0.

Wie für Ihren Code und die Ergebnisse:

Hier erstellen Sie die beobachtbaren und erhalten Observable[Long]-periodic zugewiesen. Alles wie erwartet.

scala> val periodic: Observable[Long] = Observable.interval(100 millis) 
periodic: rx.lang.scala.Observable[Long] = rx.lang.scala.JavaConversions$$anon$2@2cce3493 

Hier registrieren Sie einen Rückruf, d. H. Was passiert, wenn der Wert emitiert wird. Der Rückgabetyp der foreach Methode ist Unit, da es keinen vernünftigen Wert hat und nur für den Nebeneffekt der Registrierung von Rückrufen auftritt.

periodic.foreach(x => println(x)) 
res0: Unit =() 

Sie können keine tatsächlichen Werte anzeigen, da die Ausführung gestoppt wird. Versuchen Sie Thread.sleep einzufügen.

val periodic: Observable[Long] = Observable.interval(100.millis) 
periodic.foreach(x => println(x)) 
Thread.sleep(1000) 

Gibt Ausgabe ähnlich

periodic: rx.lang.scala.Observable[Long] = rx.lang.scala.JavaConversions$$anon$2@207cb62f 

res0: Unit =() 

0 
1 
2 
3 
4 
5 
6 
7 
8 
9 
res1: Unit =() 

Answer 2

Das Problem ist das Intervall asynchron ist, so Passende nicht auf das Ergebnis warten.

Eine weitere Möglichkeit, das Ergebnis zu warten, ist die Verwendung TestSubscriber

def interval(): Unit = { 
addHeader("Interval observable") 
Observable.interval(createDuration(100)) 
    .map(n => "New item emitted:" + n) 
    .doOnNext(s => print("\n" + s)) 
    .subscribe(); 
new TestSubscriber[Subscription].awaitTerminalEvent(1000, TimeUnit.MILLISECONDS); 
} 

Sie können hier weitere Beispiele siehe https://github.com/politrons/reactiveScala