AbstractVertxBasedMqttProtocolAdapter extends AbstractProtocolAdapterBase<T> { public final Future<Void> uploadEventMessage( final MqttContext ctx, final String tenant, final String deviceId, final Buffer payload) { Objects.requireNonNull(ctx); Objects.requireNonNull(tenant); Objects.requireNonNull(deviceId); Objects.requireNonNull(payload); if (ctx.endpoint() != EndpointType.EVENT) { throw new IllegalArgumentException("context does not contain event but " + ctx.endpoint().getCanonicalName()); } final MetricsTags.QoS qos = MetricsTags.QoS.from(ctx.message().qosLevel().value()); final Future<TenantObject> tenantTracker = getTenantConfiguration(tenant, ctx.getTracingContext()); return tenantTracker .compose(tenantObject -> uploadMessage(ctx, tenantObject, deviceId, payload, getEventSender(tenant), ctx.endpoint())) .compose(success -> { metrics.reportTelemetry( ctx.endpoint(), ctx.tenant(), tenantTracker.result(), MetricsTags.ProcessingOutcome.FORWARDED, qos, payload.length(), ctx.getTimer()); return Future.<Void> succeededFuture(); }).recover(t -> { metrics.reportTelemetry( ctx.endpoint(), ctx.tenant(), tenantTracker.result(), ProcessingOutcome.from(t), qos, payload.length(), ctx.getTimer()); return Future.failedFuture(t); }); }  final void setAuthHandler(final AuthHandler<MqttConnectContext> authHandler); @Override int getPortDefaultValue(); @Override int getInsecurePortDefaultValue(); @Autowired final void setMetrics(final MqttAdapterMetrics metrics); void setMqttSecureServer(final MqttServer server); void setMqttInsecureServer(final MqttServer server); final Future<Void> uploadMessage(
            final MqttContext ctx,
            final ResourceIdentifier resource,
            final MqttPublishMessage message); final Future<Void> uploadTelemetryMessage(
            final MqttContext ctx,
            final String tenant,
            final String deviceId,
            final Buffer payload); final Future<Void> uploadEventMessage(
            final MqttContext ctx,
            final String tenant,
            final String deviceId,
            final Buffer payload); final Future<Void> uploadCommandResponseMessage(
            final MqttContext ctx,
            final ResourceIdentifier targetAddress);  }
