expose detach method on links, add a matching detach handler for comp…

…letion/server usage

This fixes #39
This fixes #45

src/main/java/io/vertx/proton/ProtonLink.java

@@ -39,15 +39,27 @@ public interface ProtonLink<T extends ProtonLink<T>> {
   T open();
 
   /**
-   * Closes the AMQP link, i.e. allows the Detach frame to be emitted.
+   * Closes the AMQP link, i.e. allows the Detach frame to be emitted with closed=true set.
    *
-   * If the closure is being locally initiated, the {@link #closeHandler(Handler)} may be used to handle the peer
-   * sending their Detach frame.
+   * If the closure is being locally initiated, the {@link #closeHandler(Handler)} should be used to handle the peer
+   * sending their Detach frame with closed=true (and {@link #detachHandler(Handler)} can be used to handle the peer
+   * sending their Detach frame with closed=false).
    *
    * @return the link
    */
   T close();
 
+  /**
+   * Detaches the AMQP link, i.e. allows the Detach frame to be emitted with closed=false.
+   *
+   * If the detach is being locally initiated, the {@link #detachHandler(Handler)} should be used to handle the peer
+   * sending their Detach frame with closed=false (and {@link #closeHandler(Handler)} can be used to handle the peer
+   * sending their Detach frame with closed=true).
+   *
+   * @return the link
+   */
+  T detach();
+
   /**
    * Sets a handler for when an AMQP Attach frame is received from the remote peer.
    *
@@ -61,14 +73,23 @@ public interface ProtonLink<T extends ProtonLink<T>> {
   T openHandler(Handler<AsyncResult<T>> remoteOpenHandler);
 
   /**
-   * Sets a handler for when an AMQP Detach frame is received from the remote peer.
+   * Sets a handler for when an AMQP Detach frame with closed=true is received from the remote peer.
    *
    * @param remoteCloseHandler
    *          the handler
    * @return the link
    */
   T closeHandler(Handler<AsyncResult<T>> remoteCloseHandler);
 
+  /**
+   * Sets a handler for when an AMQP Detach frame with closed=false is received from the remote peer.
+   *
+   * @param remoteDetachHandler
+   *          the handler
+   * @return the link
+   */
+  T detachHandler(Handler<AsyncResult<Void>> remoteDetachHandler);
+
   /**
    * Gets the local QOS config.
    *

src/main/java/io/vertx/proton/impl/ProtonLinkImpl.java

@@ -38,6 +38,7 @@ abstract class ProtonLinkImpl<T extends ProtonLink<T>> implements ProtonLink<T>
   protected final Link link;
   private Handler<AsyncResult<T>> openHandler;
   private Handler<AsyncResult<T>> closeHandler;
+  private Handler<AsyncResult<Void>> detachHandler;
 
   ProtonLinkImpl(Link link) {
     this.link = link;
@@ -182,6 +183,7 @@ public T close() {
     return self();
   }
 
+  @Override
   public T detach() {
     link.detach();
     getSession().getConnectionImpl().flush();
@@ -200,6 +202,12 @@ public T closeHandler(Handler<AsyncResult<T>> closeHandler) {
     return self();
   }
 
+  @Override
+  public T detachHandler(Handler<AsyncResult<Void>> detachHandler) {
+    this.detachHandler = detachHandler;
+    return self();
+  }
+
   @Override
   public boolean isOpen() {
     return getLocalState() == EndpointState.ACTIVE;
@@ -252,6 +260,12 @@ void fireRemoteOpen() {
     }
   }
 
+  void fireRemoteDetach() {
+    if (detachHandler != null) {
+      detachHandler.handle(ProtonHelper.future(null, getRemoteCondition()));
+    }
+  }
+
   void fireRemoteClose() {
     if (closeHandler != null) {
       closeHandler.handle(ProtonHelper.future(self(), getRemoteCondition()));

src/main/java/io/vertx/proton/impl/ProtonTransport.java

@@ -133,6 +133,11 @@ private void handleSocketBuffer(Buffer buff) {
         }
         break;
       }
+      case LINK_REMOTE_DETACH: {
+        ProtonLinkImpl<?> link = (ProtonLinkImpl<?>) protonEvent.getLink().getContext();
+        link.fireRemoteDetach();
+        break;
+      }
       case LINK_REMOTE_CLOSE: {
         ProtonLinkImpl<?> link = (ProtonLinkImpl<?>) protonEvent.getLink().getContext();
         link.fireRemoteClose();
@@ -173,7 +178,6 @@ private void handleSocketBuffer(Buffer buff) {
       case LINK_INIT:
       case LINK_LOCAL_OPEN:
       case LINK_LOCAL_DETACH:
-      case LINK_REMOTE_DETACH:
       case LINK_LOCAL_CLOSE:
       case LINK_FINAL:
       }

src/test/java/io/vertx/proton/ProtonClientTest.java

@@ -1002,6 +1002,121 @@ public void validate(TestContext context, Map<Symbol, Object> props) {
 
   }
 
+  @Test(timeout = 20000)
+  public void testDetachHandlingWithSender(TestContext context) throws Exception {
+    doDetachHandlingTestImpl(context, true);
+  }
+
+  @Test(timeout = 20000)
+  public void testDetachHandlingWithReceiver(TestContext context) throws Exception {
+    doDetachHandlingTestImpl(context, false);
+  }
+
+  public void doDetachHandlingTestImpl(TestContext context, boolean clientSender) throws Exception {
+    server.close();
+
+    final Async clientLinkOpenAsync = context.async();
+    final Async serverLinkOpenAsync = context.async();
+    final Async serverLinkDetachAsync = context.async();
+    final Async clientLinkDetachAsync = context.async();
+    final AtomicBoolean serverLinkCloseHandlerFired = new AtomicBoolean();
+    final AtomicBoolean clientLinkCloseHandlerFired = new AtomicBoolean();
+
+    ProtonServer protonServer = null;
+    try {
+      protonServer = createServer((serverConnection) -> {
+        serverConnection.openHandler(result -> {
+          serverConnection.open();
+        });
+
+        serverConnection.sessionOpenHandler(session -> session.open());
+
+        if (clientSender) {
+          serverConnection.receiverOpenHandler(serverReceiver -> {
+            LOG.trace("Server receiver opened");
+            serverReceiver.open();
+            serverLinkOpenAsync.complete();
+
+            serverReceiver.closeHandler(res -> {
+              serverLinkCloseHandlerFired.set(true);
+            });
+
+            serverReceiver.detachHandler(res -> {
+              context.assertTrue(res.succeeded(), "expected non-errored async result");
+              serverReceiver.detach();
+              serverLinkDetachAsync.complete();
+            });
+          });
+        } else {
+          serverConnection.senderOpenHandler(serverSender -> {
+            LOG.trace("Server sender opened");
+            serverSender.open();
+            serverLinkOpenAsync.complete();
+
+            serverSender.closeHandler(res -> {
+              serverLinkCloseHandlerFired.set(true);
+            });
+
+            serverSender.detachHandler(res -> {
+              context.assertTrue(res.succeeded(), "expected non-errored async result");
+              serverSender.detach();
+              serverLinkDetachAsync.complete();
+            });
+          });
+        }
+      });
+
+      // ===== Client Handling =====
+
+      ProtonClient client = ProtonClient.create(vertx);
+      client.connect("localhost", protonServer.actualPort(), res -> {
+        context.assertTrue(res.succeeded());
+
+        ProtonConnection connection = res.result();
+        connection.openHandler(x -> {
+          LOG.trace("Client connection opened");
+          final ProtonLink<?> link;
+          if (clientSender) {
+            link = connection.createSender(null);
+          } else {
+            link = connection.createReceiver("some-address");
+          }
+
+          link.closeHandler(clientLink -> {
+            clientLinkCloseHandlerFired.set(true);
+          });
+
+          link.detachHandler(clientLink -> {
+            LOG.trace("Client link detached");
+            clientLinkDetachAsync.complete();
+          });
+
+          link.openHandler(y -> {
+            LOG.trace("Client link opened");
+            clientLinkOpenAsync.complete();
+
+            link.detach();
+          });
+          link.open();
+
+        }).open();
+      });
+
+      serverLinkOpenAsync.awaitSuccess();
+      clientLinkOpenAsync.awaitSuccess();
+
+      serverLinkDetachAsync.awaitSuccess();
+      clientLinkDetachAsync.awaitSuccess();
+    } finally {
+      if (protonServer != null) {
+        protonServer.close();
+      }
+    }
+
+    context.assertFalse(serverLinkCloseHandlerFired.get(), "server link close handler should not have fired");
+    context.assertFalse(clientLinkCloseHandlerFired.get(), "client link close handler should not have fired");
+  }
+
   private ProtonServer createServer(Handler<ProtonConnection> serverConnHandler) throws InterruptedException,
                                                                                  ExecutionException {
     ProtonServer server = ProtonServer.create(vertx);