AbstractVertxBasedMqttProtocolAdapter extends AbstractProtocolAdapterBase<T> { public final Future<Void> uploadMessage( final MqttContext ctx, final ResourceIdentifier resource, final MqttPublishMessage message) { Objects.requireNonNull(ctx); Objects.requireNonNull(resource); Objects.requireNonNull(message); switch (MetricsTags.EndpointType.fromString(resource.getEndpoint())) { case TELEMETRY: return uploadTelemetryMessage( ctx, resource.getTenantId(), resource.getResourceId(), message.payload()); case EVENT: return uploadEventMessage( ctx, resource.getTenantId(), resource.getResourceId(), message.payload()); case COMMAND: return uploadCommandResponseMessage(ctx, resource); default: return Future .failedFuture(new ClientErrorException(HttpURLConnection.HTTP_BAD_REQUEST, "unsupported endpoint")); } }  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 testUploadMessageSupportsShortAndLongEndpointNames(final VertxTestContext ctx) { final MqttServer server = getMqttServer(false); final AbstractVertxBasedMqttProtocolAdapter<MqttProtocolAdapterProperties> adapter = getAdapter(server); final Promise<ProtonDelivery> outcome = Promise.promise(); outcome.complete(mock(ProtonDelivery.class)); givenAQoS1TelemetrySender(outcome); givenAnEventSenderForOutcome(outcome); final MqttEndpoint endpoint = mockEndpoint(); when(endpoint.isConnected()).thenReturn(Boolean.TRUE); final Buffer payload = Buffer.buffer("some payload"); final MqttPublishMessage messageFromDevice = mock(MqttPublishMessage.class); when(messageFromDevice.qosLevel()).thenReturn(MqttQoS.AT_LEAST_ONCE); when(messageFromDevice.messageId()).thenReturn(5555555); when(messageFromDevice.payload()).thenReturn(payload); ResourceIdentifier resourceId = ResourceIdentifier.from("telemetry", "my-tenant", "4712"); when(messageFromDevice.topicName()).thenReturn(resourceId.toString()); adapter.uploadMessage( newMqttContext(messageFromDevice, endpoint, span), resourceId, messageFromDevice).onComplete(ctx.succeeding()); resourceId = ResourceIdentifier.from("event", "my-tenant", "4712"); when(messageFromDevice.topicName()).thenReturn(resourceId.toString()); adapter.uploadMessage( newMqttContext(messageFromDevice, endpoint, span), resourceId, messageFromDevice).onComplete(ctx.succeeding()); resourceId = ResourceIdentifier.from("t", "my-tenant", "4712"); when(messageFromDevice.topicName()).thenReturn(resourceId.toString()); adapter.uploadMessage( newMqttContext(messageFromDevice, endpoint, span), resourceId, messageFromDevice).onComplete(ctx.succeeding()); resourceId = ResourceIdentifier.from("e", "my-tenant", "4712"); when(messageFromDevice.topicName()).thenReturn(resourceId.toString()); adapter.uploadMessage( newMqttContext(messageFromDevice, endpoint, span), resourceId, messageFromDevice).onComplete(ctx.succeeding()); resourceId = ResourceIdentifier.from("unknown", "my-tenant", "4712"); when(messageFromDevice.topicName()).thenReturn(resourceId.toString()); adapter.uploadMessage( newMqttContext(messageFromDevice, endpoint, span), resourceId, messageFromDevice).onComplete(ctx.failing(t -> { ctx.verify(() -> assertThat(t).isInstanceOf(ClientErrorException.class)); })); ctx.completeNow(); }
@Test public void testMessageLimitExceededForACommandResponseMessage(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 MqttPublishMessage msg = mock(MqttPublishMessage.class); when(msg.topicName()).thenReturn("e/tenant/device"); when(msg.qosLevel()).thenReturn(MqttQoS.AT_MOST_ONCE); when(msg.payload()).thenReturn(Buffer.buffer("test")); adapter.uploadMessage(newMqttContext(msg, mockEndpoint(), span), ResourceIdentifier.fromString(String.format("%s/tenant/device/res/%s/200", getCommandEndpoint(), Command.getRequestId("cmd123", "to", "deviceId"))), msg) .onComplete(ctx.failing(t -> { ctx.verify(() -> { assertThat(((ClientErrorException) t).getErrorCode()) .isEqualTo(HttpUtils.HTTP_TOO_MANY_REQUESTS); verify(metrics).reportCommand( eq(MetricsTags.Direction.RESPONSE), eq("tenant"), any(), eq(MetricsTags.ProcessingOutcome.UNPROCESSABLE), anyInt(), any()); }); ctx.completeNow(); })); }