On Fri, Nov 20, 2015 at 12:03 PM, Florian Witteler < florian.witte...@googlemail.com> wrote:
I have an elasticsearch-database with all our articles. Each tenant has its > own index. These indexes are denormalized (the common properties of an > article are duplicated into each index) > We are about to implement streaming of articles with akka http and use the > elasticsearch-drivers' reactive streaming api. > So far so good - set aside the enthusiasm of the awesomeness of handling > breakpressure from the client through two modules back to the database ;-) > > For another project we need an export of the articles of *all* tenants in > another fashion. > In my module I'm going to query the articles for each tenant (and have a > List of akka.stream.scaladsl.Source) > > The export has to be in a format like this: > <articles> > <article code="1"> > <!-- common stuff --> > <localizations> > <localization isocode="de"> > <!-- de-stuff --> > </localization> > <localization isocode="es"> > <!-- es-stuff --> > </localization> > <localization isocode="cz"> > <!-- cz-stuff --> > </localization> > </localizations> > </article> > </articles> > The Details doesn't matter - just think of Article 1 .. n Localizations > > Now consider this constellation of articles in the streams. > > de 1 2 3 4 5 6 7 8 EOF > es 1 2 4 5 6 8 EOF > cz 1 2 4 6 7 8 EOF > > As you can see not every tenant has the same articles. > I'd like to group several streams together and generate tuples of matching > articles. What I want is s.th. like that. > > 1 -> (de,es,cz) > 2 -> (de,es,cz) > 3 -> (de) > 4 -> (de,es,cz) > 5 -> (de,es) > 6 -> (de,es,cz) > 7 -> (de,cz) > 8 -> (de,es,cz) > > I guess a GraphStage is what I need. Since the articles of each stream are > ordered in the same way, I can group them together if I look at the head > element of each stream. > Here is my pseudocode. > val sources: List[Source] //all inputStreams > val minArticleNumber = sources.map(s => s.head.articleNumber).min > val matchingStreams = sources.filter(_.head.articleNumber == > minArticleNumber) > > > //pick the current element of each of the matchingStreams, create a tuple > and push it out of this stage and advance each of the matching streams at > once. > > > Does anyone have a running example of a GraphStage with customized logic? > I’m sorry that the documentation is not ready yet. You can find plenty of implementations inside akka-streams <https://github.com/akka/akka/tree/releasing-akka-stream-and-http-experimental-2.0-M1/akka-stream/> itself. Search for extends GraphStage /Patrik Otherwise I'd like to ask for some pointers in the right direction. If I > get it done, I can provide a running example for the documentation (or a > github project) > > Thanks! > Florian > > -- > >>>>>>>>>> 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 akka-user+unsubscr...@googlegroups.com. > To post to this group, send email to akka-user@googlegroups.com. > Visit this group at http://groups.google.com/group/akka-user. > For more options, visit https://groups.google.com/d/optout. > -- Patrik Nordwall Typesafe <http://typesafe.com/> - Reactive apps on the JVM Twitter: @patriknw -- >>>>>>>>>> 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 akka-user+unsubscr...@googlegroups.com. To post to this group, send email to akka-user@googlegroups.com. Visit this group at http://groups.google.com/group/akka-user. For more options, visit https://groups.google.com/d/optout.
