Added support for Ampq v1.0 #38

AndreSteenbergen · 2018-07-17T13:55:46Z

No description provided.

Aaronontheweb · 2018-07-17T19:48:36Z

@AndreSteenbergen so is this going to publish as its own independent package?

AndreSteenbergen · 2018-07-17T19:54:43Z

I think it should, as amqp 0.9 (RabbitMq and the likes) and amqp 1.0 are incompatible. With RabbitMQ being quite popular, I don't think it is wise to remove 0.9 support.

AndreSteenbergen · 2018-07-17T19:59:08Z

BTW this pull request is just a first start to support AMQP 1.0. I hope good feedback can be given to the code, I hope this contributes something .

AndreSteenbergen · 2018-07-17T20:00:17Z

This pull request was made to support this: #37

Aaronontheweb · 2018-07-17T20:17:59Z

Ah, got it.

CC @bobanco - any thoughts on this?

marcpiechura · 2018-07-17T20:23:23Z

Amqp.V1/Akka.Streams.Amqp.V1/AmqpSinkStage.cs

+    public sealed class AmqpSinkStage<T> : GraphStageWithMaterializedValue<SinkShape<T>, Task>
+    {
+        public readonly Inlet<T> In = new Inlet<T>("AmqpSink.in");
+        public override SinkShape<T> Shape => new SinkShape<T>(In);


It's most likely that the getter is called multiple times during materialization, i.e. wenn calling .Run, so this should be set once in the constructor.

marcpiechura · 2018-07-17T20:25:38Z

Amqp.V1/Akka.Streams.Amqp.V1/AmqpSinkStage.cs

+                    try
+                    {
+                        var elem = Grab(stage.In);
+                        sender.Send(new Message(amqpSinkStage.AmpqSourceSettings.GetBytes(elem)));


Is this a blocking call ?

There is an async version. but should I await it, or just continue?

marcpiechura · 2018-07-17T20:26:37Z

Amqp.V1/Akka.Streams.Amqp.V1/AmqpSinkStage.cs

+                        Pull(stage.In);
+                    } catch (Exception e)
+                    {
+                        Debugger.Break();


Do you forgot to remove it or because it's still WIP ?

marcpiechura · 2018-07-17T20:28:54Z

Amqp.V1/Akka.Streams.Amqp.V1/AmqpSinkStage.cs

+
+            public override void PostStop()
+            {
+                promise.TrySetException(new ApplicationException("stage stopped unexpectedly"));


If the stream fails onUpstreamFailure is called, so I'm not sure why we should try to "throw" another exception from her.

Anyways, why ApplicationException? I believe MS encouraged to not use it many years ago.

marcpiechura · 2018-07-17T20:29:16Z

Amqp.V1/Akka.Streams.Amqp.V1/AmqpSourceStage.cs

+            AmqpSourceSettings = amqpSourceSettings;
+        }
+
+        public override SourceShape<T> Shape => new SourceShape<T>(Out);


As above, set once via constructor

marcpiechura · 2018-07-17T20:38:01Z

Amqp.V1/Akka.Streams.Amqp.V1/AmqpSourceStage.cs

+                        try
+                        {
+                            Push(outlet, ampqSourceSettings.Convert(message));
+                            receiver.Accept(message);


Is it a good idea to immediately accept the message ?
What if I would want to accept it at the end of the stream because processing could fail later in the chain and if so I want to try to process it a second time?

How would that work? Sending a reference to the receiverlink with the message.

marcpiechura · 2018-07-17T20:39:35Z

Amqp.V1/Akka.Streams.Amqp.V1/AmqpSourceStage.cs

+                        }
+                        catch (Exception e)
+                        {
+                            receiver.Reject(message, new Error(new Symbol(e.Message)));


Not sure about the semantics of Reject but would it maybe make sense to sync it with the result of the decider ?
E.g. if decider returns Resume or Restart one maybe doesn't want to reject the message but instead try it another time ?

marcpiechura · 2018-07-17T20:42:59Z

Amqp.V1/Akka.Streams.Amqp.V1/Dsl/AmpqSink.cs

+{
+    public static class AmpqSink<T>
+    {
+        public static Sink<T, Task> Create(IAmpqSinkSettings<T> sourceSettings)


I would move the T into the method definition, in theory you should be able to call AmpqSink.Create(settings) because the T could be inferred from the settings.

marcpiechura · 2018-07-17T20:43:17Z

Amqp.V1/Akka.Streams.Amqp.V1/Dsl/AmpqSource.cs

+{
+    public static class AmpqSource<T>
+    {
+        public static Source<T, NotUsed> Create(IAmqpSourceSettings<T> sourceSettings)


As above, move the T

marcpiechura · 2018-07-17T20:44:21Z

Amqp.V1/Akka.Streams.Amqp.V1/NamedQueueSourceSettings.cs

+            this.serializer = serializer;
+        }
+
+        public T Convert(Message message) {


Move { into the next line ;-)

marcpiechura · 2018-07-17T20:47:43Z

@AndreSteenbergen left you a few remarks,. One more general question, aren't there any async methods for the queue, e.g. ReceiveAsync, AcceptAsync ... ?

vasily-kirichenko · 2018-07-18T07:49:57Z

Amqp.V1/Akka.Streams.Amqp.V1.Tests/AmqpConnectorsTest.cs

+        public AmqpConnectorsTest()
+        {
+            materializer = ActorMaterializer.Create(Sys);
+            var serialization = Sys.Serialization;


Why this variable?

vasily-kirichenko · 2018-07-18T07:55:25Z

Amqp.V1/Akka.Streams.Amqp.V1/AmqpSourceStage.cs

+            public override void PostStop()
+            {
+                base.PostStop();
+                receiver.Close();


Are you sure we should call the base method first?

vasily-kirichenko · 2018-07-18T07:58:34Z

Amqp.V1/Akka.Streams.Amqp.V1/Util/AttributesExtension.cs

+        public static Decider GetDeciderOrDefault(this Attributes attributes)
+        {
+            var attr = attributes.GetAttribute<ActorAttributes.SupervisionStrategy>(null);
+            return attr != null ? attr.Decider : Deciders.StoppingDecider;


return attr?.attr.Decider ?? Deciders.StoppingDecider?

AndreSteenbergen · 2018-07-18T12:40:50Z

Tested the latest commit using these brokers:

TestBroker from amqplite
ActiveMQ 5.15.4
ActiveMQ Artemis 2.6.2

All seem to work now.

Added an event handler to the subscription to raise event when topic eof reached

…m actions" This reverts commit 44e63db.

AndreSteenbergen · 2018-08-13T21:28:52Z

@marcpiechura What do you think after the updates?

Aaronontheweb · 2018-08-24T21:30:31Z

@AndreSteenbergen @marcpiechura @vasily-kirichenko any word on this? I'd like to push out the Akka.Streams.Kafka package but if this is ready for merge, we can do both at the same time.

vasily-kirichenko · 2018-08-25T05:30:28Z

Do the two tests cover all this PR functionality?

AndreSteenbergen · 2018-09-12T18:04:18Z

Sorry was away on vacation; I think all functionality is captured. With amqp 1 the server implements logic, instead of the client. I couldn't think of more tests besides receiving what was send.

Horusiath · 2018-09-25T05:47:22Z

@marcpiechura what is the status of this PR? Could we merge it?

Danthar · 2019-01-08T20:29:44Z

Im merging this :P

Aaronontheweb · 2019-01-09T14:40:05Z

@Danthar thank you

https://www.youtube.com/watch?v=SxMPTEeoShU

Let's prep a new release here, along with the updated Kafka stuff if that's ready to go

Added support for Ampq v1.0

653651f

marcpiechura suggested changes Jul 17, 2018

View reviewed changes

vasily-kirichenko reviewed Jul 18, 2018

View reviewed changes

AndreSteenbergen added 2 commits July 18, 2018 11:04

Changes made after review

a595b72

Added support for ActiveMQ Artemis queues.

4f0d743

AndreSteenbergen added 4 commits July 24, 2018 19:36

Merge remote-tracking branch 'upstream/dev' into dev

a5de244

Made caal sync again in test.

e8dadff

Updates interface, so no switch statement is needed to perform actions

44e63db

Added an event handler to the subscription to raise event when topic eof reached

Revert "Updates interface, so no switch statement is needed to perfor…

73c359d

…m actions" This reverts commit 44e63db.

Updated NuGet packget

4eaaad2

Danthar merged commit c6754db into akkadotnet:dev Jan 8, 2019

Aaronontheweb mentioned this pull request Apr 20, 2022

[Docs] need to explain when to use Akka.Streams.RabbitMQ vs Akka.Streams.AMQP #963

Open

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Added support for Ampq v1.0 #38

Added support for Ampq v1.0 #38

AndreSteenbergen commented Jul 17, 2018

Aaronontheweb commented Jul 17, 2018

AndreSteenbergen commented Jul 17, 2018

AndreSteenbergen commented Jul 17, 2018

AndreSteenbergen commented Jul 17, 2018

Aaronontheweb commented Jul 17, 2018

marcpiechura Jul 17, 2018

marcpiechura Jul 17, 2018

AndreSteenbergen Jul 18, 2018

marcpiechura Jul 17, 2018

marcpiechura Jul 17, 2018

vasily-kirichenko Jul 18, 2018

marcpiechura Jul 17, 2018

marcpiechura Jul 17, 2018

AndreSteenbergen Jul 18, 2018

marcpiechura Jul 17, 2018

marcpiechura Jul 17, 2018

marcpiechura Jul 17, 2018

marcpiechura Jul 17, 2018

marcpiechura commented Jul 17, 2018

vasily-kirichenko Jul 18, 2018

vasily-kirichenko Jul 18, 2018

AndreSteenbergen Jul 18, 2018

vasily-kirichenko Jul 18, 2018

AndreSteenbergen commented Jul 18, 2018

AndreSteenbergen commented Aug 13, 2018

Aaronontheweb commented Aug 24, 2018

vasily-kirichenko commented Aug 25, 2018

AndreSteenbergen commented Sep 12, 2018

Horusiath commented Sep 25, 2018

Danthar commented Jan 8, 2019

Aaronontheweb commented Jan 9, 2019

Added support for Ampq v1.0 #38

Added support for Ampq v1.0 #38

Conversation

AndreSteenbergen commented Jul 17, 2018

Aaronontheweb commented Jul 17, 2018

AndreSteenbergen commented Jul 17, 2018

AndreSteenbergen commented Jul 17, 2018

AndreSteenbergen commented Jul 17, 2018

Aaronontheweb commented Jul 17, 2018

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

marcpiechura commented Jul 17, 2018

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

AndreSteenbergen commented Jul 18, 2018

AndreSteenbergen commented Aug 13, 2018

Aaronontheweb commented Aug 24, 2018

vasily-kirichenko commented Aug 25, 2018

AndreSteenbergen commented Sep 12, 2018

Horusiath commented Sep 25, 2018

Danthar commented Jan 8, 2019

Aaronontheweb commented Jan 9, 2019