Query Address Info

src/ArtemisNetCoreClient/AddressInfo.cs

@@ -0,0 +1,10 @@
+using ActiveMQ.Artemis.Core.Client.Framing;
+
+namespace ActiveMQ.Artemis.Core.Client;
+
+public class AddressInfo
+{
+    public string Name { get; init; }
+    public IReadOnlyList<string> QueueNames { get; init; }
+    public IReadOnlyList<RoutingType> RoutingTypes { get; init; }
+}

src/ArtemisNetCoreClient/ByteBuffer.cs

@@ -17,7 +17,7 @@ public ByteBuffer(byte[] payload)
     {
         _memoryStream = new MemoryStream(payload, writable: false);
     }
-
+    
     public ReadOnlyMemory<byte> GetBuffer()
     {
         _memoryStream.TryGetBuffer(out var buffer);
@@ -295,6 +295,12 @@ private string ReadAsShorts(int length)
         return value == DataConstants.NotNull ? ReadString() : null;
     }
 
+    public string? ReadNullableStringAsBytes()
+    {
+        var value = _memoryStream.ReadByte();
+        return value == DataConstants.NotNull ? ReadStringAsBytes() : null;
+    }
+
     public void WriteSize()
     {
         _memoryStream.TryGetBuffer(out var buffer);

src/ArtemisNetCoreClient/Framing/Codec.cs

@@ -33,6 +33,7 @@ public static (Packet packet, long channelId) Decode(ByteBuffer buffer)
         {
             CreateSessionResponseMessage.Type => new CreateSessionResponseMessage(),
             NullResponse.Type => new NullResponse(),
+            SessionBindingQueryResponseMessageV5.Type => new SessionBindingQueryResponseMessageV5(),
             _ => throw new ArgumentOutOfRangeException($"Type {type} is not supported for decoding")
         };
 

src/ArtemisNetCoreClient/Framing/SessionBindingQueryResponseMessageV5.cs

@@ -2,6 +2,10 @@ namespace ActiveMQ.Artemis.Core.Client.Framing;
 
 internal class SessionBindingQueryResponseMessageV5 : Packet
 {
+    public const byte Type = unchecked((byte) -22);
+
+    public override bool IsResponse => true;
+
     public bool Exists { get; set; }
     public IReadOnlyList<string> QueueNames { get; set; }
     public bool AutoCreateQueues { get; set; }
@@ -12,8 +16,8 @@ internal class SessionBindingQueryResponseMessageV5 : Packet
     public bool? DefaultLastValue { get; set; }
     public string? DefaultLastValueKey { get; set; }
     public bool? DefaultNonDestructive { get; set; }
-    public int DefaultConsumersBeforeDispatch { get; set; }
-    public long DefaultDelayBeforeDispatch { get; set; }
+    public int? DefaultConsumersBeforeDispatch { get; set; }
+    public long? DefaultDelayBeforeDispatch { get; set; }
     public bool SupportsMulticast { get; set; }
     public bool SupportsAnycast { get; set; }
 
@@ -26,11 +30,12 @@ public override void Decode(ByteBuffer buffer)
     {
         Exists = buffer.ReadBool();
         var numQueues = buffer.ReadInt();
-        var queueNames = new string [numQueues];
+        var queueNames = new string[numQueues];
         for (int i = 0; i < numQueues; i++)
         {
             queueNames[i] = buffer.ReadStringAsBytes();
         }
+
         QueueNames = queueNames;
         AutoCreateQueues = buffer.ReadBool();
         AutoCreateAddresses = buffer.ReadBool();
@@ -40,5 +45,9 @@ public override void Decode(ByteBuffer buffer)
         DefaultLastValue = buffer.ReadNullableBool();
         DefaultLastValueKey = buffer.ReadNullableStringAsBytes();
         DefaultNonDestructive = buffer.ReadNullableBool();
+        DefaultConsumersBeforeDispatch = buffer.ReadNullableInt();
+        DefaultDelayBeforeDispatch = buffer.ReadNullableLong();
+        SupportsMulticast = buffer.ReadBool();
+        SupportsAnycast = buffer.ReadBool();
     }
 }

src/ArtemisNetCoreClient/ISession.cs

@@ -4,5 +4,6 @@ namespace ActiveMQ.Artemis.Core.Client;
 
 public interface ISession : IAsyncDisposable
 {
-    public Task CreateAddress(string address, IEnumerable<RoutingType> routingTypes, bool autoCreated, CancellationToken cancellationToken);
+    Task CreateAddress(string address, IEnumerable<RoutingType> routingTypes, CancellationToken cancellationToken);
+    Task<AddressInfo> GetAddressInfo(string address, CancellationToken cancellationToken);
 }

src/ArtemisNetCoreClient/Session.cs

@@ -39,17 +39,46 @@ public Session(Transport transport)
         });
     }
 
