VertxBasedAmqpProtocolAdapter extends AbstractProtocolAdapterBase<AmqpAdapterProperties> { protected void onConnectRequest(final ProtonConnection con) { con.disconnectHandler(lostConnection -> { log.debug("lost connection to device [container: {}]", con.getRemoteContainer()); onConnectionLoss(con); decrementConnectionCount(con); }); con.closeHandler(remoteClose -> { handleRemoteConnectionClose(con, remoteClose); onConnectionLoss(con); decrementConnectionCount(con); }); con.sessionOpenHandler(session -> { HonoProtonHelper.setDefaultCloseHandler(session); handleSessionOpen(con, session); }); con.receiverOpenHandler(receiver -> { HonoProtonHelper.setDefaultCloseHandler(receiver); receiver.setMaxMessageSize(UnsignedLong.valueOf(getConfig().getMaxPayloadSize())); handleRemoteReceiverOpen(con, receiver); }); con.senderOpenHandler(sender -> { 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 { processRemoteOpen(remoteOpen.result()); } }); }  @Autowired void setMetrics(final AmqpAdapterMetrics metrics); @Override int getPortDefaultValue(); @Override int getInsecurePortDefaultValue();  }
@Test public void testAdapterSupportsAnonymousRelay() { final ProtonServer server = getAmqpServer(); final VertxBasedAmqpProtocolAdapter adapter = getAdapter(server); final Device authenticatedDevice = new Device(TEST_TENANT_ID, TEST_DEVICE); final Record record = new RecordImpl(); record.set(AmqpAdapterConstants.KEY_CLIENT_DEVICE, Device.class, authenticatedDevice); final ProtonConnection deviceConnection = mock(ProtonConnection.class); when(deviceConnection.attachments()).thenReturn(record); when(deviceConnection.getRemoteContainer()).thenReturn("deviceContainer"); adapter.onConnectRequest(deviceConnection); @SuppressWarnings("unchecked") final ArgumentCaptor<Handler<AsyncResult<ProtonConnection>>> openHandler = ArgumentCaptor.forClass(Handler.class); verify(deviceConnection).openHandler(openHandler.capture()); openHandler.getValue().handle(Future.succeededFuture(deviceConnection)); verify(deviceConnection).setOfferedCapabilities(argThat(caps -> Arrays.stream(caps).anyMatch(cap -> Constants.CAP_ANONYMOUS_RELAY.equals(cap)))); }
@SuppressWarnings("unchecked") @Test public void testConnectionCountForAnonymousDevice() { config.setAuthenticationRequired(false); final VertxBasedAmqpProtocolAdapter adapter = givenAnAmqpAdapter(); final ProtonConnection deviceConnection = mock(ProtonConnection.class); when(deviceConnection.attachments()).thenReturn(mock(Record.class)); adapter.onConnectRequest(deviceConnection); final ArgumentCaptor<Handler<AsyncResult<ProtonConnection>>> openHandler = ArgumentCaptor.forClass(Handler.class); verify(deviceConnection).openHandler(openHandler.capture()); openHandler.getValue().handle(Future.succeededFuture(deviceConnection)); verify(metrics).incrementUnauthenticatedConnections(); final ArgumentCaptor<Handler<ProtonConnection>> disconnectHandler = ArgumentCaptor.forClass(Handler.class); verify(deviceConnection).disconnectHandler(disconnectHandler.capture()); disconnectHandler.getValue().handle(deviceConnection); verify(metrics).decrementUnauthenticatedConnections(); final ArgumentCaptor<Handler<AsyncResult<ProtonConnection>>> closeHandler = ArgumentCaptor.forClass(Handler.class); verify(deviceConnection).closeHandler(closeHandler.capture()); closeHandler.getValue().handle(Future.succeededFuture()); verify(metrics, times(2)).decrementUnauthenticatedConnections(); }
@Test public void testConnectionFailsIfTenantLevelConnectionLimitIsExceeded() { config.setAuthenticationRequired(true); final VertxBasedAmqpProtocolAdapter adapter = givenAnAmqpAdapter(); when(resourceLimitChecks.isConnectionLimitReached(any(TenantObject.class), any(SpanContext.class))) .thenReturn(Future.succeededFuture(Boolean.TRUE)); final Device authenticatedDevice = new Device(TEST_TENANT_ID, TEST_DEVICE); final Record record = new RecordImpl(); record.set(AmqpAdapterConstants.KEY_CLIENT_DEVICE, Device.class, authenticatedDevice); final ProtonConnection deviceConnection = mock(ProtonConnection.class); when(deviceConnection.attachments()).thenReturn(record); adapter.onConnectRequest(deviceConnection); @SuppressWarnings("unchecked") final ArgumentCaptor<Handler<AsyncResult<ProtonConnection>>> openHandler = ArgumentCaptor .forClass(Handler.class); verify(deviceConnection).openHandler(openHandler.capture()); openHandler.getValue().handle(Future.succeededFuture(deviceConnection)); final ArgumentCaptor<ErrorCondition> errorConditionCaptor = ArgumentCaptor.forClass(ErrorCondition.class); verify(deviceConnection).setCondition(errorConditionCaptor.capture()); assertEquals(AmqpError.UNAUTHORIZED_ACCESS, errorConditionCaptor.getValue().getCondition()); verify(metrics).reportConnectionAttempt( ConnectionAttemptOutcome.TENANT_CONNECTIONS_EXCEEDED, TEST_TENANT_ID); }
@Test public void testConnectionFailsIfAdapterIsDisabled() { config.setAuthenticationRequired(true); final VertxBasedAmqpProtocolAdapter adapter = givenAnAmqpAdapter(); givenAConfiguredTenant(TEST_TENANT_ID, false); final Device authenticatedDevice = new Device(TEST_TENANT_ID, TEST_DEVICE); final Record record = new RecordImpl(); record.set(AmqpAdapterConstants.KEY_CLIENT_DEVICE, Device.class, authenticatedDevice); final ProtonConnection deviceConnection = mock(ProtonConnection.class); when(deviceConnection.attachments()).thenReturn(record); adapter.onConnectRequest(deviceConnection); @SuppressWarnings("unchecked") final ArgumentCaptor<Handler<AsyncResult<ProtonConnection>>> openHandler = ArgumentCaptor .forClass(Handler.class); verify(deviceConnection).openHandler(openHandler.capture()); openHandler.getValue().handle(Future.succeededFuture(deviceConnection)); final ArgumentCaptor<ErrorCondition> errorConditionCaptor = ArgumentCaptor.forClass(ErrorCondition.class); verify(deviceConnection).setCondition(errorConditionCaptor.capture()); assertEquals(AmqpError.UNAUTHORIZED_ACCESS, errorConditionCaptor.getValue().getCondition()); verify(metrics).reportConnectionAttempt( ConnectionAttemptOutcome.ADAPTER_DISABLED, TEST_TENANT_ID); }
@Test public void testConnectionFailsForAuthenticatedDeviceIfAdapterLevelConnectionLimitIsExceeded() { config.setAuthenticationRequired(true); final VertxBasedAmqpProtocolAdapter adapter = givenAnAmqpAdapter(); when(connectionLimitManager.isLimitExceeded()).thenReturn(true); final Device authenticatedDevice = new Device(TEST_TENANT_ID, TEST_DEVICE); final Record record = new RecordImpl(); record.set(AmqpAdapterConstants.KEY_CLIENT_DEVICE, Device.class, authenticatedDevice); final ProtonConnection deviceConnection = mock(ProtonConnection.class); when(deviceConnection.attachments()).thenReturn(record); adapter.onConnectRequest(deviceConnection); @SuppressWarnings("unchecked") final ArgumentCaptor<Handler<AsyncResult<ProtonConnection>>> openHandler = ArgumentCaptor .forClass(Handler.class); verify(deviceConnection).openHandler(openHandler.capture()); openHandler.getValue().handle(Future.succeededFuture(deviceConnection)); final InOrder metricsInOrderVerifier = inOrder(metrics); metricsInOrderVerifier.verify(metrics).incrementConnections(TEST_TENANT_ID); @SuppressWarnings("unchecked") 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()); metricsInOrderVerifier.verify(metrics).decrementConnections(TEST_TENANT_ID); verify(metrics).reportConnectionAttempt(ConnectionAttemptOutcome.ADAPTER_CONNECTIONS_EXCEEDED, TEST_TENANT_ID); }
@Test public void testConnectionFailsForUnauthenticatedDeviceIfAdapterLevelConnectionLimitIsExceeded() { config.setAuthenticationRequired(false); final VertxBasedAmqpProtocolAdapter adapter = givenAnAmqpAdapter(); when(connectionLimitManager.isLimitExceeded()).thenReturn(true); final ProtonConnection deviceConnection = mock(ProtonConnection.class); when(deviceConnection.attachments()).thenReturn(new RecordImpl()); adapter.onConnectRequest(deviceConnection); @SuppressWarnings("unchecked") final ArgumentCaptor<Handler<AsyncResult<ProtonConnection>>> openHandler = ArgumentCaptor .forClass(Handler.class); verify(deviceConnection).openHandler(openHandler.capture()); openHandler.getValue().handle(Future.succeededFuture(deviceConnection)); final InOrder metricsInOrderVerifier = inOrder(metrics); metricsInOrderVerifier.verify(metrics).incrementUnauthenticatedConnections(); @SuppressWarnings("unchecked") 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()); metricsInOrderVerifier.verify(metrics).decrementUnauthenticatedConnections(); verify(metrics).reportConnectionAttempt(ConnectionAttemptOutcome.ADAPTER_CONNECTIONS_EXCEEDED, null); }