如何转换Rx可观察到的播放枚举器
-
21-12-2019 - |
题
我使用其本机枚举器构造成功设置了一个WebSocket,调用一些返回字符串的代码:
def operationStatusFeed = WebSocket.using[String] { implicit request =>
val in = Iteratee.ignore[String]
val out = Enumerator.repeatM {
Promise.timeout(operation, 3 seconds)
}
(in, out)
}
.
现在我希望我的operation
函数返回一个字符串而不是字符串,而且我想在进入后立即输出任何字符串。如何将其映射到rx.lang.scala.Observable[String]
?
解决方案
可以使用Bryan Gilbert的隐式转换。这将完全正常,但要小心使用更新版本的Bryan Gilbert的转换!从Jeroen Kransen(这不好!)中从未呼吁取消订阅。
/*
* 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)
}
.
要完成,这是从枚举器转换到可观察到的:
/*
* 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 for websockets在play
中 另一方面,您可以评论您在播放中处理迭代和枚举器的大部分时间是当您使用WebSockets(如您在此处)。我们都同意迭代真的不太直观,观察到,这可能是你在播放项目中使用Rx的原因。从该观察开始,我构建了一个名为 WidgetManager 的库,它确实如此:RX集成在播放中摆脱迭代操纵。
使用该库,您的代码可以简单地是:
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
}
.
图书馆在这里on github: rxplay (欢迎贡献)
其他提示
我做了这个解决方案,它受到了 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))
})
}
.
隐式函数将可观察到枚举器转换为枚举器,而无需任何其他代码。
不隶属于 StackOverflow