Implement the missing part of amqp 1.0 (#63)

Message Header
AMQP value
Currently the fields are in readonly mode.

Author: Gsantomaggio

RabbitMQ.Stream.Client/AMQP/DescribedFormatCode.cs

@@ -22,10 +22,12 @@ public static int Write(Span<byte> span, byte data)
             offset += WireFormatting.WriteByte(span.Slice(offset), data);
             return offset;
         }
-        
+
         public const byte ApplicationData = 0x75;
         public const byte MessageAnnotations = 0x72;
         public const byte MessageProperties = 0x73;
         public const byte ApplicationProperties = 0x74;
+        public const byte MessageHeader = 0x70;
+        public const byte AmqpValue = 0x77;
     }
 }

RabbitMQ.Stream.Client/AMQP/Header.cs

@@ -0,0 +1,57 @@
+using System;
+using System.Buffers;
+
+namespace RabbitMQ.Stream.Client.AMQP
+{
+    public class Header
+    {
+        public bool Durable { get; internal set; }
+        public byte Priority { get; internal set; }
+        public uint Ttl { get; internal set; } // from milliseconds
+        public bool FirstAcquirer { get; internal set; }
+        public uint DeliveryCount { get; internal set; }
+
+
+        public static Header Parse(ReadOnlySequence<byte> amqpData, ref int byteRead)
+        {
+            var offset = AmqpWireFormatting.ReadCompositeHeader(amqpData, out var fields, out var next);
+            //TODO WIRE check the next
+            var h = new Header();
+            for (var index = 0; index < fields; index++)
+            {
+                offset += AmqpWireFormatting.TryReadNull(amqpData.Slice(offset), out var value);
+
+                if (!value)
+                {
+                    switch (index)
+                    {
+                        case 0:
+                            offset += AmqpWireFormatting.ReadAny(amqpData.Slice(offset), out var durable);
+                            h.Durable = (bool) durable;
+                            break;
+                        case 1:
+                            offset += AmqpWireFormatting.ReadUByte(amqpData.Slice(offset), out var priority);
+                            h.Priority = priority;
+                            break;
+                        case 2:
+                            offset += AmqpWireFormatting.ReadAny(amqpData.Slice(offset), out var ttl);
+                            h.Ttl = (uint) ttl;
+                            break;
+                        case 3:
+                            offset += AmqpWireFormatting.ReadBool(amqpData.Slice(offset), out var firstAcquirer);
+                            h.FirstAcquirer = firstAcquirer;
+                            break;
+                        case 4:
+                            offset += AmqpWireFormatting.ReadUint32(amqpData.Slice(offset), out var deliveryCount);
+                            h.DeliveryCount = deliveryCount;
+                            break;
+                    }
+                }
+            }
+
+
+            byteRead += offset;
+            return h;
+        }
+    }
+}

RabbitMQ.Stream.Client/Message.cs

@@ -26,8 +26,16 @@ public Message(Data data)
         public ApplicationProperties ApplicationProperties { get; set; }
 
         public Properties Properties { get; set; }
+
         public Data Data { get; }
 
+        // MessageHeader and AmqpValue are only in get.
+        // Just to have the compatibility with AMQP 1.0
+        // In this specific case it is not needed
+        public Header MessageHeader { get; internal set; }
+        public object AmqpValue { get; internal set; }
+
+
         public int Size => Data.Size +
                            (Properties?.Size ?? 0) +
                            (Annotations?.Size ?? 0) +
@@ -86,8 +94,10 @@ public static Message From(ReadOnlySequence<byte> amqpData)
             //parse AMQP encoded data
             var offset = 0;
             Annotations annotations = null;
+            Header header = null;
             Data data = default;
             Properties properties = null;
