Hi Tim, This looks great - I was just thinking of implementing something like this myself, so the timing couldn't have been better. ^_^ I do have a couple of questions, though.
>From what I see in AsyncAckingConsumer, the default error handling strategy is to acknowledge the message once the retry limit has been reached. This means that some messages could, theoretically, be lost (no at-least-once-delivery). Have you considered rejecting the message instead as a default, or providing another built-in strategy that does that? It's not a problem to implement it independently, but it could be a bit surprising for the user. Regarding configuration, is there any way of configuring the connection dynamically? I couldn't find anywhere in the code that overrides the settings read from the default config file. For example, in my use case I have to be able to open connections to several different RabbitMQ clusters, and it doesn't seem to be possible with the current implementation. As a side note, is it possible to change the configuration element to something a little less general (maybe rabbit-op.rabbitmq)? Finally, a couple of things regarding the stream module: >From my understanding of streams, creating a Source and creating a flow should not depend on the actual RabbitMQ connection/subscription. Instead, the subscription (and possible the connection/channel as well?) should only be created once the flow gets materialized. Have you considered using Source.actorPublisher(Props) to create the actor and subscribe in the actor's preStart or something? The other thing about streams ties in with Roland's comments (I think) about the use of Futures with streams. It means that the entire flow must now be aware of the fact that it's a RabbitMQ flow (or at the very least, that its messages contain the Future), so it is not as composable as it might have been otherwise. Also, I don't see how it plays with streams' error handling mechanism / strategies. At the same time, the use of Futures to "track" messages is very elegant, and I don't see any easy way of achieving something similar with streams (maybe something using BidiFlow?). In any case, like I already said, this looks like a very nice library. If you need any help with it, please let me know - I would love to contribute to it. Tal On Monday, May 11, 2015 at 7:23:50 AM UTC+3, Tim Harper wrote: > > I have developed a high-level library for efficiently setting up > resilient, fault-tolerant RabbitMQ consumers using Akka and Akka Reactive > Streams. > > Some of the features: > > - Recovery: > - Consumers automatically reconnect and subscribe if the connection is > lost > - Messages published can optionally > - Integration > - Connection settings pulled from Typesafe config library > - Asyncronous, concurrent consumption using Scala native Futures or > the new Akka Streams project. > - Common pattern for serialization allows easy integration with > serialization libraries such play-json or json4s > - Common pattern for exception handling to publish errors to Airbrake, > Syslog, or all of the above > - Modular > - Composition favored over inheritance enabling flexible and high code > reuse. > - Modeled > - Queue binding, exchange binding modeled with case classes > - Publishing mechansims also modeled > - Reliability > - Builds on the excellent [Akka RabbitMQ client]( > https://github.com/thenewmotion/akka-rabbitmq) library for easy recovery. > - Built-in consumer error recovery strategy in which messages are > re-delivered to the message queue and retried (not implemented for > akka-streams integration as retry mechanism affects message order) > - With a single message, pause all consumers if service health check > fails (IE: database unavailable); easily resume the same. > - Graceful shutdown > - Consumers and streams can immediately unsubscribe, but stay alive > long enough to wait for any messages to finish being processed. > - Tested > - Extensive integration tests > > The source is available here: https://github.com/SpinGo/op-rabbit > > We have been using the library internally at SpinGo for a year and I am > working towards a 1.0.0 release candidate. We're using the streaming > integration as the foundation for a billing system which is heavily based > on reliable message-order, and at-least-once-delivery guarantees. I'm > rather excited to share it with the world, and would be grateful for > feedback. I plan on creating an Activator project to help people learn the > library quickly. > > Some examples are on the github page. More examples can be found in the > tests. > > Feedback, is of course, appreciated. > > Tim > -- >>>>>>>>>> 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.
