Small fixes after address format v2 changes

Author: acogoluegnes

src/docs/asciidoc/usage.adoc

@@ -65,21 +65,21 @@ include::{test-examples}/Api.java[tag=message-publishing]
 --------
 include::{test-examples}/Api.java[tag=target-address-exchange-key]
 --------
-<1> Translates to `/exchange/foo/key/bar`
+<1> Translates to `/e/foo/bar`
 
 .Target address format: exchange
 [source,java,indent=0]
 --------
 include::{test-examples}/Api.java[tag=target-address-exchange]
 --------
-<1> Translates to `/exchange/foo`
+<1> Translates to `/e/foo`
 
 .Target address format: queue
 [source,java,indent=0]
 --------
 include::{test-examples}/Api.java[tag=target-address-queue]
 --------
-<1> Translates to `/queue/some-queue`
+<1> Translates to `/q/some-queue`
 
 .Target address format: address in `to` field
 [source,java,indent=0]

src/main/java/com/rabbitmq/client/amqp/impl/AmqpConsumer.java

@@ -61,7 +61,7 @@ final class AmqpConsumer extends ResourceBase implements Consumer {
   private final MetricsCollector metricsCollector;
   private final SessionHandler sessionHandler;
   private final AtomicLong unsettledMessageCount = new AtomicLong(0);
-  private final Runnable replenishCreditOperation = () -> replenishCreditIfNeeded();
+  private final Runnable replenishCreditOperation = this::replenishCreditIfNeeded;
   // native receiver internal state, accessed only in the native executor/scheduler
   private ProtonReceiver protonReceiver;
   private volatile Scheduler protonExecutor;
@@ -78,7 +78,9 @@ final class AmqpConsumer extends ResourceBase implements Consumer {
             .connection()
             .observationCollector()
             .subscribe(builder.queue(), builder.messageHandler());
-    this.address = "/queue/" + builder.queue();
+    DefaultAddressBuilder<?> addressBuilder = Utils.addressBuilder();
+    addressBuilder.queue(builder.queue());
+    this.address = addressBuilder.address();
     this.filters = Collections.unmodifiableMap(builder.filters());
     this.connection = builder.connection();
     this.sessionHandler = this.connection.createSessionHandler();

src/main/java/com/rabbitmq/client/amqp/impl/AmqpRpcClient.java

@@ -97,7 +97,7 @@ class AmqpRpcClient implements RpcClient {
     }
 
     if (builder.requestPostProcessor() == null) {
-      DefaultAddressBuilder<?> addressBuilder = new DefaultAddressBuilder<>(null) {};
+      DefaultAddressBuilder<?> addressBuilder = Utils.addressBuilder();
       addressBuilder.queue(replyTo);
       String replyToAddress = addressBuilder.address();
       this.requestPostProcessor =

src/main/java/com/rabbitmq/client/amqp/impl/Utils.java

@@ -153,6 +153,10 @@ static void throwIfInterrupted() throws InterruptedException {
     }
   }
 
+  static DefaultAddressBuilder<?> addressBuilder() {
+    return new DefaultAddressBuilder<>(null) {};
+  }
+
   private static class NameSupplier implements Supplier<String> {
 
     private final String prefix;

src/test/java/com/rabbitmq/client/amqp/impl/AddressFormatTest.java

@@ -193,7 +193,7 @@ void exchangeKeyInToField(TestInfo info) {
   }
 
   @Test
-  void noToFieldDoesNotCloseAllConnectionPublishers() throws InterruptedException {
+  void noToFieldDoesNotCloseAllConnectionPublishers() {
     Management management = connection.management();
     String q = management.queue().exclusive(true).declare().name();
     TestUtils.Sync sync = TestUtils.sync();

src/test/java/com/rabbitmq/client/amqp/impl/ClientTest.java

@@ -91,7 +91,7 @@ void deliveryCount() throws Exception {
                       publisher.message("".getBytes(UTF_8)), context -> publishLatch.countDown()));
 
       org.apache.qpid.protonj2.client.Connection protonConnection = connection(client);
-      Receiver receiver = protonConnection.openReceiver("/queue/" + q, new ReceiverOptions());
+      Receiver receiver = protonConnection.openReceiver("/q/" + q, new ReceiverOptions());
       int receivedMessages = 0;
       while (receiver.receive(100, TimeUnit.MILLISECONDS) != null) {
         receivedMessages++;
@@ -108,15 +108,15 @@ void largeMessage() throws Exception {
           connection(client, o -> o.traceFrames(false).maxFrameSize(maxFrameSize));
 
       Sender sender =
-          connection.openSender("/queue/" + q, new SenderOptions().deliveryMode(AT_LEAST_ONCE));
+          connection.openSender("/q/" + q, new SenderOptions().deliveryMode(AT_LEAST_ONCE));
       byte[] body = new byte[maxFrameSize * 4];
       Arrays.fill(body, (byte) 'A');
       Tracker tracker = sender.send(Message.create(body));
       tracker.awaitSettlement();
 
       Receiver receiver =
           connection.openReceiver(
-              "/queue/" + q,
+              "/q/" + q,
               new ReceiverOptions()
                   .deliveryMode(AT_LEAST_ONCE)
                   .autoSettle(false)
@@ -138,7 +138,7 @@ void largeMessageStreamSupport() throws Exception {
 
       StreamSender sender =
           connection.openStreamSender(
-              "/queue/" + q, new StreamSenderOptions().deliveryMode(AT_LEAST_ONCE));
+              "/q/" + q, new StreamSenderOptions().deliveryMode(AT_LEAST_ONCE));
       StreamSenderMessage message = sender.beginMessage();
       byte[] body = new byte[maxFrameSize * 4];
       Arrays.fill(body, (byte) 'A');
@@ -155,7 +155,7 @@ void largeMessageStreamSupport() throws Exception {
 
       StreamReceiver receiver =
           connection.openStreamReceiver(
-              "/queue/" + q,
+              "/q/" + q,
               new StreamReceiverOptions()
                   .deliveryMode(AT_LEAST_ONCE)
                   .autoAccept(false)
@@ -269,7 +269,7 @@ void queueDeletionImpactOnReceiver(TestInfo info) throws Exception {
 
       org.apache.qpid.protonj2.client.Connection protonConnection = connection(client);
       Session session = protonConnection.openSession();
-      Receiver receiver = session.openReceiver("/queue/" + queue);
+      Receiver receiver = session.openReceiver("/q/" + queue);
       receiver.openFuture().get();
       Delivery delivery = receiver.tryReceive();
       assertThat(delivery).isNull();
@@ -295,8 +295,7 @@ void exchangeDeletionImpactOnSender(TestInfo info) throws Exception {
       org.apache.qpid.protonj2.client.Connection protonConnection = connection(client);
       Session session = protonConnection.openSession();
       Sender sender =
-          session.openSender(
-              "/exchange/" + exchange, new SenderOptions().deliveryMode(AT_LEAST_ONCE));
+          session.openSender("/e/" + exchange, new SenderOptions().deliveryMode(AT_LEAST_ONCE));
       Tracker tracker = sender.send(Message.create(body));
       tracker.awaitSettlement(10, SECONDS);
       assertThat(tracker.remoteState()).isEqualTo(released());