+            object amqpValue = null;
             ApplicationProperties applicationProperties = null;
             while (offset != amqpData.Length)
             {
@@ -110,6 +120,13 @@ public static Message From(ReadOnlySequence<byte> amqpData)
                         applicationProperties =
                             ApplicationProperties.Parse<ApplicationProperties>(amqpData.Slice(offset), ref offset);
                         break;
+                    case DescribedFormatCode.MessageHeader:
+                        header = Header.Parse(amqpData.Slice(offset), ref offset);
+                        break;
+                    case DescribedFormatCode.AmqpValue:
+                        offset += DescribedFormatCode.Size;
+                        offset += AmqpWireFormatting.ReadAny(amqpData.Slice(offset), out amqpValue);
+                        break;
                     default:
                         Console.WriteLine($"dataCode: {dataCode} not handled. Please open an issue.");
                         throw new ArgumentOutOfRangeException($"dataCode: {dataCode} not handled");
@@ -120,7 +137,9 @@ public static Message From(ReadOnlySequence<byte> amqpData)
             {
                 Annotations = annotations,
                 Properties = properties,
-                ApplicationProperties = applicationProperties
+                ApplicationProperties = applicationProperties,
+                AmqpValue = amqpValue,
+                MessageHeader = header
             };
             return msg;
         }

Tests/Amqp10Tests.cs

@@ -100,12 +100,12 @@ public void ReadsThrowsExceptionInvalidType()
             {
                 AmqpWireFormatting.ReadSByte(new ReadOnlySequence<byte>(data), out var value);
             });
-            
+
             Assert.Throws<AmqpParseException>(() =>
             {
                 AmqpWireFormatting.ReadBool(new ReadOnlySequence<byte>(data), out var value);
             });
-            
+
             Assert.Throws<AmqpParseException>(() =>
             {
                 AmqpWireFormatting.ReadInt16(new ReadOnlySequence<byte>(data), out var value);
@@ -189,13 +189,12 @@ public void ValidateFormatCode()
             var longValueBin64 = new byte[] {0x81, 0xff, 0xff, 0xff, 0xe6, 0x21, 0x42, 0xfe, 0x39};
             RunValidateFormatCode(longValue64, longValueBin64);
 
-            
-            const long longValue8 = 127; 
+
+            const long longValue8 = 127;
             var longValueBin8 = new byte[] {0x55, 0x7F};
             RunValidateFormatCode(longValue8, longValueBin8);
-            
-            
-            
+
+
             const float floatValue = -88.88f;
             var floatValueBin = new byte[] {0x72, 0xc2, 0xb1, 0xc2, 0x8f};
             RunValidateFormatCode(floatValue, floatValueBin);
@@ -374,6 +373,19 @@ public void ValidateMessagesFromGo()
             Assert.Equal("test", msgStaticTest.Annotations["test"]);
             Assert.Equal((long) 1, msgStaticTest.Annotations[(long) 1]);
             Assert.Equal((long) 100_000, msgStaticTest.Annotations[(long) 100_000]);
+
+
+            var header = SystemUtils.GetFileContent("header_amqpvalue_message");
+            var msgHeader = Message.From(new ReadOnlySequence<byte>(header));
+            Assert.NotNull(msgHeader);
+            Assert.NotNull(msgHeader.MessageHeader);
+            Assert.NotNull(msgHeader.AmqpValue);
+            Assert.True(msgHeader.MessageHeader.Durable);
+            Assert.True(msgHeader.MessageHeader.FirstAcquirer);
+            Assert.Equal(100, msgHeader.MessageHeader.Priority);
+            Assert.Equal((uint) 300, msgHeader.MessageHeader.DeliveryCount);
+            Assert.True(msgHeader.MessageHeader.Ttl == 0);
+            Assert.Equal("amqpValue", msgHeader.AmqpValue);
         }
 
         [Fact]
@@ -397,11 +409,6 @@ public void MapEntriesWithAnEmptyKeyShouldNotBeWrittenToTheWire()
 
             // we do not expect the new entry to be written
             Assert.Equal(expectedMapSize, actualMapSize);
-
-
         }
     }
-
-   
-
 }