KuraProtocolAdapter extends AbstractVertxBasedMqttProtocolAdapter<KuraAdapterProperties> { Future<ResourceIdentifier> mapTopic(final MqttContext ctx) { final Promise<ResourceIdentifier> result = Promise.promise(); final ResourceIdentifier topic = ctx.topic(); final ResourceIdentifier mappedTopic; if (getConfig().getControlPrefix().equals(topic.getEndpoint())) { ctx.setContentType(getConfig().getCtrlMsgContentType()); final String[] mappedPath = Arrays.copyOf(topic.getResourcePath(), topic.getResourcePath().length); mappedPath[0] = getEndpoint(ctx.message().qosLevel()); mappedTopic = ResourceIdentifier.fromPath(mappedPath); } else { ctx.setContentType(getConfig().getDataMsgContentType()); final String[] mappedPath = new String[topic.getResourcePath().length + 1]; System.arraycopy(topic.getResourcePath(), 0, mappedPath, 1, topic.getResourcePath().length); mappedPath[0] = getEndpoint(ctx.message().qosLevel()); mappedTopic = ResourceIdentifier.fromPath(mappedPath); } if (mappedTopic.getResourcePath().length < 3) { result.fail(new ClientErrorException(HttpURLConnection.HTTP_BAD_REQUEST, "topic does not comply with Kura format")); } else { log.debug("mapped Kura message [topic: {}, QoS: {}] to Hono message [to: {}, device_id: {}, content-type: {}]", topic, ctx.message().qosLevel(), mappedTopic.getBasePath(), mappedTopic.getResourceId(), ctx.contentType()); result.complete(mappedTopic); } return result.future(); }    }
@Test public void testMapTopicMapsKuraControlMessagesToEventApi(final VertxTestContext ctx) { final MqttContext context = newContext(MqttQoS.AT_LEAST_ONCE, "$EDC/my-scope/4711", span); adapter.mapTopic(context).onComplete(ctx.succeeding(msg -> { ctx.verify(() -> { assertAddress(msg, EventConstants.EVENT_ENDPOINT, "my-scope", "4711"); assertThat(context.contentType()).isEqualTo(config.getCtrlMsgContentType()); }); ctx.completeNow(); })); }
@Test public void testMapTopicMapsKuraControlMessagesToTelemetryApi(final VertxTestContext ctx) { config.setCtrlMsgContentType("control-msg"); final MqttContext context = newContext(MqttQoS.AT_MOST_ONCE, "$EDC/my-scope/4711", span); adapter.mapTopic(context).onComplete(ctx.succeeding(msg -> { ctx.verify(() -> { assertAddress(msg, TelemetryConstants.TELEMETRY_ENDPOINT, "my-scope", "4711"); assertThat(context.contentType()).isEqualTo(config.getCtrlMsgContentType()); }); ctx.completeNow(); })); }
@Test public void testMapTopicRecognizesControlMessagesWithCustomControlPrefix(final VertxTestContext ctx) { config.setControlPrefix("bumlux"); final MqttContext context = newContext(MqttQoS.AT_MOST_ONCE, "bumlux/my-scope/4711", span); adapter.mapTopic(context).onComplete(ctx.succeeding(msg -> { ctx.verify(() -> { assertAddress(msg, TelemetryConstants.TELEMETRY_ENDPOINT, "my-scope", "4711"); assertThat(context.contentType()).isEqualTo(config.getCtrlMsgContentType()); }); ctx.completeNow(); })); }
@Test public void testMapTopicMapsKuraDataMessagesToTelemetryApi(final VertxTestContext ctx) { config.setDataMsgContentType("data-msg"); final MqttContext context = newContext(MqttQoS.AT_MOST_ONCE, "my-scope/4711", span); adapter.mapTopic(context).onComplete(ctx.succeeding(msg -> { ctx.verify(() -> { assertAddress(msg, TelemetryConstants.TELEMETRY_ENDPOINT, "my-scope", "4711"); assertThat(context.contentType()).isEqualTo(config.getDataMsgContentType()); }); ctx.completeNow(); })); }
@Test public void testMapTopicMapsKuraDataMessagesToEventApi(final VertxTestContext ctx) { final MqttContext context = newContext(MqttQoS.AT_LEAST_ONCE, "my-scope/4711", span); adapter.mapTopic(context).onComplete(ctx.succeeding(msg -> { ctx.verify(() -> { assertAddress(msg, EventConstants.EVENT_ENDPOINT, "my-scope", "4711"); assertThat(context.contentType()).isEqualTo(config.getDataMsgContentType()); }); ctx.completeNow(); })); }