HttpBasedMessageMapping implements MessageMapping<MqttContext> { @Override public Future<MappedMessage> mapMessage( final MqttContext ctx, final ResourceIdentifier targetAddress, final JsonObject registrationInfo) { Objects.requireNonNull(ctx); Objects.requireNonNull(registrationInfo); final Promise<MappedMessage> result = Promise.promise(); final Object mapperObject = registrationInfo.getValue(RegistrationConstants.FIELD_MAPPER); if (mapperObject instanceof String) { final String mapper = (String) mapperObject; if (mapper.isBlank()) { LOG.debug("no payload mapping configured for {}", ctx.authenticatedDevice()); result.complete(new MappedMessage(targetAddress, ctx.message().payload())); } else { final MapperEndpoint mapperEndpoint = mqttProtocolAdapterProperties.getMapperEndpoint(mapper); if (mapperEndpoint == null) { LOG.debug("no mapping endpoint [name: {}] found for {}", mapper, ctx.authenticatedDevice()); result.complete(new MappedMessage(targetAddress, ctx.message().payload())); } else { mapMessageRequest(ctx, targetAddress, registrationInfo, mapperEndpoint, result); } } } else { LOG.debug("no payload mapping configured for {}", ctx.authenticatedDevice()); result.complete(new MappedMessage(targetAddress, ctx.message().payload())); } return result.future(); } HttpBasedMessageMapping(final WebClient webClient, final MqttProtocolAdapterProperties mqttProtocolAdapterProperties); @Override Future<MappedMessage> mapMessage(
            final MqttContext ctx,
            final ResourceIdentifier targetAddress,
            final JsonObject registrationInfo);  }
@Test public void testMapMessageSucceedsIfNoMapperIsSet(final VertxTestContext ctx) { config.setMapperEndpoints(Map.of("mapper", MapperEndpoint.from("host", 1234, "/uri", false))); final ResourceIdentifier targetAddress = ResourceIdentifier.from(TelemetryConstants.TELEMETRY_ENDPOINT, TEST_TENANT_ID, "gateway"); final MqttPublishMessage message = newMessage(MqttQoS.AT_LEAST_ONCE, TelemetryConstants.TELEMETRY_ENDPOINT); final MqttContext context = newContext(message, span, new Device(TEST_TENANT_ID, "gateway")); messageMapping.mapMessage(context, targetAddress, new JsonObject()) .onComplete(ctx.succeeding(mappedMessage -> { ctx.verify(() -> { assertThat(mappedMessage.getTargetAddress()).isEqualTo(targetAddress); assertThat(mappedMessage.getPayload()).isEqualTo(message.payload()); assertThat(mappedMessage.getAdditionalProperties()).isEmpty(); verify(mapperWebClient, never()).post(anyInt(), anyString(), anyString()); }); ctx.completeNow(); })); }
@Test public void testMapMessageSucceedsIfNoMapperEndpointIsConfigured(final VertxTestContext ctx) { final ResourceIdentifier targetAddress = ResourceIdentifier.from(TelemetryConstants.TELEMETRY_ENDPOINT, TEST_TENANT_ID, "gateway"); final MqttPublishMessage message = newMessage(MqttQoS.AT_LEAST_ONCE, TelemetryConstants.TELEMETRY_ENDPOINT); final MqttContext context = newContext(message, span, new Device(TEST_TENANT_ID, "gateway")); messageMapping.mapMessage(context, targetAddress, new JsonObject().put(RegistrationConstants.FIELD_MAPPER, "mapper")) .onComplete(ctx.succeeding(mappedMessage -> { ctx.verify(() -> { assertThat(mappedMessage.getTargetAddress()).isEqualTo(targetAddress); assertThat(mappedMessage.getPayload()).isEqualTo(message.payload()); assertThat(mappedMessage.getAdditionalProperties()).isEmpty(); verify(mapperWebClient, never()).post(anyInt(), anyString(), anyString()); }); ctx.completeNow(); })); }
@SuppressWarnings("unchecked") @Test public void testMapMessageSucceeds(final VertxTestContext ctx) { config.setMapperEndpoints(Map.of("mapper", MapperEndpoint.from("host", 1234, "/uri", false))); final ResourceIdentifier targetAddress = ResourceIdentifier.from(TelemetryConstants.TELEMETRY_ENDPOINT, TEST_TENANT_ID, "gateway"); final String newDeviceId = "new-device"; final HttpRequest<Buffer> httpRequest = mock(HttpRequest.class, withSettings().defaultAnswer(RETURNS_SELF)); final MultiMap responseHeaders = MultiMap.caseInsensitiveMultiMap(); responseHeaders.add(MessageHelper.APP_PROPERTY_DEVICE_ID, newDeviceId); responseHeaders.add("foo", "bar"); final Buffer responseBody = Buffer.buffer("changed"); final HttpResponse<Buffer> httpResponse = mock(HttpResponse.class); when(httpResponse.headers()).thenReturn(responseHeaders); when(httpResponse.bodyAsBuffer()).thenReturn(responseBody); when(httpResponse.statusCode()).thenReturn(200); when(mapperWebClient.post(anyInt(), anyString(), anyString())).thenReturn(httpRequest); final MqttPublishMessage message = newMessage(MqttQoS.AT_LEAST_ONCE, TelemetryConstants.TELEMETRY_ENDPOINT); final MqttContext context = newContext(message, span, new Device(TEST_TENANT_ID, "gateway")); messageMapping.mapMessage(context, targetAddress, new JsonObject().put(RegistrationConstants.FIELD_MAPPER, "mapper")) .onComplete(ctx.succeeding(mappedMessage -> { ctx.verify(() -> { assertThat(mappedMessage.getTargetAddress().getResourceId()).isEqualTo("new-device"); assertThat(mappedMessage.getPayload()).isEqualTo(responseBody); assertThat(mappedMessage.getAdditionalProperties()).doesNotContainKey(MessageHelper.APP_PROPERTY_DEVICE_ID); assertThat(mappedMessage.getAdditionalProperties()).containsEntry("foo", "bar"); }); ctx.completeNow(); })); final ArgumentCaptor<Handler<AsyncResult<HttpResponse<Buffer>>>> captor = ArgumentCaptor.forClass(Handler.class); verify(httpRequest).sendBuffer(any(Buffer.class), captor.capture()); captor.getValue().handle(Future.succeededFuture(httpResponse)); }