0

観測可能なものがある場合: 

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

新しいオブザーバブルを構築する方法: 

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

短い質問かもしれませんが、達成する方法が本当にわかりません。ありがとうございます

ありがとう、私は方法を見つけて、変数の削除を検討しています 

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()
}

私は var がまったく好きではありませんでした。

最後 に、私は光の道を手に入れました.希望は他の人を助けることができます.

 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
 }

他の回答もとても参考になりました〜

4

2 に答える 2

0

とtumblingBufferを使用して長さ 2 の を取得し、 を使用してcount = 2それらをペアに変えることができます。ObservableSeqmap

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)
  }
}

ソース Observable の要素数が奇数の場合、これは失敗することに注意してください。したがって、このケースについては で説明する必要がありますseqToPair。

于 2016-06-06T16:45:03.210 に答える