The AMQP client sender throws an NPE when processing an acknowledgeme…

…nt as it assumes the delivery remote state is never null.

The sender is modified to handle the null case and report it as an unknown message state.

src/main/java/io/vertx/amqp/impl/AmqpSenderImpl.java

@@ -28,6 +28,7 @@
 import io.vertx.proton.ProtonSender;
 import io.vertx.proton.impl.ProtonSenderImpl;
 import org.apache.qpid.proton.amqp.messaging.Rejected;
+import org.apache.qpid.proton.amqp.transport.DeliveryState;
 
 public class AmqpSenderImpl implements AmqpSender {
   private static final Logger LOGGER = LoggerFactory.getLogger(AmqpSender.class);
@@ -138,6 +139,8 @@ public AmqpSender send(AmqpMessage message) {
 
   private AmqpSender doSend(AmqpMessage message, Handler<AsyncResult<Void>> acknowledgmentHandler) {
     Handler<ProtonDelivery> ack = delivery -> {
+      DeliveryState remoteState = delivery.getRemoteState();
+
       Handler<AsyncResult<Void>> handler = acknowledgmentHandler;
       if (acknowledgmentHandler == null) {
         handler = ar -> {
@@ -147,9 +150,14 @@ private AmqpSender doSend(AmqpMessage message, Handler<AsyncResult<Void>> acknow
         };
       }
 
-      switch (delivery.getRemoteState().getType()) {
+      if (remoteState == null) {
+        handler.handle(Future.failedFuture("Unknown message state"));
+        return;
+      }
+
+      switch (remoteState.getType()) {
         case Rejected:
-          handler.handle(Future.failedFuture("message rejected (REJECTED): " + ((Rejected) delivery.getRemoteState()).getError()));
+          handler.handle(Future.failedFuture("message rejected (REJECTED): " + ((Rejected) remoteState).getError()));
           break;
         case Modified:
           handler.handle(Future.failedFuture("message rejected (MODIFIED)"));
@@ -161,7 +169,7 @@ private AmqpSender doSend(AmqpMessage message, Handler<AsyncResult<Void>> acknow
           handler.handle(Future.succeededFuture());
           break;
         default:
-          handler.handle(Future.failedFuture("Unsupported delivery type: " + delivery.getRemoteState().getType()));
+          handler.handle(Future.failedFuture("Unsupported delivery type: " + remoteState.getType()));
       }
     };
 

src/test/java/io/vertx/amqp/SenderUnknownAckStateTest.java

@@ -0,0 +1,99 @@
+/*
+ * Copyright (c) 2018-2019 The original author or authors
+ *
+ * All rights reserved. This program and the accompanying materials
+ * are made available under the terms of the Eclipse Public License v1.0
+ * and Apache License v2.0 which accompanies this distribution.
+ *
+ *        The Eclipse Public License is available at
+ *        http://www.eclipse.org/legal/epl-v10.html
+ *
+ *        The Apache License v2.0 is available at
+ *        http://www.opensource.org/licenses/apache2.0.php
+ *
+ * You may elect to redistribute this code under either of these licenses.
+ */
+package io.vertx.amqp;
+
+import io.vertx.ext.unit.TestContext;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.util.*;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+public class SenderUnknownAckStateTest extends BareTestBase {
+
+  private AmqpConnection connection;
+  private String address;
+  private MockServer server;
+
+  @Before
+  public void init() throws Exception {
+    server = setupMockServer();
+
+    CountDownLatch latch = new CountDownLatch(1);
+    AtomicReference<AmqpConnection> reference = new AtomicReference<>();
+    client = AmqpClient.create(vertx, new AmqpClientOptions()
+      .setHost("localhost")
+      .setPort(server.actualPort()))
+      .connect(connection -> {
+        reference.set(connection.result());
+        if (connection.failed()) {
+          connection.cause().printStackTrace();
+        }
+        latch.countDown();
+      });
+
+    assertThat(latch.await(6, TimeUnit.SECONDS)).isTrue();
+    this.connection = reference.get();
+    assertThat(connection).isNotNull();
+    this.address = UUID.randomUUID().toString();
+  }
+
+  @After
+  public void tearDown() throws InterruptedException {
+    super.tearDown();
+    server.close();
+  }
+
+  @Test(timeout = 10000)
+  public void test(TestContext context) throws Exception {
+    connection.createSender(address, context.asyncAssertSuccess(sender -> {
+      AmqpMessage msg = AmqpMessage.create().withBooleanAsBody(true).build();
+      sender
+        .write(msg)
+        .onComplete(context.asyncAssertFailure(expected -> {
+        // Expected
+      }));
+    }));
+  }
+
+  private MockServer setupMockServer() throws Exception {
+    return new MockServer(vertx, serverConnection -> {
+      serverConnection.openHandler(serverSender -> {
+        serverConnection.closeHandler(x -> serverConnection.close());
+        serverConnection.open();
+      });
+
+      serverConnection.sessionOpenHandler(serverSession -> {
+        serverSession.closeHandler(x -> serverSession.close());
+        serverSession.open();
+      });
+
+      serverConnection.receiverOpenHandler(serverReceiver-> {
+        serverReceiver.handler((delivery, message) -> {
+          // Triggers message state to be null
+          delivery.settle();
+        });
+
+        serverReceiver.open();
+      });
+    });
+  }
+}