Create Producer

src/ArtemisNetCoreClient/ConsumerConfiguration.cs

@@ -0,0 +1,6 @@
+namespace ActiveMQ.Artemis.Core.Client;
+
+public class ConsumerConfiguration
+{
+    public required string QueueName { get; init; }
+}

src/ArtemisNetCoreClient/Framing/Codec.cs

@@ -18,6 +18,8 @@ public static void Encode(ByteBuffer buffer, Packet packet, long channelId)
             SessionQueueQueryMessage => SessionQueueQueryMessage.Type,
             SessionCreateConsumerMessage => SessionCreateConsumerMessage.Type,
             SessionConsumerCloseMessage => SessionConsumerCloseMessage.Type,
+            CreateProducerMessage => CreateProducerMessage.Type,
+            RemoveProducerMessage => RemoveProducerMessage.Type,
             _ => throw new ArgumentOutOfRangeException(nameof(packet), packet, $"{packet.GetType()} is not supported for encoding")
         };
         buffer.WriteByte(type);

src/ArtemisNetCoreClient/Framing/CreateProducerMessage.cs

@@ -0,0 +1,19 @@
+namespace ActiveMQ.Artemis.Core.Client.Framing;
+
+internal class CreateProducerMessage : Packet
+{
+    public const byte Type = unchecked((byte) -20);
+
+    public required int Id { get; init; }
+    public string? Address { get; init; }
+
+    public override void Encode(ByteBuffer buffer)
+    {
+        buffer.WriteInt(Id);
+        buffer.WriteNullableByteString(Address);
+    }
+
+    public override void Decode(ByteBuffer buffer)
+    {
+    }
+}

src/ArtemisNetCoreClient/Framing/RemoveProducerMessage.cs

@@ -0,0 +1,17 @@
+namespace ActiveMQ.Artemis.Core.Client.Framing;
+
+internal class RemoveProducerMessage : Packet
+{
+    public const byte Type = unchecked((byte) -21);
+
+    public required int Id { get; init; }
+
+    public override void Encode(ByteBuffer buffer)
+    {
+        buffer.WriteInt(Id);
+    }
+
+    public override void Decode(ByteBuffer buffer)
+    {
+    }
+}

src/ArtemisNetCoreClient/IProducer.cs

@@ -0,0 +1,21 @@
+using ActiveMQ.Artemis.Core.Client.Framing;
+
+namespace ActiveMQ.Artemis.Core.Client;
+
+public interface IProducer : IAsyncDisposable
+{
+}
+
+internal class Producer(Session session) : IProducer
+{
+    public required int ProducerId { get; init; }
+
+    public async ValueTask DisposeAsync()
+    {
+        var request = new RemoveProducerMessage()
+        {
+            Id = ProducerId
+        };
+        await session.SendAsync(request, default);
+    }
+}

src/ArtemisNetCoreClient/ISession.cs

@@ -9,9 +9,5 @@ public interface ISession : IAsyncDisposable
     Task CreateQueue(QueueConfiguration queueConfiguration, CancellationToken cancellationToken);
     Task<QueueInfo?> GetQueueInfo(string queueName, CancellationToken cancellationToken);
     Task<IConsumer> CreateConsumerAsync(ConsumerConfiguration consumerConfiguration, CancellationToken cancellationToken);
-}
-
-public class ConsumerConfiguration
-{
-    public required string QueueName { get; init; }
+    Task<IProducer> CreateProducerAsync(ProducerConfiguration producerConfiguration, CancellationToken cancellationToken);
 }

src/ArtemisNetCoreClient/ProducerConfiguration.cs

@@ -0,0 +1,6 @@
+namespace ActiveMQ.Artemis.Core.Client;
+
+public class ProducerConfiguration
+{
+    public string? Address { get; init; }
+}

src/ArtemisNetCoreClient/Session.cs

@@ -136,6 +136,20 @@ public async Task<IConsumer> CreateConsumerAsync(ConsumerConfiguration consumerC
         };
     }
 
+    public async Task<IProducer> CreateProducerAsync(ProducerConfiguration producerConfiguration, CancellationToken cancellationToken)
+    {
+        var request = new CreateProducerMessage
+        {
+            Id = 0,
+            Address = producerConfiguration.Address
+        };
+        await _transport.SendAsync(request, ChannelId, cancellationToken);
+        return new Producer(this)
+        {
+            ProducerId = request.Id
+        };
+    }
+
     public async ValueTask DisposeAsync()
     {
         _ = await SendBlockingAsync<SessionStop, NullResponse>(new SessionStop(), default);
@@ -163,6 +177,11 @@ internal async Task<TResponse> SendBlockingAsync<TRequest, TResponse>(TRequest r
         }
     }
 
+    internal async Task SendAsync<TRequest>(TRequest request, CancellationToken cancellationToken) where TRequest : Packet
+    {
+        await _transport.SendAsync(request, ChannelId, cancellationToken);
+    }
+
     public long ChannelId { get; init; }
 
     public async Task StartAsync(CancellationToken cancellationToken)

test/ArtemisNetCoreClient.Tests/SessionTests.cs

@@ -45,7 +45,7 @@
 
         // Assert
         var addressInfo = await session.GetAddressInfo(addressName, default);
         Assert.That(addressInfo.Name, Is.EqualTo(addressName));
         CollectionAssert.AreEqual(routingTypes, addressInfo.RoutingTypes);
     }
 
@@ -62,11 +62,11 @@
             User = "artemis",
             Password = "artemis"
         });
-        var addressName = $"{Guid.NewGuid().ToString()}";
+        var addressName = Guid.NewGuid().ToString();
         await session.CreateAddress(addressName, [routingType], default);
 
         // Act
-        var queueName = $"{Guid.NewGuid().ToString()}";
+        var queueName = Guid.NewGuid().ToString();
         await session.CreateQueue(new QueueConfiguration
         {
             Address = addressName,
@@ -136,7 +136,7 @@
             User = "artemis",
             Password = "artemis"
         });
-        var addressName = $"{Guid.NewGuid().ToString()}";
+        var addressName = Guid.NewGuid().ToString();
         await session.CreateAddress(addressName, new [] { RoutingType.Multicast }, default);
 
         var queueName = Guid.NewGuid().ToString();
@@ -155,4 +155,28 @@
 
         await consumer.DisposeAsync();
     }
+
+    [Test]
+    public async Task should_create_and_dispose_producer()
+    {
+        // Arrange
+        var connectionFactory = new SessionFactory();
+        await using var session = await connectionFactory.CreateAsync(new Endpoint
+        {
+            Host = "localhost",
+            Port = 5445,
+            User = "artemis",
+            Password = "artemis"
+        });
+        var addressName = Guid.NewGuid().ToString();
+        await session.CreateAddress(addressName, new [] { RoutingType.Multicast }, default);
+
+        // Act
+        var producer = await session.CreateProducerAsync(new ProducerConfiguration()
+        {
+            Address = addressName
+        }, default);
+
+        await producer.DisposeAsync();
+    }
 }