VertxBasedMqttProtocolAdapter extends AbstractVertxBasedMqttProtocolAdapter<MqttProtocolAdapterProperties> { @Override protected Future<Void> onPublishedMessage(final MqttContext ctx) { return mapTopic(ctx) .compose(address -> validateAddress(address, ctx.authenticatedDevice())) .compose(targetAddress -> mapMessage(ctx, targetAddress)) .compose(mappedMessage -> { final MqttPublishMessage mqttPublishMessage = MqttPublishMessage.create( ctx.message().messageId(), ctx.message().qosLevel(), ctx.message().isDup(), ctx.message().isRetain(), ctx.message().topicName(), mappedMessage.getPayload().getByteBuf()); return uploadMessage(ctx, mappedMessage.getTargetAddress(), mqttPublishMessage); }) .recover(t -> { log.debug("discarding message [topic: {}] from {}", ctx.message().topicName(), ctx.authenticatedDevice(), t); return Future.failedFuture(t); }); }  @Autowired void setMessageMapping(final MessageMapping<MqttContext> messageMappingService);  }
@Test public void testOnPublishedMessageFailsForMissingTenant(final VertxTestContext ctx) { givenAnAdapter(); final MqttContext context = newContext(MqttQoS.AT_MOST_ONCE, TelemetryConstants.TELEMETRY_ENDPOINT, span); adapter.onPublishedMessage(context).onComplete(ctx.failing(t -> { assertServiceInvocationException(ctx, t, HttpURLConnection.HTTP_BAD_REQUEST); ctx.completeNow(); })); }
@Test public void testOnPublishedMessageFailsForMissingDeviceId(final VertxTestContext ctx) { givenAnAdapter(); final MqttContext context = newContext(MqttQoS.AT_MOST_ONCE, TelemetryConstants.TELEMETRY_ENDPOINT + "/my-tenant", span); adapter.onPublishedMessage(context).onComplete(ctx.failing(t -> { assertServiceInvocationException(ctx, t, HttpURLConnection.HTTP_BAD_REQUEST); ctx.completeNow(); })); }
@Test public void testOnPublishedAuthenticatedMessageFailsForMissingDeviceId(final VertxTestContext ctx) { givenAnAdapter(); final MqttContext context = newContext( MqttQoS.AT_MOST_ONCE, TelemetryConstants.TELEMETRY_ENDPOINT + "/my-tenant", span, new Device("my-tenant", "device")); adapter.onPublishedMessage(context).onComplete(ctx.failing(t -> { assertServiceInvocationException(ctx, t, HttpURLConnection.HTTP_BAD_REQUEST); ctx.completeNow(); })); }
@Test public void testOnPublishedAuthenticatedMessageFailsForNonMatchingTenant(final VertxTestContext ctx) { givenAnAdapter(); final MqttContext context = newContext( MqttQoS.AT_MOST_ONCE, TelemetryConstants.TELEMETRY_ENDPOINT + "/other-tenant/4711", span, new Device("my-tenant", "gateway")); adapter.onPublishedMessage(context).onComplete(ctx.failing(t -> { assertServiceInvocationException(ctx, t, HttpURLConnection.HTTP_FORBIDDEN); ctx.completeNow(); })); }