Ticket #106 : Support AMQP, can publish message

habarthierry-hue · habarthierry-hue · commit 4b2c3fbd3559 · 2022-05-24T17:03:21.000+02:00
diff --git a/src/EventMesh/FaasNet.EventMesh.Client/EventMeshClient.cs b/src/EventMesh/FaasNet.EventMesh.Client/EventMeshClient.cs
@@ -139,6 +139,7 @@ internal static void EnsureSuccessStatus(Package packageRequest, Package package
     public interface IEventMeshClientPubSession
     {
         Task Publish(string topicName, object obj, CancellationToken cancellationToken = default(CancellationToken));
+        Task Publish(string topicName, string str, CancellationToken cancellationToken = default(CancellationToken));
         Task Publish(string topicName, CloudEvent cloudEvent, CancellationToken cancellationToken = default(CancellationToken));
     }
 
@@ -177,6 +178,11 @@ public EventMeshClientPubSession(HelloResponse session, IPAddress ipAddr, int po
         }
 
         public Task Publish(string topicName, object obj, CancellationToken cancellationToken = default(CancellationToken))
+        {
+            return Publish(topicName, JsonSerializer.Serialize(obj), cancellationToken);
+        }
+
+        public Task Publish(string topicName, string str, CancellationToken cancellationToken = default(CancellationToken))
         {
             var cloudEvt = new CloudEvent
             {
@@ -185,7 +191,7 @@ public EventMeshClientPubSession(HelloResponse session, IPAddress ipAddr, int po
                 Source = new Uri("http://localhost"),
                 Type = topicName,
                 DataContentType = "application/json",
-                Data = JsonSerializer.Serialize(obj),
+                Data = str,
                 Time = DateTimeOffset.UtcNow
             };
             return Publish(topicName, cloudEvt, cancellationToken);
diff --git a/src/EventMesh/FaasNet.EventMesh.Protocols.AMQP/AMQPProxy.cs b/src/EventMesh/FaasNet.EventMesh.Protocols.AMQP/AMQPProxy.cs
@@ -3,6 +3,7 @@
 using Amqp.Types;
 using FaasNet.EventMesh.Protocols.AMQP.Framing;
 using FaasNet.EventMesh.Protocols.AMQP.Handlers;
+using Microsoft.Extensions.Logging;
 using Microsoft.Extensions.Options;
 using System;
 using System.Collections.Generic;
@@ -19,11 +20,13 @@ public class AMQPProxy : BaseProxy
         private readonly EventMeshAMQPOptions _options;
         private readonly IEnumerable<IRequestHandler> _requestHandlers;
         private static ManualResetEvent _lock = new ManualResetEvent(false);
+        private readonly ILogger<AMQPProxy> _logger;
 
-        public AMQPProxy(IOptions<EventMeshAMQPOptions> options, IEnumerable<IRequestHandler> requestHandlers)
+        public AMQPProxy(IOptions<EventMeshAMQPOptions> options, IEnumerable<IRequestHandler> requestHandlers, ILogger<AMQPProxy> logger)
         {
             _options = options.Value;
             _requestHandlers = requestHandlers;
+            _logger = logger;
         }
 
         protected override void Init()
@@ -112,9 +115,10 @@ private async Task ReplyFrame(StateObject stateObject, byte[] buffer)
             byte[] payload = null;
             Frame.Decode(frameBuffer, out channel, out command);
             if (frameBuffer.Length > 0) payload = buffer.Skip(frameBuffer.Offset + FixedWidth.UInt).Take(frameBuffer.Length - FixedWidth.UInt).ToArray();
-            Console.WriteLine($"Receive {command.Descriptor.Name}");
+            _logger.LogInformation("Receive {command}", command.Descriptor.Name);
             var requestHandler = _requestHandlers.First(r => r.RequestName == command.Descriptor.Name);
-            var result = await requestHandler.Handle(stateObject, command, payload, channel, TokenSource.Token);
+            var parameter = new RequestParameter(command, payload, channel);
+            var result = await requestHandler.Handle(stateObject, parameter, TokenSource.Token);
             if (result == null)
             {
                 var newState = new StateObject { WorkSocket = stateObject.WorkSocket, Session = stateObject.Session };
diff --git a/src/EventMesh/FaasNet.EventMesh.Protocols.AMQP/EventMeshAMQPOptions.cs b/src/EventMesh/FaasNet.EventMesh.Protocols.AMQP/EventMeshAMQPOptions.cs
@@ -5,12 +5,22 @@ public class EventMeshAMQPOptions
         public EventMeshAMQPOptions()
         {
             Port = 5672;
+            MaxFrameSize = 1000;
             EventMeshVpn = Client.Constants.DefaultVpn;
             EventMeshUrl = Client.Constants.DefaultUrl;
             EventMeshPort = Client.Constants.DefaultPort;
         }
 
         public int Port { get; set; }
+        /// <summary>
+        /// Largest frame size that the sending peer is able to accept on this connection.
+        /// </summary>
+        public uint MaxFrameSize { get; set; }
+        /// <summary>
+        /// The channel-max value is the highest channel number that can be used on the connection.
+        /// This value plus one is the maximum number of sessions that can simultaneously active on the connection.
+        /// </summary>
+        public ushort MaxChannel { get; set; }
         public string EventMeshVpn { get; set; }
         public string EventMeshUrl { get; set; }
         public int EventMeshPort { get; set; }
diff --git a/src/EventMesh/FaasNet.EventMesh.Protocols.AMQP/FaasNet.EventMesh.Protocols.AMQP.csproj b/src/EventMesh/FaasNet.EventMesh.Protocols.AMQP/FaasNet.EventMesh.Protocols.AMQP.csproj
@@ -5,6 +5,7 @@
   <ItemGroup>
 	  <PackageReference Include="AMQPNetLite" Version="2.4.4" />
 	  <PackageReference Include="Microsoft.Extensions.Options" Version="6.0.0" />
+	  <PackageReference Include="Microsoft.Extensions.Logging" Version="6.0.0" />
   </ItemGroup>
   <ItemGroup>
     <ProjectReference Include="..\FaasNet.EventMesh.Protocols\FaasNet.EventMesh.Protocols.csproj" />
diff --git a/src/EventMesh/FaasNet.EventMesh.Protocols.AMQP/Handlers/AttachHandler.cs b/src/EventMesh/FaasNet.EventMesh.Protocols.AMQP/Handlers/AttachHandler.cs
@@ -1,6 +1,5 @@
 ﻿using Amqp;
 using Amqp.Framing;
-using Amqp.Types;
 using FaasNet.EventMesh.Client;
 using FaasNet.EventMesh.Protocols.AMQP.Framing;
 using Microsoft.Extensions.Options;
@@ -21,15 +20,16 @@ public AttachHandler(IOptions<EventMeshAMQPOptions> options)
 
         public string RequestName => "amqp:attach:list";
 
-        public async Task<IEnumerable<ByteBuffer>> Handle(StateObject state, DescribedList cmd, byte[] payload, ushort channel, CancellationToken cancellationToken)
+        public async Task<IEnumerable<ByteBuffer>> Handle(StateObject state, RequestParameter parameter, CancellationToken cancellationToken)
         {
-            var attachCmd = cmd as Attach;
+            var attachCmd = parameter.Cmd as Attach;
             var target = attachCmd.Target as Target;
-            if(target != null && !string.IsNullOrWhiteSpace(target.Address)) state.Session.EventMeshPubSession = await CreatePubSession(state.Session, cancellationToken);
+            state.Session.Link = attachCmd;
+            if (target != null && !string.IsNullOrWhiteSpace(target.Address)) state.Session.EventMeshPubSession = await CreatePubSession(state.Session, cancellationToken);
             return new ByteBuffer[]
             {
-                BuildAttachResponse(channel, attachCmd),
-                BuildFrameResponse(channel, attachCmd)
+                BuildAttachResponse(parameter.Channel, attachCmd),
+                BuildFrameResponse(parameter.Channel, attachCmd)
             };
         }
 
diff --git a/src/EventMesh/FaasNet.EventMesh.Protocols.AMQP/Handlers/BeginHandler.cs b/src/EventMesh/FaasNet.EventMesh.Protocols.AMQP/Handlers/BeginHandler.cs
@@ -1,6 +1,5 @@
 ﻿using Amqp;
 using Amqp.Framing;
-using Amqp.Types;
 using FaasNet.EventMesh.Protocols.AMQP.Framing;
 using System.Collections.Generic;
 using System.Threading;
@@ -12,11 +11,10 @@ public class BeginHandler : IRequestHandler
     {
         public string RequestName => "amqp:begin:list";
 
-        public Task<IEnumerable<ByteBuffer>> Handle(StateObject state, DescribedList cmd, byte[] payload, ushort channel, CancellationToken cancellationToken)
+        public Task<IEnumerable<ByteBuffer>> Handle(StateObject state, RequestParameter parameter, CancellationToken cancellationToken)
         {
-            var requestBegin = cmd as Begin;
-            var beginResult = new Frame { Channel = channel, Type = FrameTypes.Amqp };
-            var begin = new Begin { RemoteChannel = channel };
+            var beginResult = new Frame { Channel = parameter.Channel, Type = FrameTypes.Amqp };
+            var begin = new Begin { RemoteChannel = parameter.Channel };
             IEnumerable<ByteBuffer> result = new[]
             {
                 beginResult.Serialize(begin)
diff --git a/src/EventMesh/FaasNet.EventMesh.Protocols.AMQP/Handlers/IRequestHandler.cs b/src/EventMesh/FaasNet.EventMesh.Protocols.AMQP/Handlers/IRequestHandler.cs
@@ -1,5 +1,4 @@
 ﻿using Amqp;
-using Amqp.Types;
 using System.Collections.Generic;
 using System.Threading;
 using System.Threading.Tasks;
@@ -9,6 +8,6 @@ namespace FaasNet.EventMesh.Protocols.AMQP.Handlers
     public interface IRequestHandler
     {
         string RequestName { get; }
-        Task<IEnumerable<ByteBuffer>> Handle(StateObject state, DescribedList cmd, byte[] payload, ushort channel, CancellationToken cancellationToken);
+        Task<IEnumerable<ByteBuffer>> Handle(StateObject state, RequestParameter parameter, CancellationToken cancellationToken);
     }
 }
diff --git a/src/EventMesh/FaasNet.EventMesh.Protocols.AMQP/Handlers/OpenHandler.cs b/src/EventMesh/FaasNet.EventMesh.Protocols.AMQP/Handlers/OpenHandler.cs
@@ -1,6 +1,6 @@
 ﻿using Amqp;
-using Amqp.Types;
 using FaasNet.EventMesh.Protocols.AMQP.Framing;
+using Microsoft.Extensions.Options;
 using System.Collections.Generic;
 using System.Threading;
 using System.Threading.Tasks;
@@ -9,14 +9,23 @@ namespace FaasNet.EventMesh.Protocols.AMQP.Handlers
 {
     public class OpenHandler : IRequestHandler
     {
+        private readonly EventMeshAMQPOptions _options;
+
+        public OpenHandler(IOptions<EventMeshAMQPOptions> options)
+        {
+            _options = options.Value;
+        }
+
         public string RequestName => "amqp:open:list";
 
-        public Task<IEnumerable<ByteBuffer>> Handle(StateObject state, DescribedList cmd, byte[] payload, ushort channel, CancellationToken cancellationToken)
+        public Task<IEnumerable<ByteBuffer>> Handle(StateObject state, RequestParameter parameter, CancellationToken cancellationToken)
         {
-            var openResult = new Frame { Channel = channel, Type = FrameTypes.Amqp };
+            var openResult = new Frame { Channel = parameter.Channel, Type = FrameTypes.Amqp };
             var cmdResult = new Amqp.Framing.Open
             {
-                HostName = "localhost"
+                HostName = "localhost",
+                MaxFrameSize = _options.MaxFrameSize,
+                ChannelMax = _options.MaxChannel
             };
             IEnumerable<ByteBuffer> result = new[]
             {
diff --git a/src/EventMesh/FaasNet.EventMesh.Protocols.AMQP/Handlers/RequestParameter.cs b/src/EventMesh/FaasNet.EventMesh.Protocols.AMQP/Handlers/RequestParameter.cs
@@ -0,0 +1,18 @@
+﻿using Amqp.Types;
+
+namespace FaasNet.EventMesh.Protocols.AMQP.Handlers
+{
+    public class RequestParameter
+    {
+        public RequestParameter(DescribedList cmd, byte[] payload, ushort channel)
+        {
+            Cmd = cmd;
+            Payload = payload;
+            Channel = channel;
+        }
+
+        public DescribedList Cmd { get; set; }
+        public byte[] Payload { get; set; }
+        public ushort Channel { get; set; }
+    }
+}
diff --git a/src/EventMesh/FaasNet.EventMesh.Protocols.AMQP/Handlers/SASLInitHandler.cs b/src/EventMesh/FaasNet.EventMesh.Protocols.AMQP/Handlers/SASLInitHandler.cs
@@ -14,9 +14,9 @@ public class SASLInitHandler : IRequestHandler
     {
         public string RequestName => "amqp:sasl-init:list";
 
-        public Task<IEnumerable<ByteBuffer>> Handle(StateObject state, DescribedList cmd, byte[] payload, ushort channel, CancellationToken cancellationToken)
+        public Task<IEnumerable<ByteBuffer>> Handle(StateObject state, RequestParameter parameter, CancellationToken cancellationToken)
         {
-            var saslInit = cmd as SaslInit;
+            var saslInit = parameter.Cmd as SaslInit;
             if (saslInit.Mechanism.ToString() != "PLAIN") throw new InvalidOperationException("Only PLAIN authentication mechanisme is supported");
             var saslInitResult = new Frame { Channel = 0, Type = FrameTypes.Sasl };
             var salsOutcome = new SaslOutcome { Code = SaslCode.Auth };
diff --git a/src/EventMesh/FaasNet.EventMesh.Protocols.AMQP/Handlers/TransferHandler.cs b/src/EventMesh/FaasNet.EventMesh.Protocols.AMQP/Handlers/TransferHandler.cs
@@ -1,43 +1,45 @@
 ﻿using Amqp;
 using Amqp.Framing;
-using Amqp.Types;
 using FaasNet.EventMesh.Protocols.AMQP.Framing;
-using Microsoft.Extensions.Options;
 using System.Collections.Generic;
+using System.Text;
 using System.Threading;
 using System.Threading.Tasks;
 
 namespace FaasNet.EventMesh.Protocols.AMQP.Handlers
 {
     public class TransferHandler : IRequestHandler
     {
-        private readonly EventMeshAMQPOptions _options;
-
-        public TransferHandler(IOptions<EventMeshAMQPOptions> options)
-        {
-            _options = options.Value;
-        }
-
         public string RequestName => "amqp:transfer:list";
 
-        public Task<IEnumerable<ByteBuffer>> Handle(StateObject state, DescribedList cmd, byte[] payload, ushort channel, CancellationToken cancellationToken)
+        public async Task<IEnumerable<ByteBuffer>> Handle(StateObject state, RequestParameter parameter, CancellationToken cancellationToken)
         {
-            var transfer = cmd as Transfer;
-            var disposeResult = new Frame { Channel = channel, Type = FrameTypes.Amqp };
-            var dispose = new Dispose
-            {
-                State = transfer.State
-            };
-            IEnumerable<ByteBuffer> result = new List<ByteBuffer>
+            await PublishMessage(state, parameter, cancellationToken);
+            return new List<ByteBuffer>
             {
-                disposeResult.Serialize(dispose)
+                BuildDispose(state, parameter)
             };
-            return Task.FromResult(result);
         }
 
-        private async Task PublishMessage(StateObject state, byte[] payload)
+        private async Task PublishMessage(StateObject state, RequestParameter parameter, CancellationToken cancellationToken)
         {
-            // state.Session.EventMeshPubSession.Publish()
+            var target = state.Session.Link.Target as Target;
+            await state.Session.EventMeshPubSession.Publish(target.Address, Encoding.UTF8.GetString(parameter.Payload), cancellationToken);
+        }
+
+        private ByteBuffer BuildDispose(StateObject state, RequestParameter parameter)
+        {
+            var transfer = parameter.Cmd as Transfer;
+            var disposeResult = new Frame { Channel = parameter.Channel, Type = FrameTypes.Amqp };
+            var dispose = new Dispose 
+            { 
+                State = transfer.State, 
+                Role = true,
+                First = transfer.DeliveryId,
+                Last = transfer.DeliveryId,
+                Settled = true
+            };
+            return disposeResult.Serialize(dispose);
         }
     }
 }
diff --git a/src/EventMesh/FaasNet.EventMesh.Protocols.AMQP/ServiceCollectionExtensions.cs b/src/EventMesh/FaasNet.EventMesh.Protocols.AMQP/ServiceCollectionExtensions.cs
@@ -11,6 +11,7 @@ public static IServiceCollection AddAMQPProtocol(this IServiceCollection service
         {
             if (callback == null) services.Configure<EventMeshAMQPOptions>((o) => { });
             services.Configure(callback);
+            services.AddLogging();
             services.AddTransient<IProxy, AMQPProxy>();
             services.AddTransient<IRequestHandler, OpenHandler>();
             services.AddTransient<IRequestHandler, SASLInitHandler>();
diff --git a/src/EventMesh/FaasNet.EventMesh.Protocols.AMQP/StateSessionObject.cs b/src/EventMesh/FaasNet.EventMesh.Protocols.AMQP/StateSessionObject.cs
@@ -1,4 +1,5 @@
-﻿using FaasNet.EventMesh.Client;
+﻿using Amqp.Framing;
+using FaasNet.EventMesh.Client;
 
 namespace FaasNet.EventMesh.Protocols.AMQP
 {
@@ -7,7 +8,7 @@ public class StateSessionObject
     {
         public string ClientId { get; set; }
         public string Password { get; set; }
-
         public IEventMeshClientPubSession EventMeshPubSession { get; set; }
+        public Attach Link { get; set; }
     }
 }

Original file line number	Diff line number	Diff line change
`@@ -139,6 +139,7 @@ internal static void EnsureSuccessStatus(Package packageRequest, Package package`
`139`	`139`	`public interface IEventMeshClientPubSession`
`140`	`140`	`{`
`141`	`141`	`Task Publish(string topicName, object obj, CancellationToken cancellationToken = default(CancellationToken));`
	`142`	`+ Task Publish(string topicName, string str, CancellationToken cancellationToken = default(CancellationToken));`
`142`	`143`	`Task Publish(string topicName, CloudEvent cloudEvent, CancellationToken cancellationToken = default(CancellationToken));`
`143`	`144`	`}`
`144`	`145`
`@@ -177,6 +178,11 @@ public EventMeshClientPubSession(HelloResponse session, IPAddress ipAddr, int po`
`177`	`178`	`}`
`178`	`179`
`179`	`180`	`public Task Publish(string topicName, object obj, CancellationToken cancellationToken = default(CancellationToken))`
	`181`	`+ {`
	`182`	`+ return Publish(topicName, JsonSerializer.Serialize(obj), cancellationToken);`
	`183`	`+ }`
	`184`	`+`
	`185`	`+ public Task Publish(string topicName, string str, CancellationToken cancellationToken = default(CancellationToken))`
`180`	`186`	`{`
`181`	`187`	`var cloudEvt = new CloudEvent`
`182`	`188`	`{`
`@@ -185,7 +191,7 @@ public EventMeshClientPubSession(HelloResponse session, IPAddress ipAddr, int po`
`185`	`191`	`Source = new Uri("http://localhost"),`
`186`	`192`	`Type = topicName,`
`187`	`193`	`DataContentType = "application/json",`
`188`		`- Data = JsonSerializer.Serialize(obj),`
	`194`	`+ Data = str,`
`189`	`195`	`Time = DateTimeOffset.UtcNow`
`190`	`196`	`};`
`191`	`197`	`return Publish(topicName, cloudEvt, cancellationToken);`
Original file line number	Diff line number	Diff line change
`@@ -1,5 +1,4 @@`
`1`	`1`	`using Amqp;`
`2`		`-using Amqp.Types;`
`3`	`2`	`using System.Collections.Generic;`
`4`	`3`	`using System.Threading;`
`5`	`4`	`using System.Threading.Tasks;`
`@@ -9,6 +8,6 @@ namespace FaasNet.EventMesh.Protocols.AMQP.Handlers`
`9`	`8`	`public interface IRequestHandler`
`10`	`9`	`{`
`11`	`10`	`string RequestName { get; }`
`12`		`- Task<IEnumerable<ByteBuffer>> Handle(StateObject state, DescribedList cmd, byte[] payload, ushort channel, CancellationToken cancellationToken);`
	`11`	`+ Task<IEnumerable<ByteBuffer>> Handle(StateObject state, RequestParameter parameter, CancellationToken cancellationToken);`
`13`	`12`	`}`
`14`	`13`	`}`
Original file line number	Diff line number	Diff line change
`@@ -14,9 +14,9 @@ public class SASLInitHandler : IRequestHandler`
`14`	`14`	`{`
`15`	`15`	`public string RequestName => "amqp:sasl-init:list";`
`16`	`16`
`17`		`- public Task<IEnumerable<ByteBuffer>> Handle(StateObject state, DescribedList cmd, byte[] payload, ushort channel, CancellationToken cancellationToken)`
	`17`	`+ public Task<IEnumerable<ByteBuffer>> Handle(StateObject state, RequestParameter parameter, CancellationToken cancellationToken)`
`18`	`18`	`{`
`19`		`- var saslInit = cmd as SaslInit;`
	`19`	`+ var saslInit = parameter.Cmd as SaslInit;`
`20`	`20`	`if (saslInit.Mechanism.ToString() != "PLAIN") throw new InvalidOperationException("Only PLAIN authentication mechanisme is supported");`
`21`	`21`	`var saslInitResult = new Frame { Channel = 0, Type = FrameTypes.Sasl };`
`22`	`22`	`var salsOutcome = new SaslOutcome { Code = SaslCode.Auth };`