@Test public void testOnUnauthenticatedMessageDoesNotSendPubAckOnFailure(final VertxTestContext ctx) { final Promise<ProtonDelivery> outcome = Promise.promise(); givenAnEventSenderForOutcome(outcome); final MqttServer server = getMqttServer(false); final AbstractVertxBasedMqttProtocolAdapter<MqttProtocolAdapterProperties> adapter = getAdapter(server); final Buffer payload = Buffer.buffer("some payload"); final MqttEndpoint endpoint = mockEndpoint(); when(endpoint.isConnected()).thenReturn(Boolean.TRUE); final MqttPublishMessage messageFromDevice = mock(MqttPublishMessage.class); when(messageFromDevice.qosLevel()).thenReturn(MqttQoS.AT_LEAST_ONCE); when(messageFromDevice.messageId()).thenReturn(5555555); when(messageFromDevice.topicName()).thenReturn("e/tenant/device"); final MqttContext context = newMqttContext(messageFromDevice, endpoint, span); adapter.uploadEventMessage(context, "my-tenant", "4712", payload).onComplete(ctx.failing()); outcome.fail(new ClientErrorException(HttpURLConnection.HTTP_BAD_REQUEST)); ctx.verify(() -> { verify(endpoint, never()).publishAcknowledge(anyInt()); verify(metrics, never()).reportTelemetry( any(MetricsTags.EndpointType.class), anyString(), any(), eq(MetricsTags.ProcessingOutcome.FORWARDED), any(MetricsTags.QoS.class), anyInt(), any()); }); ctx.completeNow(); }
@Test public void testMessageLimitExceededForAnEventMessage(final VertxTestContext ctx) { final AbstractVertxBasedMqttProtocolAdapter<MqttProtocolAdapterProperties> adapter = getAdapter( getMqttServer(false)); forceClientMocksToConnected(); when(resourceLimitChecks.isMessageLimitReached(any(TenantObject.class), anyLong(), any(SpanContext.class))) .thenReturn(Future.succeededFuture(Boolean.TRUE)); final DownstreamSender sender = mock(DownstreamSender.class); when(downstreamSenderFactory.getOrCreateEventSender(anyString())) .thenReturn(Future.succeededFuture(sender)); final MqttEndpoint client = mockEndpoint(); final MqttPublishMessage msg = mock(MqttPublishMessage.class); when(msg.topicName()).thenReturn("e/my-tenant/the-device"); when(msg.qosLevel()).thenReturn(MqttQoS.AT_LEAST_ONCE); adapter.uploadEventMessage( newMqttContext(msg, client, span), "my-tenant", "the-device", Buffer.buffer("test")).onComplete(ctx.failing(t -> { ctx.verify(() -> { verify(sender, never()).send(any(Message.class)); assertThat(((ClientErrorException) t).getErrorCode()) .isEqualTo(HttpUtils.HTTP_TOO_MANY_REQUESTS); verify(client, never()).publishAcknowledge(anyInt()); verify(metrics).reportTelemetry( any(MetricsTags.EndpointType.class), anyString(), any(), eq(MetricsTags.ProcessingOutcome.UNPROCESSABLE), any(MetricsTags.QoS.class), anyInt(), any()); }); ctx.completeNow(); })); }
@Test public void verifyEventMessageUsesTtlValueGivenInPropertyBag(final VertxTestContext ctx) { final AbstractVertxBasedMqttProtocolAdapter<MqttProtocolAdapterProperties> adapter = getAdapter( getMqttServer(false)); forceClientMocksToConnected(); final Promise<ProtonDelivery> outcome = Promise.promise(); outcome.complete(mock(ProtonDelivery.class)); final DownstreamSender sender = givenAnEventSenderForOutcome(outcome); final MqttPublishMessage msg = mock(MqttPublishMessage.class); when(msg.topicName()).thenReturn("e/tenant/device/?hono-ttl=30&param2=value2"); when(msg.qosLevel()).thenReturn(MqttQoS.AT_LEAST_ONCE); adapter.uploadEventMessage( newMqttContext(msg, mockEndpoint(), span), "tenant", "device", Buffer.buffer("test")).onComplete(ctx.succeeding(t -> { ctx.verify(() -> { final ArgumentCaptor<Message> messageCaptor = ArgumentCaptor.forClass(Message.class); verify(sender).sendAndWaitForOutcome(messageCaptor.capture(), (SpanContext) any()); assertThat(messageCaptor.getValue().getTtl()).isEqualTo(30 * 1000); }); ctx.completeNow(); })); }
@Test public void verifyEventMessageLimitsTtlToMaxValue(final VertxTestContext ctx) { final TenantObject myTenantConfig = TenantObject.from("tenant", true) .setResourceLimits(new ResourceLimits().setMaxTtl(10L)); when(tenantClient.get(eq("tenant"), (SpanContext) any())).thenReturn(Future.succeededFuture(myTenantConfig)); final AbstractVertxBasedMqttProtocolAdapter<MqttProtocolAdapterProperties> adapter = getAdapter( getMqttServer(false)); forceClientMocksToConnected(); final Promise<ProtonDelivery> outcome = Promise.promise(); outcome.complete(mock(ProtonDelivery.class)); final DownstreamSender sender = givenAnEventSenderForOutcome(outcome); final MqttPublishMessage msg = mock(MqttPublishMessage.class); when(msg.topicName()).thenReturn("e/tenant/device/?hono-ttl=30&param2=value2"); when(msg.qosLevel()).thenReturn(MqttQoS.AT_LEAST_ONCE); adapter.uploadEventMessage( newMqttContext(msg, mockEndpoint(), span), "tenant", "device", Buffer.buffer("test")).onComplete(ctx.succeeding(t -> { ctx.verify(() -> { final ArgumentCaptor<Message> messageCaptor = ArgumentCaptor.forClass(Message.class); verify(sender).sendAndWaitForOutcome(messageCaptor.capture(), (SpanContext) any()); assertThat(messageCaptor.getValue().getTtl()).isEqualTo(10 * 1000); }); ctx.completeNow(); })); }
@Test public void testUploadEventMessageSendsPubAckOnSuccess(final VertxTestContext ctx) { final Promise<ProtonDelivery> outcome = Promise.promise(); givenAnEventSenderForOutcome(outcome); testUploadQoS1MessageSendsPubAckOnSuccess( outcome, EndpointType.EVENT, (adapter, mqttContext) -> { adapter.uploadEventMessage(mqttContext, "my-tenant", "4712", mqttContext.message().payload()) .onComplete(ctx.completing()); }); }