truncate on subscribe

MergedSubject.scala

package org.snippets

import rx.lang.scala._
import rx.lang.scala.subjects.ReplaySubject

import scala.concurrent.Promise
import scala.concurrent.ExecutionContext.Implicits.global

object MergedSubject {

  def main(args: Array[String]) {

    val r = ReplaySubject[Int]()

    r.onNext(1)
    r.onNext(2)
    r.onNext(3)

    val promise = Promise[List[Int]]()

    Observable.defer(Observable[Int] { observer =>
      r.doOnNext { e =>
        println(s"next $e")
        observer.onNext(e)
      }.subscribe()
      observer.onCompleted()
    })
      .subscribeOn(schedulers.NewThreadScheduler()).toList
      .subscribe { l =>
        println(s"total $l")
        promise.success(l)
      }

    promise.future map println

    r.onNext(4)

    Thread.sleep(2000)
  }

}