MessageBodyInterpolator implements UriInterpolator { public URI interpolate(EndpointAddress endpoint, Message message) throws InterpolationException { UriTemplate template = templateCache.getUnchecked(endpoint.getEndpoint()); String[] variables = template.getVariables(); if (variables.length > 0) { Map<String, Object> values = Maps.newHashMap(); String payload = new String(message.getData(), Charsets.UTF_8); for (String variable : variables) { JsonPath path = variableCompiler.getUnchecked(variable); try { values.put(variable, path.read(payload)); } catch (InvalidPathException e) { throw new InterpolationException(String.format("Missing variable on path %s", path.getPath()), e); } } try { return URI.create(template.expand(values)); } catch (VariableExpansionException e) { throw new InterpolationException("Cannot expand template", e); } } return endpoint.getUri(); }  URI interpolate(EndpointAddress endpoint, Message message);  }
@Test public void willReturnURIAsIsIfNoTemplate() throws InterpolationException { EndpointAddress endpoint = EndpointAddress.of("http: URI interpolated = new MessageBodyInterpolator().interpolate(endpoint, SAMPLE_MSG); assertThat(interpolated).isEqualTo(endpoint.getUri()); }
@Test public void willInterpolateJsonPathFromTemplate() throws InterpolationException { EndpointAddress endpoint = EndpointAddress.of("http: URI expectedEndpoint = URI.create("http: String jsonMessage = "{\"some\": {\"object\": 100}}"; Message msg = withTestMessage() .withTopic("some.topic") .withContent(jsonMessage, StandardCharsets.UTF_8) .withPublishingTimestamp(121422L) .withReadingTimestamp(121423L) .withPartitionOffset(new PartitionOffset(KafkaTopicName.valueOf("kafka_topic"), 0, 0)) .build(); URI interpolated = new MessageBodyInterpolator().interpolate(endpoint, msg); assertThat(interpolated).isEqualTo(expectedEndpoint); }
@Test public void willReturnURIOnEmptyEndpoint() throws InterpolationException { EndpointAddress endpoint = EndpointAddress.of(""); URI expectedEndpoint = URI.create(""); URI interpolated = new MessageBodyInterpolator().interpolate(endpoint, SAMPLE_MSG); assertThat(interpolated).isEqualTo(expectedEndpoint); }
@Test public void willInterpolateMultipleJsonPathsFromTemplate() throws InterpolationException { EndpointAddress endpoint = EndpointAddress.of("http: URI expectedEndpoint = URI.create("http: String jsonMessage = "{\"some\": {\"object\": 100, \"test\": \"hello\"}}"; Message msg = withTestMessage() .withTopic("some.topic") .withContent(jsonMessage, StandardCharsets.UTF_8) .withPublishingTimestamp(12323L) .withReadingTimestamp(123234L) .build(); URI interpolated = new MessageBodyInterpolator().interpolate(endpoint, msg); assertThat(interpolated).isEqualTo(expectedEndpoint); }
@Test(expected = InterpolationException.class) public void willThrowExceptionOnInvalidPayload() throws InterpolationException { EndpointAddress endpoint = EndpointAddress.of("http: new MessageBodyInterpolator().interpolate(endpoint, SAMPLE_MSG); }
@Test(expected = InterpolationException.class) public void willThrowExceptionOnInterpolationException() throws InterpolationException { EndpointAddress endpoint = EndpointAddress.of("http: new MessageBodyInterpolator().interpolate(endpoint, SAMPLE_MSG); }
@Test public void willInterpolateMultipleJsonPathsFromTemplateInReverseOrder() throws InterpolationException { EndpointAddress endpoint = EndpointAddress.of("http: URI expectedEndpoint = URI.create("http: String jsonMessage = "{\"some\": {\"test\": \"hello\", \"object\": 100}}"; Message msg = withTestMessage() .withTopic("some.topic") .withContent(jsonMessage, StandardCharsets.UTF_8) .withPublishingTimestamp(1232443L) .withReadingTimestamp(12324434L) .withPartitionOffset(new PartitionOffset(KAFKA_TOPIC, 0, 0)) .build(); URI interpolated = new MessageBodyInterpolator().interpolate(endpoint, msg); assertThat(interpolated).isEqualTo(expectedEndpoint); }