Ticket #106 : Can accept AMQP connection

Author: habarthierry-hue

kubernetes/localrabbitmq.yml

@@ -1,3 +1,6 @@
+# Cluster operator must be installed : kubectl apply -f "https://github.com/rabbitmq/cluster-operator/releases/latest/download/cluster-operator.yml"
+# default user : hello-world-default-user
+# default password: hello-world-default-user
 apiVersion: v1
 kind: Service
 metadata:
@@ -14,72 +17,14 @@ spec:
     protocol: TCP
     nodePort: 30007
   selector:
-    app: rabbitmq
+    app.kubernetes.io/name: hello-world
   type: NodePort
 ---
-apiVersion: v1
-kind: Service
-metadata:
-  name: rabbitmq-entry
-spec:
-  type: ClusterIP
-  selector:
-    app: rabbitmq
-  ports:
-  - port: 5672
-    protocol: TCP
-    targetPort: 5672
----
-apiVersion: apps/v1
-kind: StatefulSet
+apiVersion: rabbitmq.com/v1beta1
+kind: RabbitmqCluster
 metadata:
-  name: rabbitmq
+    name: hello-world
 spec:
-  serviceName: "rabbitmq"
-  selector:
-    matchLabels:
-      app: rabbitmq
-  replicas: 1
-  template:
-    metadata:
-      labels:
-        app: rabbitmq
-    spec:
-      containers:
-      - name: rabbitmq
-        image: rabbitmq:3.6.6-management-alpine
-        lifecycle:
-          postStart:
-            exec:
-              command:
-              - /bin/sh
-              - -c
-              - >
-                if [ -z "$(grep rabbitmq /etc/resolv.conf)" ]; then
-                  sed "s/^search \([^ ]\+\)/search rabbitmq.\1 \1/" /etc/resolv.conf > /etc/resolv.conf.new;
-                  cat /etc/resolv.conf.new > /etc/resolv.conf;
-                  rm /etc/resolv.conf.new;
-                fi;
-                until rabbitmqctl node_health_check; do sleep 1; done;
-                if [[ "$HOSTNAME" != "rabbitmq-0" && -z "$(rabbitmqctl cluster_status | grep rabbitmq-0)" ]]; then
-                  rabbitmqctl stop_app;
-                  rabbitmqctl join_cluster rabbit@rabbitmq-0;
-                  rabbitmqctl start_app;
-                fi;
-                rabbitmqctl set_policy ha-all "." '{"ha-mode":"exactly","ha-params":3,"ha-sync-mode":"automatic"}'
-        ports:
-        - containerPort: 5672
-          name: amqp
-        volumeMounts:
-        - name: rabbitmq
-          mountPath: /var/lib/rabbitmq
-  volumeClaimTemplates:
-  - metadata:
-      name: rabbitmq
-      annotations:
-        volume.alpha.kubernetes.io/storage-class: anything
-    spec:
-      accessModes: [ "ReadWriteOnce" ]
-      resources:
-        requests:
-          storage: 1Gi
+  rabbitmq:
+    additionalPlugins:
+      - rabbitmq_amqp1_0

src/EventMesh/FaasNet.EventMesh.Protocols.AMQP/Framing/Frame.cs

@@ -1,5 +1,6 @@
 using Amqp;
 using Amqp.Framing;
+using Amqp.Types;
 using System;
 
 namespace FaasNet.EventMesh.Protocols.AMQP.Framing
@@ -29,31 +30,33 @@ public class Frame
         public FrameTypes Type { get; set; }
         public ushort Channel { get; set; }
 
-        public static void Decode(ByteBuffer buffer, out ushort channel)
+        public static void Decode(ByteBuffer buffer, out ushort channel, out DescribedList command)
         {
             AmqpBitConverter.ReadUInt(buffer);
             AmqpBitConverter.ReadUByte(buffer);
             AmqpBitConverter.ReadUByte(buffer);
             channel = AmqpBitConverter.ReadUShort(buffer);
+            if (buffer.Length > 0)
+            {
+                var tt = Encoder.ReadDescribed(buffer, Encoder.ReadFormatCode(buffer));
+                command = (DescribedList)tt;
+            }
+            else
+            {
+                command = null;
+            }
         }
 
-        public byte[] Serialize()
+        public ByteBuffer Serialize(DescribedList cmd)
         {
             var buffer = new ByteBuffer(CmdBufferSize, false);
             buffer.Append(FixedWidth.UInt);
             AmqpBitConverter.WriteUByte(buffer, DOFF);
             AmqpBitConverter.WriteUByte(buffer, (byte)Type);
-            switch(Type)
-            {
-                case FrameTypes.Amqp:
-                    AmqpBitConverter.WriteUShort(buffer, Channel);
-                    break;
-            }
-
-            var accepted = new Accepted();
-            accepted.Encode(buffer);
+            AmqpBitConverter.WriteUShort(buffer, Channel);
+            cmd.Encode(buffer);
             AmqpBitConverter.WriteInt(buffer.Buffer, buffer.Offset, buffer.Length);
-            return buffer.Buffer;
+            return buffer;
         }
     }
 }

