VertxBasedMqttProtocolAdapter extends AbstractVertxBasedMqttProtocolAdapter<MqttProtocolAdapterProperties> { Future<ResourceIdentifier> mapTopic(final MqttContext context) { final Promise<ResourceIdentifier> result = Promise.promise(); final ResourceIdentifier topic = context.topic(); final MqttQoS qos = context.message().qosLevel(); switch (MetricsTags.EndpointType.fromString(topic.getEndpoint())) { case TELEMETRY: if (MqttQoS.EXACTLY_ONCE.equals(qos)) { result.fail(new ClientErrorException(HttpURLConnection.HTTP_BAD_REQUEST, "QoS 2 not supported for telemetry messages")); } else { result.complete(topic); } break; case EVENT: if (MqttQoS.AT_LEAST_ONCE.equals(qos)) { result.complete(topic); } else { result.fail(new ClientErrorException(HttpURLConnection.HTTP_BAD_REQUEST, "Only QoS 1 supported for event messages")); } break; case COMMAND: if (MqttQoS.EXACTLY_ONCE.equals(qos)) { result.fail(new ClientErrorException(HttpURLConnection.HTTP_BAD_REQUEST, "QoS 2 not supported for command response messages")); } else { result.complete(topic); } break; default: log.debug("no such endpoint [{}]", topic.getEndpoint()); result.fail(new ClientErrorException(HttpURLConnection.HTTP_NOT_FOUND, "no such endpoint")); } return result.future(); }  @Autowired void setMessageMapping(final MessageMapping<MqttContext> messageMappingService);  }
@Test public void testMapTopicFailsForUnknownEndpoint(final VertxTestContext ctx) { givenAnAdapter(); final MqttPublishMessage message = newMessage(MqttQoS.AT_MOST_ONCE, "unknown"); adapter.mapTopic(newContext(message, span, null)).onComplete(ctx.failing(t -> { assertServiceInvocationException(ctx, t, HttpURLConnection.HTTP_NOT_FOUND); ctx.completeNow(); })); }
@Test public void testMapTopicFailsForQoS2TelemetryMessage(final VertxTestContext ctx) { givenAnAdapter(); final MqttPublishMessage message = newMessage(MqttQoS.EXACTLY_ONCE, TelemetryConstants.TELEMETRY_ENDPOINT); adapter.mapTopic(newContext(message, span, null)).onComplete(ctx.failing(t -> { assertServiceInvocationException(ctx, t, HttpURLConnection.HTTP_BAD_REQUEST); ctx.completeNow(); })); }
@Test public void testMapTopicFailsForQoS0EventMessage(final VertxTestContext ctx) { givenAnAdapter(); final MqttPublishMessage message = newMessage(MqttQoS.AT_MOST_ONCE, EventConstants.EVENT_ENDPOINT); adapter.mapTopic(newContext(message, span, null)).onComplete(ctx.failing(t -> { assertServiceInvocationException(ctx, t, HttpURLConnection.HTTP_BAD_REQUEST); ctx.completeNow(); })); }
@Test public void testMapTopicFailsForQoS2EventMessage(final VertxTestContext ctx) { givenAnAdapter(); final MqttPublishMessage message = newMessage(MqttQoS.EXACTLY_ONCE, EventConstants.EVENT_ENDPOINT); adapter.mapTopic(newContext(message, span, null)).onComplete(ctx.failing(t -> { assertServiceInvocationException(ctx, t, HttpURLConnection.HTTP_BAD_REQUEST); ctx.completeNow(); })); }
@Test public void testMapTopicSupportsShortAndLongTopicNames(final VertxTestContext ctx) { givenAnAdapter(); MqttPublishMessage message = newMessage(MqttQoS.AT_LEAST_ONCE, EventConstants.EVENT_ENDPOINT); MqttContext context = newContext(message, span, null); adapter.mapTopic(context).onComplete(ctx.succeeding(address -> { ctx.verify(() -> assertThat(MetricsTags.EndpointType.fromString(address.getEndpoint())).isEqualTo(MetricsTags.EndpointType.EVENT)); })); message = newMessage(MqttQoS.AT_LEAST_ONCE, EventConstants.EVENT_ENDPOINT_SHORT); context = newContext(message, span, null); adapter.mapTopic(context).onComplete(ctx.succeeding(address -> { ctx.verify(() -> assertThat(MetricsTags.EndpointType.fromString(address.getEndpoint())).isEqualTo(MetricsTags.EndpointType.EVENT)); })); message = newMessage(MqttQoS.AT_LEAST_ONCE, TelemetryConstants.TELEMETRY_ENDPOINT); context = newContext(message, span, null); adapter.mapTopic(context).onComplete(ctx.succeeding(address -> { ctx.verify(() -> assertThat(MetricsTags.EndpointType.fromString(address.getEndpoint())).isEqualTo(MetricsTags.EndpointType.TELEMETRY)); })); message = newMessage(MqttQoS.AT_LEAST_ONCE, TelemetryConstants.TELEMETRY_ENDPOINT_SHORT); context = newContext(message, span, null); adapter.mapTopic(context).onComplete(ctx.succeeding(address -> { ctx.verify(() -> assertThat(MetricsTags.EndpointType.fromString(address.getEndpoint())).isEqualTo(MetricsTags.EndpointType.TELEMETRY)); })); message = newMessage(MqttQoS.AT_LEAST_ONCE, "unknown"); context = newContext(message, span, null); adapter.mapTopic(context).onComplete(ctx.failing()); ctx.completeNow(); }