AbstractVertxBasedMqttProtocolAdapter extends AbstractProtocolAdapterBase<T> { public final Future<Void> uploadTelemetryMessage( 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.TELEMETRY) { throw new IllegalArgumentException("context does not contain telemetry message 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, getTelemetrySender(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 testUploadTelemetryMessageIncludesRetainAnnotation(final VertxTestContext ctx) { final Promise<ProtonDelivery> outcome = Promise.promise(); outcome.complete(mock(ProtonDelivery.class)); final DownstreamSender sender = givenAQoS1TelemetrySender(outcome); final MqttServer server = getMqttServer(false); final AbstractVertxBasedMqttProtocolAdapter<MqttProtocolAdapterProperties> adapter = getAdapter(server); final MqttEndpoint endpoint = mockEndpoint(); when(endpoint.isConnected()).thenReturn(Boolean.TRUE); final Buffer payload = Buffer.buffer("hello"); final MqttPublishMessage messageFromDevice = mock(MqttPublishMessage.class); when(messageFromDevice.qosLevel()).thenReturn(MqttQoS.AT_LEAST_ONCE); when(messageFromDevice.messageId()).thenReturn(5555555); when(messageFromDevice.topicName()).thenReturn("t/my-tenant/4712"); when(messageFromDevice.isRetain()).thenReturn(Boolean.TRUE); when(messageFromDevice.payload()).thenReturn(payload); final MqttContext context = newMqttContext(messageFromDevice, endpoint, span); adapter.uploadTelemetryMessage(context, "my-tenant", "4712", payload).onComplete(ctx.succeeding(ok -> { ctx.verify(() -> { verify(endpoint).publishAcknowledge(5555555); final ArgumentCaptor<Message> msgCaptor = ArgumentCaptor.forClass(Message.class); verify(sender).sendAndWaitForOutcome(msgCaptor.capture(), (SpanContext) any()); assertThat(MessageHelper.getAnnotation(msgCaptor.getValue(), MessageHelper.ANNOTATION_X_OPT_RETAIN, Boolean.class)).isTrue(); verify(metrics).reportTelemetry( eq(MetricsTags.EndpointType.TELEMETRY), eq("my-tenant"), any(), eq(MetricsTags.ProcessingOutcome.FORWARDED), eq(MetricsTags.QoS.AT_LEAST_ONCE), eq(payload.length()), any()); }); ctx.completeNow(); })); }
@Test public void testMessageLimitExceededForATelemetryMessage(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.getOrCreateTelemetrySender(anyString())) .thenReturn(Future.succeededFuture(sender)); final MqttEndpoint client = mockEndpoint(); final MqttPublishMessage msg = mock(MqttPublishMessage.class); when(msg.topicName()).thenReturn("t/my-tenant/the-device"); when(msg.qosLevel()).thenReturn(MqttQoS.AT_LEAST_ONCE); adapter.uploadTelemetryMessage( 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 testUploadTelemetryMessageSendsPubAckOnSuccess(final VertxTestContext ctx) { final Promise<ProtonDelivery> outcome = Promise.promise(); givenAQoS1TelemetrySender(outcome); testUploadQoS1MessageSendsPubAckOnSuccess( outcome, EndpointType.TELEMETRY, (adapter, mqttContext) -> { adapter.uploadTelemetryMessage(mqttContext, "my-tenant", "4712", mqttContext.message().payload()) .onComplete(ctx.completing()); }); }