Rx Programmierung, wie man Element mit früherem Element in single Observable kombinieren?

Question

Wenn wir die beobachtbare:

1 -> 2 -> 3 -> 4 -> 5 -> ... 

wie man die neuen beobachtbar zu konstruieren:

(1, 2) -> (3, 4) -> ... 

Vielleicht ist die Frage zu kurz, aber ich finde wirklich nicht, wie Sie achieve.Thank

Danke jeder, ich finde eine Methode und unter betracht entferne var

import java.util.concurrent.TimeUnit 

import rx.lang.scala.{Subject, Observable} 

import scala.concurrent.duration.Duration 
object ObservableEx { 
    implicit class ObservableImpl[T](src: Observable[T]) { 
    /** 
     * 1 -> 2 -> 3 -> 4 ->... 
     * (1,2) -> (3,4) -> ... 
     */ 
    def pair: Observable[(T, T)] = { 
     val sub = Subject[(T, T)]() 
     var former: Option[T] = None //help me to kill it 
     src.subscribe(
     x => { 
      if (former.isEmpty) { 
      former = Some(x) 
      } 
      else { 
      sub.onNext(former.get, x) 
      former = None 
      } 
     }, 
     e => sub.onError(e), 
     () => sub.onCompleted() 
     ) 

     sub 
    } 
    } 
} 

object Test extends App { 
    import ObservableEx._ 
    val pair = Observable.interval(Duration(1L, TimeUnit.SECONDS)).pair 
    pair.subscribe(x => println("1 - " + x)) 
    pair.subscribe(x => println("2 - " + x)) 

    Thread.currentThread().join() 
} 

Ich mochte Var überhaupt nicht, danke nochmal!

ENDLICH Ich bekomme einen leichten Weg, Hoffnung kann anderen helfen.

def pairPure[T](src: Observable[T]): Observable[(T, T)] = { 
    def pendingPair(former: Option[T], sub: Subject[(T, T)]): Unit = { 
    val p = Promise[Unit] 
    val subscription = src.subscribe(
     x => { 
     if (former.isEmpty) { 
      p.trySuccess(Unit) 
      pendingPair(Some(x), sub) 
     } 
     else { 
      sub.onNext(former.get, x) 
      p.trySuccess(Unit) 
      pendingPair(None, sub) 
     } 
     }, 
     e => sub.onError(e), 
     () => sub.onCompleted() 
    ) 

    p.future.map{x => subscription.unsubscribe()} 
    } 

    val sub = Subject[(T,T)]() 
    pendingPair(None, sub) 
    sub 
} 

Andere Antworten auch sehr hilfreich ~

Answer 1

Sie können tumblingBuffer mit count = 2 verwenden eine Observable von Seq s der Länge 2 und unter Verwendung von map zu bekommen, können Sie sie in Paare drehen:

implicit class ObservableImpl[T](src: Observable[T]) { 
    def pair: Observable[(T, T)] = { 
    def seqToPair(seq: Seq[T]): (T, T) = seq match { 
     case Seq(first, second) => (first, second) 
    } 
    src.tumblingBuffer(2).map(seqToPair) 
    } 
} 

Beachten Sie, dass dies, wenn die Anzahl der Elemente wird scheitern in die Quelle Observable ist ungerade, so dass Sie diesen Fall in seqToPair abdecken müssen.

Answer 2

Versuchen groupBy Operator. Viel Glück.