-    public async Task CreateAddress(string address, IEnumerable<RoutingType> routingTypes, bool autoCreated, CancellationToken cancellationToken)
+    public async Task CreateAddress(string address, IEnumerable<RoutingType> routingTypes, CancellationToken cancellationToken)
     {
         var createAddressMessage = new CreateAddressMessage
         {
             Address = address,
             RoutingTypes = routingTypes.ToArray(),
-            AutoCreated = autoCreated,
+            AutoCreated = false,
             RequiresResponse = true
         };
         _ = await SendBlockingAsync<CreateAddressMessage, NullResponse>(createAddressMessage, cancellationToken);
     }
+
+    public async Task<AddressInfo> GetAddressInfo(string address, CancellationToken cancellationToken)
+    {
+        var request = new SessionBindingQueryMessage
+        {
+            Address = address
+        };
+        var response = await SendBlockingAsync<SessionBindingQueryMessage, SessionBindingQueryResponseMessageV5>(request, cancellationToken);
+
+        return new AddressInfo
+        {
+            Name = address,
+            QueueNames = response.QueueNames,
+            RoutingTypes = GetRoutingTypes(response).ToArray(),
+        };
+    }
+
+    private static IEnumerable<RoutingType> GetRoutingTypes(SessionBindingQueryResponseMessageV5 sessionBindingQueryResponseMessageV5)
+    {
+        if (sessionBindingQueryResponseMessageV5.SupportsAnycast)
+        {
+            yield return RoutingType.Anycast;
+        }
+
+        if (sessionBindingQueryResponseMessageV5.SupportsMulticast)
+        {
+            yield return RoutingType.Multicast;
+        }
+    }
 
     public async ValueTask DisposeAsync()
     {
@@ -62,7 +91,7 @@ private async Task<TResponse> SendBlockingAsync<TRequest, TResponse>(TRequest re
     {
         var tcs = new TaskCompletionSource<Packet>();
 
-        // TODO: Handle scenario when we cannot CorrelationId
+        // TODO: Handle scenario when we cannot add request for this CorrelationId, because there is already another pending request
         _ = _completionSources.TryAdd(request.CorrelationId, tcs);
 
         await _transport.SendAsync(request, ChannelId, cancellationToken);

test/ArtemisNetCoreClient.Tests/SessionTests.cs

@@ -41,25 +41,11 @@ public async Task should_create_address_with_selected_routing_type(RoutingType[]
 
         // Act
         var addressName = $"{Guid.NewGuid().ToString()}-{string.Join("-", routingTypes)}";
-        await session.CreateAddress(addressName, routingTypes, false, default);
-    }
-
-    [TestCase(false)]
-    [TestCase(true)]
-    public async Task should_create_address_with_autoCreated_flag(bool autoCreated)
-    {
-        // Arrange
-        var connectionFactory = new SessionFactory();
-        await using var session = await connectionFactory.CreateAsync(new Endpoint
-        {
-            Host = "localhost",
-            Port = 5445,
-            User = "artemis",
-            Password = "artemis"
-        });
-
-        // Act
-        var addressName = $"{Guid.NewGuid().ToString()}";
-        await session.CreateAddress(addressName, new[] { RoutingType.Multicast }, autoCreated: autoCreated, default);
+        await session.CreateAddress(addressName, routingTypes, default);
+
+        // Assert
+        var addressInfo = await session.GetAddressInfo(addressName, default);
+        Assert.That(addressInfo.Name, Is.EqualTo(addressName));
+        CollectionAssert.AreEqual(routingTypes, addressInfo.RoutingTypes);
     }
 }