I struggle in nicely defining a Source that gets its elements from an
arbitrary event stream. At the moment my code looks like this:
def watchKey[A : Reads](key: SettingKey[A])(implicit ctx:
ExecutionContext): Source[Out[A]] = {
Source(new Publisher[Out[A]] {
var requestedElems = 0L
var cancellation: sbt.client.Subscription = _
val subs = new Subscription {
def request(n: Long): Unit = {
requestedElems = n
}
def cancel(): Unit = {
cancellation.cancel()
}
}
override def subscribe(s: Subscriber[_ >: Out[A]]): Unit = {
def sendElem(elem: Out[A]) = {
requestedElems -= 1
s.onNext(elem)
}
s.onSubscribe(subs)
cancellation = client.lazyWatch(key) { (key, res) ⇒
val elem = res map (key → _)
if (requestedElems > 0)
sendElem(elem)
else
??? // TODO handle case of no requested elems
}
}
})
}
I had to define my own (incorrect) Publisher+Subscription, which seems to
me not being the right way to do this. The `lazyWatch` method takes a
function that is called each time an event occurs. Furthermore a
subscription needs to be canceled when no new events should be sent. What
abstractions does akke-streams provide to make doing this sort of thing
easier?
--
>>>>>>>>>> Read the docs: http://akka.io/docs/
>>>>>>>>>> Check the FAQ:
>>>>>>>>>> http://doc.akka.io/docs/akka/current/additional/faq.html
>>>>>>>>>> Search the archives: https://groups.google.com/group/akka-user
---
You received this message because you are subscribed to the Google Groups "Akka
User List" group.
To unsubscribe from this group and stop receiving emails from it, send an email
to [email protected].
To post to this group, send email to [email protected].
Visit this group at http://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.
