That's quite a nice little trick Endre, way better that writing an ActorPublisher if you don't need to communicate with the Producer. I did something similar to create an infinite stream from Amazon SQS (using an infinite Iterator[Unit] and mapAsyncUnordered) but this seems a much better approach.
2015-02-20 10:20 GMT+00:00 Endre Varga <[email protected]>: > 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. > -- >>>>>>>>>> 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.
