Send Message

src/ArtemisNetCoreClient/ByteBuffer.cs

@@ -324,4 +324,9 @@ public void WriteSize()
         _memoryStream.TryGetBuffer(out var buffer);
         BinaryPrimitives.WriteInt32BigEndian(buffer, buffer.Count - sizeof(int));
     }
+
+    public long ReadableBytes()
+    {
+        return _memoryStream.Length - _memoryStream.Position;
+    }
 }

src/ArtemisNetCoreClient/Framing/Message.cs

@@ -21,6 +21,8 @@ public class Message
 
     internal void Encode(ByteBuffer buffer)
     {
+        // TODO: Add comment why it's 17 ;-)
+        buffer.WriteInt(Body.Length + 17);
         buffer.WriteInt(Body.Length);
         buffer.WriteBytes(Body);
 

src/ArtemisNetCoreClient/Framing/NullResponse.cs

@@ -10,5 +10,13 @@ public override void Encode(ByteBuffer buffer)
 
     public override void Decode(ByteBuffer buffer)
     {
+        if (buffer.ReadableBytes() >= sizeof(long))
+        {
+            CorrelationId = buffer.ReadLong();
+        }
+        else
+        {
+            CorrelationId = -1;
+        }
     }
 }

src/ArtemisNetCoreClient/Framing/Packet.cs

@@ -2,7 +2,7 @@ namespace ActiveMQ.Artemis.Core.Client.Framing;
 
 internal abstract class Packet
 {
-    public long CorrelationId { get; init; } = -1;
+    public long CorrelationId { get; set; } = -1;
     public virtual bool IsResponse => false;
 
     public abstract void Encode(ByteBuffer buffer);

src/ArtemisNetCoreClient/IProducer.cs

@@ -27,7 +27,7 @@ await session.SendBlockingAsync<SessionSendMessageV3, NullResponse>(new SessionS
             Message = message,
             ProducerId = ProducerId,
             RequiresResponse = true,
-            CorrelationId = 1000
+            CorrelationId = -4
         }, cancellationToken);
     }
 }

test/ArtemisNetCoreClient.Tests/ProducerTests.cs

@@ -4,7 +4,7 @@ namespace ActiveMQ.Artemis.Core.Client.Tests;
 
 public class ProducerTests
 {
-    [Test, Ignore("Failing")]
+    [Test]
     public async Task should_send_message()
     {
         // Arrange