src/EventMesh/FaasNet.EventMesh.Protocols.AMQP/Receiver.cs

@@ -1,4 +1,7 @@
-using FaasNet.EventMesh.Protocols.AMQP.Framing;
+using Amqp;
+using Amqp.Sasl;
+using Amqp.Types;
+using FaasNet.EventMesh.Protocols.AMQP.Framing;
 using System.Net;
 using System.Net.Sockets;
 
@@ -29,13 +32,35 @@ private void Launch()
 
         private void Listen()
         {
+            var saslHeaderNegotiation = new ProtocolHeader { Id = 3, Major = 1, Minor = 0, Revision = 0 };
             var handler = _server.Accept();
-            // Header.
-            var header = ReadHeader(handler);
-            handler.Send(header.Serialize());
-            // Frames
-            var frame = new Frame { Channel = 0, Type = FrameTypes.Amqp };
-            handler.Send(frame.Serialize());
+            // Check header & protocol.
+            ReadHeader(handler);
+            handler.Send(saslHeaderNegotiation.Serialize(), SocketFlags.None);
+            // Return supported authentication mechanisms.
+            var saslFrame = new Frame { Channel = 0, Type = FrameTypes.Sasl };
+            var cmd = new SaslMechanisms 
+            { 
+                SaslServerMechanisms = new Amqp.Types.Symbol[] 
+                {
+                    new Amqp.Types.Symbol("PLAIN")
+                }
+            };
+            Send(handler, saslFrame.Serialize(cmd));
+            // Authenticate the user.
+            ReadFrame(handler);
+            var sCmd = new SaslOutcome { Code = SaslCode.Ok };
+            Send(handler, saslFrame.Serialize(sCmd));
+            // Open
+            var saslSecuredConnection = new ProtocolHeader { Id = 0, Major = 1, Minor = 0, Revision = 0 };
+            handler.Send(saslSecuredConnection.Serialize(), SocketFlags.None);
+            var amqpFrame = new Frame { Channel = 0, Type = FrameTypes.Amqp };
+            var tCmd = new Amqp.Framing.Open
+            {
+                HostName = "localhost"
+            };
+            Send(handler, amqpFrame.Serialize(tCmd));
+            string s = "";
         }
 
         private ProtocolHeader ReadHeader(Socket socket)
@@ -45,5 +70,30 @@ private ProtocolHeader ReadHeader(Socket socket)
             var receivedHeader = ProtocolHeader.Create(headerBuffer, 0);
             return receivedHeader;
         }
+
+        private Sasl.SaslPlainProfile ReadSaslPlainProfile()
+        {
+            return null;
+        }
+
+        private static void Send(Socket socket, ByteBuffer buffer)
+        {
+            socket.Send(buffer.Buffer, buffer.Offset, buffer.Length, SocketFlags.None);
+        }
+
+        private static DescribedList ReadFrame(Socket socket)
+        {
+            var sizeFrameBuffer = new byte[FixedWidth.UInt];
+            socket.Receive(sizeFrameBuffer, 0, FixedWidth.UInt, SocketFlags.None);
+            int size = AmqpBitConverter.ReadInt(sizeFrameBuffer, 0);
+            var frameBuffer = new ByteBuffer(size, false);
+            AmqpBitConverter.WriteInt(frameBuffer, size);
+            socket.Receive(frameBuffer.Buffer, frameBuffer.Length, frameBuffer.Size, SocketFlags.None);
+            frameBuffer.Append(frameBuffer.Size);
+            ushort channel;
+            DescribedList command;
+            Frame.Decode(frameBuffer, out channel, out command);
+            return command;
+        }
     }
 }

src/EventMesh/FaasNet.EventMesh.Protocols.AMQP/Sasl/SaslPlainProfile.cs

@@ -0,0 +1,10 @@
+namespace FaasNet.EventMesh.Protocols.AMQP.Sasl
+{
+    public class SaslPlainProfile
+    {
+        public string User { get; set; }
+        public string Password { get; set; }
+
+
+    }
+}

src/EventMesh/FaasNet.EventMesh.Runtime.Console/Program.cs

@@ -10,11 +10,7 @@
 using FaasNet.RaftConsensus.Core.Stores;
 using Microsoft.Extensions.DependencyInjection;
 using RabbitMQ.Client;
-
-var frame = new Frame { Channel = 4, Type = 0 };
-var payload = frame.Serialize();
-var buffer = new ByteBuffer(payload, 0, payload.Count(), 0);
-Frame.Decode(buffer, out ushort channel);
+using ConnectionFactory = RabbitMQ.Client.ConnectionFactory;
 
 var amqpServer = new AMQPServer();
 amqpServer.Start();