0

如果我们有可观察的： 

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

如何构造新的 observable： 

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

也许问题很短，但我真的找不到如何实现。谢谢

谢谢大家，我找到了一种方法并考虑删除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()
}

我根本不喜欢 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

尝试groupBy运算符。祝你好运。

于 2016-06-06T09:53:45.047 回答
0

您可以使用tumblingBuffer来count = 2获得长度为 2Observable的Seqs，使用map，您可以将它们变成对：

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 回答