Merge pull request #46 from zalando-incubator/bugfix/#44-flow_id-sending

#44 flow id sending

nakadi-producer-spring-boot-starter/pom.xml

@@ -10,7 +10,7 @@
     <parent>
         <groupId>org.zalando</groupId>
         <artifactId>nakadi-producer-reactor</artifactId>
-        <version>4.0.1</version>
+        <version>4.0.2</version>
     </parent>
 
     <artifactId>nakadi-producer-spring-boot-starter</artifactId>

nakadi-producer/pom.xml

@@ -10,7 +10,7 @@
     <parent>
         <groupId>org.zalando</groupId>
         <artifactId>nakadi-producer-reactor</artifactId>
-        <version>4.0.1</version>
+        <version>4.0.2</version>
     </parent>
 
     <artifactId>nakadi-producer</artifactId>
@@ -79,6 +79,11 @@
             <version>4.12</version>
             <scope>test</scope>
         </dependency>
+        <dependency>
+        	<groupId>com.jayway.jsonpath</groupId>
+        	<artifactId>json-path</artifactId>
+        	<scope>test</scope>
+        </dependency>
     </dependencies>
 
     <build>

nakadi-producer/src/main/java/org/zalando/nakadiproducer/transmission/impl/EventTransmissionService.java

@@ -62,6 +62,7 @@ public NakadiEvent mapToNakadiEvent(final EventLog event) {
         final NakadiMetadata metadata = new NakadiMetadata();
         metadata.setEid(convertToUUID(event.getId()));
         metadata.setOccuredAt(event.getCreated());
+        metadata.setFlowId(event.getFlowId());
         nakadiEvent.setMetadata(metadata);
 
         HashMap<String, Object> payloadDTO;

nakadi-producer/src/main/java/org/zalando/nakadiproducer/transmission/impl/NakadiMetadata.java

@@ -1,11 +1,11 @@
 package org.zalando.nakadiproducer.transmission.impl;
 
-import com.fasterxml.jackson.annotation.JsonFormat;
-import com.fasterxml.jackson.annotation.JsonProperty;
+import lombok.Data;
 
 import java.time.Instant;
 
-import lombok.Data;
+import com.fasterxml.jackson.annotation.JsonFormat;
+import com.fasterxml.jackson.annotation.JsonProperty;
 
 @Data
 public class NakadiMetadata {
@@ -17,4 +17,7 @@ public class NakadiMetadata {
     @JsonFormat(shape = JsonFormat.Shape.STRING)
     private Instant occuredAt;
 
+    @JsonProperty("flow_id")
+    private String flowId;
+
 }

nakadi-producer/src/test/java/org/zalando/nakadiproducer/transmission/impl/TransmissionServiceTest.java

@@ -0,0 +1,64 @@
+package org.zalando.nakadiproducer.transmission.impl;
+
+import static com.jayway.jsonpath.Criteria.where;
+import static com.jayway.jsonpath.JsonPath.read;
+import static java.time.Instant.now;
+import static org.hamcrest.Matchers.empty;
+import static org.hamcrest.Matchers.hasSize;
+import static org.hamcrest.Matchers.is;
+import static org.junit.Assert.assertThat;
+import static org.mockito.Mockito.mock;
+
+import java.util.List;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.zalando.nakadiproducer.eventlog.impl.EventLog;
+import org.zalando.nakadiproducer.eventlog.impl.EventLogRepository;
+import org.zalando.nakadiproducer.transmission.MockNakadiPublishingClient;
+import org.zalando.nakadiproducer.util.Fixture;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+
+public class TransmissionServiceTest {
+
+    private EventTransmissionService service;
+    private EventLogRepository repo;
+    private MockNakadiPublishingClient publishingClient;
+    private ObjectMapper mapper;
+
+    @Before
+    public void setUp() {
+        repo = mock(EventLogRepository.class);
+        publishingClient = new MockNakadiPublishingClient();
+        mapper = new ObjectMapper();
+        service = new EventTransmissionService(repo, publishingClient, mapper);
+    }
+
+    @Test
+    public void testWithFlowId() throws JsonProcessingException {
+        String flowId = "XYZ";
+        String payloadString = mapper.writeValueAsString(Fixture.mockPayload(42, "bla"));
+        EventLog ev = new EventLog(27, "type", payloadString, flowId, now(), now(), null, null);
+
+        service.sendEvent(ev);
+
+        List<String> events = publishingClient.getSentEvents("type");
+        assertThat(events, hasSize(1));
+        assertThat(read(events.get(0), "$.metadata.flow_id"), is(flowId));
+    }
+
+    @Test
+    public void testWithoutFlowId() throws JsonProcessingException {
+        String payloadString = mapper.writeValueAsString(Fixture.mockPayload(42, "bla"));
+        EventLog ev = new EventLog(27, "type", payloadString, null, now(), now(), null, null);
+
+        service.sendEvent(ev);
+
+        List<String> events = publishingClient.getSentEvents("type");
+        assertThat(events, hasSize(1));
+        assertThat(read(events.get(0), "$.metadata.[?]", where("flow_id").exists(true)), is(empty()));
+    }
+
+}

pom.xml

@@ -15,7 +15,7 @@
 
     <artifactId>nakadi-producer-reactor</artifactId>
     <groupId>org.zalando</groupId>
-    <version>4.0.1</version>
+    <version>4.0.2</version>
     <packaging>pom</packaging>
     <name>Nakadi Event Producer Reactor</name>