Hi Simon,
One trick I like to use is to define a Source in terms of a PushPullStage.
Now this sounds strange, since a PushPullStage is supposed to be someting
that transforms incoming element into outgoing elements, how can that be a
Source? Well, the trick is this:
def mySource = Source.empty.transform(...)
Since the upstream Source of the stage is immediately completed one, you
can call ctx.absorbTermination() and then only handle element emission from
onPull. Of course you have to propagate the first pull upstream. For
example:
Source.empty.transform(() ⇒ {
new PushPullStage[Nothing, T] {
val iterator: Iterator[T] = myIterator
// Upstream is guaranteed to be empty
override def onPush(elem: Nothing, ctx: Context[T]): Directive =
throw new UnsupportedOperationException("The IterableSource stage
cannot be pushed")
override def onUpstreamFinish(ctx: Context[T]): TerminationDirective = {
if (iterator.hasNext) ctx.absorbTermination()
else ctx.finish()
}
override def onPull(ctx: Context[T]): Directive = {
if (!ctx.isFinishing) {
ctx.pull()
} else {
val elem = iterator.next()
if (iterator.hasNext) ctx.push(elem)
else ctx.pushAndFinish(elem)
}
}
}
The above is a simplified version of the new upcoming iterator source in M4.
I guess this pattern can be made proper by a simple DSL that handles the
boilerplate. Currently the drawback of this approach is that you cannot
send external async events to this kind of source, so it is a bit limited
for now, but we will solve that in the future.
-Endre
On Thu, Feb 19, 2015 at 11:41 PM, Simon Schäfer <[email protected]> wrote:
> 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.
>
--
>>>>>>>>>> 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.
