Come convertire RX osservabile per giocare Enumerator

Question

Configura correttamente un WebSocket in Play utilizzando il suo costrutto Enumerator nativo, chiamando un codice che restituisce una stringa:

def operationStatusFeed = WebSocket.using[String] { implicit request =>
  val in = Iteratee.ignore[String]
  val out = Enumerator.repeatM {
   Promise.timeout(operation, 3 seconds)
  }
  (in, out)
}

.

Ora voglio che la mia funzione operation restituisca un rx.lang.scala.Observable[String] invece di una stringa, e voglio emettere qualsiasi stringa non appena entra.Come posso mappare questo osservabile a un play.api.libs.iteratee.Enumerator?

Answer 1

È possibile utilizzare la conversione implicita da Bryan Gilbert. Questo funzionerà perfettamente bene, ma fai attenzione a utilizzare Versione aggiornata delle conversioni di Bryan Gilbert ! L'annullamento dell'iscrizione non è mai chiamato nella risposta da Jeroen Kransen (e questo è cattivo!).

  /*
   * Observable to Enumerator
   */
  implicit def observable2Enumerator[T](obs: Observable[T]): Enumerator[T] = {
    // unicast create a channel where you can push data and returns an Enumerator
    Concurrent.unicast { channel =>
      val subscription = obs.subscribe(new ChannelObserver(channel))
      val onComplete = { () => subscription.unsubscribe }
      val onError = { (_: String, _: Input[T]) => subscription.unsubscribe }
      (onComplete, onError)
    }
  }

  class ChannelObserver[T](channel: Channel[T]) extends rx.lang.scala.Observer[T] {
    override def onNext(elem: T): Unit = channel.push(elem)
    override def onCompleted(): Unit = channel.end()
    override def onError(e: Throwable): Unit = channel.end(e)
  }

.

Per essere completato, ecco la conversione da Enumerator a Osservabile:

  /*
   * Enumerator to Observable
   */
  implicit def enumerator2Observable[T](enum: Enumerator[T]): Observable[T] = {
    // creating the Observable that we return
    Observable({ observer: Observer[T] =>
      // keeping a way to unsubscribe from the observable
      var cancelled = false

      // enumerator input is tested with this predicate
      // once cancelled is set to true, the enumerator will stop producing data
      val cancellableEnum = enum through Enumeratee.breakE[T](_ => cancelled)

      // applying iteratee on producer, passing data to the observable
      cancellableEnum (
        Iteratee.foreach(observer.onNext(_))
      ).onComplete { // passing completion or error to the observable
        case Success(_) => observer.onCompleted()
        case Failure(e) => observer.onError(e)
      }

      // unsubscription will change the var to stop the enumerator above via the breakE function
      new Subscription { override def unsubscribe() = { cancelled = true } }
    })
  }

.

---

.

RX per WebSeets in Play

D'altra parte, potresti notare che la maggior parte del tempo che ti occupi di iterati e gli enumeratori in gioco è quando lavori con WebSeets (come fai qui). Siamo tutti d'accordo sul fatto che i iterati sono davvero meno intuitivi che gli osservabili e questo è probabilmente il motivo per cui stai usando RX nel tuo progetto di gioco.

Da quell'osservazione, ho costruito una biblioteca chiamata widgetmanager che fa esattamente questo: integrazione di rx in gioco per sbarazzarsi della manipolazione iterata.

Usando quella libreria, il tuo codice potrebbe semplicemente essere:

def operationStatusFeed = WebSocket.using[String] { implicit request =>

  // you can optionally give a function to process data from the client (processClientData)
  // and a function to execute when connection is closed (onClientClose)
  val w = new WidgetManager()

  w.addObservable("op", operation)

  // subscribe to it and push data in the socket to the client (automatic JS callback called)
  w.subscribePush("op")

  // deals with Iteratees and Enumerators for you and returns what's needed
  w.webSocket
}

.

La biblioteca è su GitHub qui: rxplay (i contributi sono i benvenuti)

Answer 2

Ho fatto questa soluzione ispirata a Brian Gilbert :

class ChannelObserver[T](chan: Channel[T]) extends Observer[T] {
  override def onNext(arg: T): Unit = chan.push(arg)
  override def onCompleted(): Unit = chan.end()
  override def onError(e: Throwable): Unit = chan.end(e)
  override val asJavaObserver: rx.Observer[T] = new rx.Observer[T] {
    def onCompleted() {
      chan.end()
    }

    def onError(e: Throwable) {
      chan.end(e)
    }

    def onNext(arg: T) {
      chan.push(arg)
    }
  }
}

implicit def observable2Enumerator[T](obs: Observable[T]): Enumerator[T] = {
  Concurrent.unicast[T](onStart = { chan =>
      obs.subscribe(new ChannelObserver(chan))
  })
}

.

La funzione implicita converte osservabili per enumeratori senza alcun codice aggiuntivo.