VertxBasedAmqpProtocolAdapter extends AbstractProtocolAdapterBase<AmqpAdapterProperties> { protected Future<ProtonDelivery> onMessageReceived(final AmqpContext ctx) { final Span msgSpan = ctx.getTracingSpan(); return validateEndpoint(ctx) .compose(validatedEndpoint -> validateAddress(validatedEndpoint.getAddress(), validatedEndpoint.getAuthenticatedDevice())) .compose(validatedAddress -> uploadMessage(ctx, validatedAddress, msgSpan)) .map(d -> { ProtonHelper.accepted(ctx.delivery(), true); return d; }).recover(t -> { if (t instanceof ClientErrorException) { final ErrorCondition condition = AbstractProtocolAdapterBase.getErrorCondition(t); MessageHelper.rejected(ctx.delivery(), condition); } else { ProtonHelper.released(ctx.delivery(), true); } TracingHelper.logError(msgSpan, t); return Future.failedFuture(t); }); }  @Autowired void setMetrics(final AmqpAdapterMetrics metrics); @Override int getPortDefaultValue(); @Override int getInsecurePortDefaultValue();  }
@Test public void testUploadTelemetryWithAtMostOnceDeliverySemantics(final VertxTestContext ctx) { final VertxBasedAmqpProtocolAdapter adapter = givenAnAmqpAdapter(); final DownstreamSender telemetrySender = givenATelemetrySenderForAnyTenant(); when(telemetrySender.send(any(Message.class), (SpanContext) any())).thenReturn(Future.succeededFuture(mock(ProtonDelivery.class))); final TenantObject tenantObject = givenAConfiguredTenant(TEST_TENANT_ID, true); final ProtonDelivery delivery = mock(ProtonDelivery.class); when(delivery.remotelySettled()).thenReturn(true); final Buffer payload = Buffer.buffer("payload"); final String to = ResourceIdentifier.from(TelemetryConstants.TELEMETRY_ENDPOINT, TEST_TENANT_ID, TEST_DEVICE).toString(); adapter.onMessageReceived(AmqpContext.fromMessage(delivery, getFakeMessage(to, payload), span, null)) .onComplete(ctx.succeeding(d -> { ctx.verify(() -> { verify(telemetrySender).send(any(Message.class), (SpanContext) any()); verify(delivery).disposition(any(Accepted.class), eq(true)); verify(metrics).reportTelemetry( eq(EndpointType.TELEMETRY), eq(TEST_TENANT_ID), eq(tenantObject), eq(ProcessingOutcome.FORWARDED), eq(QoS.AT_MOST_ONCE), eq(payload.length()), any()); }); ctx.completeNow(); })); }
@Test public void testUploadTelemetryWithAtLeastOnceDeliverySemantics() { final VertxBasedAmqpProtocolAdapter adapter = givenAnAmqpAdapter(); final DownstreamSender telemetrySender = givenATelemetrySenderForAnyTenant(); final Promise<ProtonDelivery> downstreamDelivery = Promise.promise(); when(telemetrySender.sendAndWaitForOutcome(any(Message.class), (SpanContext) any())) .thenReturn(downstreamDelivery.future()); final TenantObject tenantObject = givenAConfiguredTenant(TEST_TENANT_ID, true); final ProtonDelivery delivery = mock(ProtonDelivery.class); when(delivery.remotelySettled()).thenReturn(false); final Buffer payload = Buffer.buffer("payload"); final String to = ResourceIdentifier.from(TelemetryConstants.TELEMETRY_ENDPOINT_SHORT, TEST_TENANT_ID, TEST_DEVICE).toString(); final Message mockMessage = getFakeMessage(to, payload); adapter.onMessageReceived(AmqpContext.fromMessage(delivery, mockMessage, span, null)); verify(telemetrySender).sendAndWaitForOutcome(any(Message.class), (SpanContext) any()); verify(mockMessage).setAddress(TelemetryConstants.TELEMETRY_ENDPOINT + "/" + TEST_TENANT_ID); verify(delivery, never()).disposition(any(DeliveryState.class), anyBoolean()); downstreamDelivery.complete(mock(ProtonDelivery.class)); verify(delivery).disposition(any(Accepted.class), eq(true)); verify(metrics).reportTelemetry( eq(EndpointType.TELEMETRY), eq(TEST_TENANT_ID), eq(tenantObject), eq(ProcessingOutcome.FORWARDED), eq(QoS.AT_LEAST_ONCE), eq(payload.length()), any()); }
@Test public void testUploadTelemetryMessageFailsForDisabledAdapter(final VertxTestContext ctx) { final VertxBasedAmqpProtocolAdapter adapter = givenAnAmqpAdapter(); final DownstreamSender telemetrySender = givenATelemetrySenderForAnyTenant(); final TenantObject tenantObject = givenAConfiguredTenant(TEST_TENANT_ID, false); final ProtonDelivery delivery = mock(ProtonDelivery.class); when(delivery.remotelySettled()).thenReturn(false); final String to = ResourceIdentifier.from(TelemetryConstants.TELEMETRY_ENDPOINT, TEST_TENANT_ID, TEST_DEVICE).toString(); final Buffer payload = Buffer.buffer("some payload"); adapter.onMessageReceived(AmqpContext.fromMessage(delivery, getFakeMessage(to, payload), span, null)) .onComplete(ctx.failing(t -> { ctx.verify(() -> { verify(telemetrySender, never()).send(any(Message.class), (SpanContext) any()); verify(telemetrySender, never()).sendAndWaitForOutcome(any(Message.class), (SpanContext) any()); verify(delivery).disposition(any(Rejected.class), eq(true)); verify(metrics).reportTelemetry( eq(EndpointType.TELEMETRY), eq(TEST_TENANT_ID), eq(tenantObject), eq(ProcessingOutcome.UNPROCESSABLE), eq(QoS.AT_LEAST_ONCE), eq(payload.length()), any()); }); ctx.completeNow(); })); }
@Test public void testUploadEventFailsForGatewayOfDifferentTenant(final VertxTestContext ctx) { final VertxBasedAmqpProtocolAdapter adapter = givenAnAmqpAdapter(); final DownstreamSender eventSender = givenAnEventSender(Promise.promise()); givenAConfiguredTenant(TEST_TENANT_ID, true); final Device gateway = new Device(TEST_TENANT_ID, "gw"); final ProtonDelivery delivery = mock(ProtonDelivery.class); when(delivery.remotelySettled()).thenReturn(false); final String to = ResourceIdentifier.from(EventConstants.EVENT_ENDPOINT, "other-tenant", TEST_DEVICE).toString(); final Buffer payload = Buffer.buffer("some payload"); adapter.onMessageReceived(AmqpContext.fromMessage(delivery, getFakeMessage(to, payload), span, gateway)) .onComplete(ctx.failing(t -> { ctx.verify(() -> { verify(eventSender, never()).send(any(Message.class), (SpanContext) any()); verify(eventSender, never()).sendAndWaitForOutcome(any(Message.class), (SpanContext) any()); verify(delivery).disposition(any(Rejected.class), eq(true)); }); ctx.completeNow(); })); }
@Test public void testUploadCommandResponseSucceeds(final VertxTestContext ctx) { final VertxBasedAmqpProtocolAdapter adapter = givenAnAmqpAdapter(); final CommandResponseSender responseSender = givenACommandResponseSenderForAnyTenant(); final ProtonDelivery delivery = mock(ProtonDelivery.class); when(responseSender.sendCommandResponse(any(CommandResponse.class), (SpanContext) any())).thenReturn(Future.succeededFuture(delivery)); final TenantObject tenantObject = givenAConfiguredTenant(TEST_TENANT_ID, true); final String replyToAddress = String.format("%s/%s/%s", getCommandResponseEndpoint(), TEST_TENANT_ID, Command.getDeviceFacingReplyToId("test-reply-id", TEST_DEVICE)); final Map<String, Object> propertyMap = new HashMap<>(); propertyMap.put(MessageHelper.APP_PROPERTY_STATUS, 200); final ApplicationProperties props = new ApplicationProperties(propertyMap); final Buffer payload = Buffer.buffer("some payload"); final Message message = getFakeMessage(replyToAddress, payload); when(message.getCorrelationId()).thenReturn("correlation-id"); when(message.getApplicationProperties()).thenReturn(props); adapter.onMessageReceived(AmqpContext.fromMessage(delivery, message, span, null)).onComplete(ctx.succeeding(ok -> { ctx.verify(() -> { verify(responseSender).sendCommandResponse((CommandResponse) any(), (SpanContext) any()); verify(metrics).reportCommand( eq(Direction.RESPONSE), eq(TEST_TENANT_ID), eq(tenantObject), eq(ProcessingOutcome.FORWARDED), eq(payload.length()), any()); }); ctx.completeNow(); })); }
@Test public void testUploadCommandResponseWithoutPayloadSucceeds(final VertxTestContext ctx) { final VertxBasedAmqpProtocolAdapter adapter = givenAnAmqpAdapter(); final CommandResponseSender responseSender = givenACommandResponseSenderForAnyTenant(); final ProtonDelivery delivery = mock(ProtonDelivery.class); when(responseSender.sendCommandResponse(any(CommandResponse.class), (SpanContext) any())) .thenReturn(Future.succeededFuture(delivery)); final TenantObject tenantObject = givenAConfiguredTenant(TEST_TENANT_ID, true); final String replyToAddress = String.format("%s/%s/%s", getCommandResponseEndpoint(), TEST_TENANT_ID, Command.getDeviceFacingReplyToId("test-reply-id", TEST_DEVICE)); final Map<String, Object> propertyMap = new HashMap<>(); propertyMap.put(MessageHelper.APP_PROPERTY_STATUS, 200); final ApplicationProperties props = new ApplicationProperties(propertyMap); final Message message = getFakeMessage(replyToAddress, null); when(message.getCorrelationId()).thenReturn("correlation-id"); when(message.getApplicationProperties()).thenReturn(props); adapter.onMessageReceived(AmqpContext.fromMessage(delivery, message, span, null)).onComplete(ctx.succeeding(ok -> { ctx.verify(() -> { verify(responseSender).sendCommandResponse((CommandResponse) any(), (SpanContext) any()); verify(metrics).reportCommand( eq(Direction.RESPONSE), eq(TEST_TENANT_ID), eq(tenantObject), eq(ProcessingOutcome.FORWARDED), eq(0), any()); }); ctx.completeNow(); })); }
@Test public void testMessageLimitExceededForATelemetryMessage(final VertxTestContext ctx) { final VertxBasedAmqpProtocolAdapter adapter = givenAnAmqpAdapter(); final DownstreamSender telemetrySender = givenATelemetrySenderForAnyTenant(); final TenantObject tenantObject = givenAConfiguredTenant(TEST_TENANT_ID, true); when(resourceLimitChecks.isMessageLimitReached(any(TenantObject.class), anyLong(), any(SpanContext.class))) .thenReturn(Future.succeededFuture(Boolean.TRUE)); final ProtonDelivery delivery = mock(ProtonDelivery.class); when(delivery.remotelySettled()).thenReturn(false); final String to = ResourceIdentifier.from(TelemetryConstants.TELEMETRY_ENDPOINT, TEST_TENANT_ID, TEST_DEVICE) .toString(); final Buffer payload = Buffer.buffer("some payload"); adapter.onMessageReceived(AmqpContext.fromMessage(delivery, getFakeMessage(to, payload), span, null)) .onComplete(ctx.failing(t -> { ctx.verify(() -> { verify(telemetrySender, never()).send(any(Message.class), (SpanContext) any()); verify(telemetrySender, never()).sendAndWaitForOutcome(any(Message.class), (SpanContext) any()); assertThat(((ClientErrorException) t).getErrorCode()).isEqualTo(HttpUtils.HTTP_TOO_MANY_REQUESTS); verify(delivery).disposition(any(Rejected.class), eq(true)); verify(metrics).reportTelemetry( eq(EndpointType.TELEMETRY), eq(TEST_TENANT_ID), eq(tenantObject), eq(ProcessingOutcome.UNPROCESSABLE), eq(QoS.AT_LEAST_ONCE), eq(payload.length()), any()); }); ctx.completeNow(); })); }
@Test public void testMessageLimitExceededForAnEventMessage(final VertxTestContext ctx) { final VertxBasedAmqpProtocolAdapter adapter = givenAnAmqpAdapter(); final DownstreamSender eventSender = givenAnEventSender(Promise.promise()); final TenantObject tenantObject = givenAConfiguredTenant(TEST_TENANT_ID, true); when(resourceLimitChecks.isMessageLimitReached(any(TenantObject.class), anyLong(), any(SpanContext.class))) .thenReturn(Future.succeededFuture(Boolean.TRUE)); final ProtonDelivery delivery = mock(ProtonDelivery.class); when(delivery.remotelySettled()).thenReturn(false); final String to = ResourceIdentifier.from(EventConstants.EVENT_ENDPOINT, TEST_TENANT_ID, TEST_DEVICE) .toString(); final Buffer payload = Buffer.buffer("some payload"); adapter.onMessageReceived(AmqpContext.fromMessage(delivery, getFakeMessage(to, payload), span, null)) .onComplete(ctx.failing(t -> { ctx.verify(() -> { verify(eventSender, never()).send(any(Message.class), (SpanContext) any()); verify(eventSender, never()).sendAndWaitForOutcome(any(Message.class), (SpanContext) any()); assertThat(((ClientErrorException) t).getErrorCode()).isEqualTo(HttpUtils.HTTP_TOO_MANY_REQUESTS); verify(delivery).disposition(any(Rejected.class), eq(true)); verify(metrics).reportTelemetry( eq(EndpointType.EVENT), eq(TEST_TENANT_ID), eq(tenantObject), eq(ProcessingOutcome.UNPROCESSABLE), eq(QoS.AT_LEAST_ONCE), eq(payload.length()), any()); }); ctx.completeNow(); })); }
@Test public void testMessageLimitExceededForACommandResponseMessage(final VertxTestContext ctx) { final VertxBasedAmqpProtocolAdapter adapter = givenAnAmqpAdapter(); final CommandResponseSender responseSender = givenACommandResponseSenderForAnyTenant(); final ProtonDelivery delivery = mock(ProtonDelivery.class); final TenantObject tenantObject = givenAConfiguredTenant(TEST_TENANT_ID, true); when(resourceLimitChecks.isMessageLimitReached(any(TenantObject.class), anyLong(), any(SpanContext.class))) .thenReturn(Future.succeededFuture(Boolean.TRUE)); final String replyToAddress = String.format("%s/%s/%s", getCommandResponseEndpoint(), TEST_TENANT_ID, Command.getDeviceFacingReplyToId("test-reply-id", TEST_DEVICE)); final Buffer payload = Buffer.buffer("payload"); final Message message = getFakeMessage(replyToAddress, payload); when(message.getCorrelationId()).thenReturn("correlation-id"); final Map<String, Object> propertyMap = new HashMap<>(); propertyMap.put(MessageHelper.APP_PROPERTY_STATUS, 200); final ApplicationProperties props = new ApplicationProperties(propertyMap); when(message.getApplicationProperties()).thenReturn(props); adapter.onMessageReceived(AmqpContext.fromMessage(delivery, message, span, null)) .onComplete(ctx.failing(t -> { ctx.verify(() -> { verify(responseSender, never()).send(any(Message.class), any()); verify(responseSender, never()).sendAndWaitForOutcome(any(Message.class), any()); assertThat(((ClientErrorException) t).getErrorCode()).isEqualTo(HttpUtils.HTTP_TOO_MANY_REQUESTS); verify(delivery).disposition(any(Rejected.class), eq(true)); verify(metrics).reportCommand( eq(Direction.RESPONSE), eq(TEST_TENANT_ID), eq(tenantObject), eq(ProcessingOutcome.UNPROCESSABLE), eq(payload.length()), any()); }); ctx.completeNow(); })); }