[#1856] Fix negative authenticated AMQP connection count

Connection count was erroneously not incremented when a connection was rejected due to the connection limit of an adapter, but was decremented in that case as well which lead to negative connection counts. Fixes #1856 Signed-off-by: Florian Kaltner <[email protected]>
eclipse-hono · Jun 25, 2020 · 4e62471 · 4e62471
1 parent c267680
commit 4e62471
Show file tree

Hide file tree

Showing 2 changed files with 18 additions and 6 deletions.
diff --git a/...vertx/src/main/java/org/eclipse/hono/adapter/amqp/impl/VertxBasedAmqpProtocolAdapter.java b/...vertx/src/main/java/org/eclipse/hono/adapter/amqp/impl/VertxBasedAmqpProtocolAdapter.java
@@ -332,6 +332,13 @@ protected void onConnectRequest(final ProtonConnection con) {
             handleRemoteSenderOpenForCommands(con, sender);
         });
         con.openHandler(remoteOpen -> {
+            final Device authenticatedDevice = getAuthenticatedDevice(con);
+            if (authenticatedDevice == null) {
+                metrics.incrementUnauthenticatedConnections();
+            } else {
+                metrics.incrementConnections(authenticatedDevice.getTenantId());
+            }
+
             if (remoteOpen.failed()) {
                 log.debug("ignoring device's open frame containing error", remoteOpen.cause());
             } else {
@@ -417,11 +424,6 @@ private void processRemoteOpen(final ProtonConnection con) {
             con.open();
             log.debug("connection with device [container: {}] established", con.getRemoteContainer());
             span.log("connection established");
-            if (authenticatedDevice == null) {
-                metrics.incrementUnauthenticatedConnections();
-            } else {
-                metrics.incrementConnections(authenticatedDevice.getTenantId());
-            }
             return null;
         }).otherwise(t -> {
             con.setCondition(getErrorCondition(t));

diff --git a/...x/src/test/java/org/eclipse/hono/adapter/amqp/impl/VertxBasedAmqpProtocolAdapterTest.java b/...x/src/test/java/org/eclipse/hono/adapter/amqp/impl/VertxBasedAmqpProtocolAdapterTest.java
@@ -21,6 +21,7 @@
 import static org.mockito.ArgumentMatchers.argThat;
 import static org.mockito.ArgumentMatchers.eq;
 import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.inOrder;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.times;
@@ -88,6 +89,7 @@
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
 import org.mockito.ArgumentCaptor;
+import org.mockito.InOrder;
 
 import io.opentracing.Span;
 import io.opentracing.SpanContext;
@@ -1189,10 +1191,18 @@ public void testConnectionFailsIfAdapterLevelConnectionLimitIsExceeded() {
                 .forClass(Handler.class);
         verify(deviceConnection).openHandler(openHandler.capture());
         openHandler.getValue().handle(Future.succeededFuture(deviceConnection));
-        // THEN the adapter does not accept the incoming connection request. 
+        // THEN the connection count should be incremented when the connection is opened
+        final InOrder metricsInOrderVerifier = inOrder(metrics);
+        metricsInOrderVerifier.verify(metrics).incrementConnections(TEST_TENANT_ID);
+        // AND the adapter should close the connection right after it opened it
+        final ArgumentCaptor<Handler<AsyncResult<ProtonConnection>>> closeHandler = ArgumentCaptor.forClass(Handler.class);
+        verify(deviceConnection).closeHandler(closeHandler.capture());
+        closeHandler.getValue().handle(Future.succeededFuture());
         final ArgumentCaptor<ErrorCondition> errorConditionCaptor = ArgumentCaptor.forClass(ErrorCondition.class);
         verify(deviceConnection).setCondition(errorConditionCaptor.capture());
         assertEquals(AmqpError.UNAUTHORIZED_ACCESS, errorConditionCaptor.getValue().getCondition());
+        // AND the connection count should be decremented accordingly when the connection is closed
+        metricsInOrderVerifier.verify(metrics).decrementConnections(TEST_TENANT_ID);
         verify(metrics).reportConnectionAttempt(ConnectionAttemptOutcome.ADAPTER_CONNECTION_LIMIT_EXCEEDED);
     }