WIP: trying to use an external Nakadi mock

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

@@ -19,6 +19,12 @@
             <artifactId>nakadi-producer-spring-boot-starter</artifactId>
             <version>${project.version}</version>
         </dependency>
+        <dependency>
+        	<groupId>org.zalando</groupId>
+        	<artifactId>nakadi-mock</artifactId>
+        	<version>0.0.1-SNAPSHOT</version>
+        	<scope>test</scope>
+        </dependency>
         <dependency>
             <groupId>org.springframework.boot</groupId>
             <artifactId>spring-boot-starter-web</artifactId>
@@ -63,6 +69,7 @@
             <groupId>io.rest-assured</groupId>
             <artifactId>rest-assured</artifactId>
             <version>3.1.0</version>
+            <scope>test</scope>
         </dependency>
 
     </dependencies>

nakadi-producer-starter-spring-boot-2-test/src/main/java/org/zalando/nakadiproducer/tests/Application.java

@@ -5,19 +5,16 @@
 import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Primary;
-import org.zalando.nakadiproducer.EnableNakadiProducer;
 import org.zalando.nakadiproducer.snapshots.SimpleSnapshotEventGenerator;
 import org.zalando.nakadiproducer.snapshots.Snapshot;
 import org.zalando.nakadiproducer.snapshots.SnapshotEventGenerator;
 
 import javax.sql.DataSource;
 
 import java.io.IOException;
-import java.util.ArrayList;
 import java.util.Collections;
 
 @EnableAutoConfiguration
-@EnableNakadiProducer
 public class Application {
 
     public static void main(String[] args) throws Exception {
@@ -39,14 +36,24 @@ public EmbeddedPostgres embeddedPostgres() throws IOException {
     public SnapshotEventGenerator snapshotEventGenerator() {
         return new SimpleSnapshotEventGenerator("eventtype", (withIdGreaterThan, filter) -> {
             if (withIdGreaterThan == null) {
-                return Collections.singletonList(new Snapshot("1", "foo", filter));
+                return Collections.singletonList(new Snapshot("1", "foo", new Data("1", filter)));
             } else if (withIdGreaterThan.equals("1")) {
-                return Collections.singletonList(new Snapshot("2", "foo", filter));
+                return Collections.singletonList(new Snapshot("2", "foo", new Data("2", filter)));
             } else {
-                return new ArrayList<>();
+                return Collections.emptyList();
             }
         });
 
         // Todo: Test that some events arrive at a local nakadi mock
     }
+
+    static class Data {
+        public String id;
+        public String filter;
+        public Data(String id, String filter) {
+            super();
+            this.id = id;
+            this.filter = filter;
+        }
+    }
 }

nakadi-producer-starter-spring-boot-2-test/src/test/java/org/zalando/nakadiproducer/tests/ApplicationIT.java → nakadi-producer-starter-spring-boot-2-test/src/test/java/org/zalando/nakadiproducer/tests/ApplicationWithMockClientIT.java

@@ -22,11 +22,11 @@
         // This line looks like that by intention: We want to test that the MockNakadiPublishingClient will be picked up
         // by our starter *even if* it has been defined *after* the application itself. This has been a problem until
         // this commit.
-        classes = { Application.class, MockNakadiConfig.class },
-        properties = { "nakadi-producer.transmission-polling-delay=30"},
+        classes = { Application.class, MockNakadiClientConfig.class },
+        properties = { "nakadi-producer.transmission-polling-delay=30" },
         webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT
 )
-public class ApplicationIT {
+public class ApplicationWithMockClientIT {
     @LocalManagementPort
     private int localManagementPort;
 
@@ -54,7 +54,9 @@ public void cleanUpMock() {
 
     @Test
     public void shouldSuccessfullyStartAndSnapshotCanBeTriggered() throws InterruptedException {
-        given().baseUri("http://localhost:" + localManagementPort).contentType("application/json")
+        given().baseUri("http://localhost:" + localManagementPort)
+               .contentType("application/json")
+               .body("{'filter':'Example filter'}".replace('\'', '"'))
         .when().post("/actuator/snapshot-event-creation/eventtype")
         .then().statusCode(204);
 
@@ -64,6 +66,4 @@ public void shouldSuccessfullyStartAndSnapshotCanBeTriggered() throws Interrupte
         List<String> events = mockClient.getSentEvents("eventtype");
         assertThat(events, hasSize(2));
     }
-
-
 }

nakadi-producer-starter-spring-boot-2-test/src/test/java/org/zalando/nakadiproducer/tests/ApplicationWithMockNakadiIT.java

@@ -0,0 +1,73 @@
+package org.zalando.nakadiproducer.tests;
+
+import org.junit.BeforeClass;
+import org.junit.ClassRule;
+import org.junit.Test;
+import org.junit.contrib.java.lang.system.EnvironmentVariables;
+import org.junit.runner.RunWith;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.actuate.autoconfigure.web.server.LocalManagementPort;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit4.SpringRunner;
+import org.zalando.nakadi_mock.EventSubmissionCallback.CollectingCallback;
+import org.zalando.nakadi_mock.NakadiMock;
+import org.zalando.nakadiproducer.tests.Application.Data;
+import org.zalando.nakadiproducer.transmission.impl.EventTransmitter;
+
+import java.io.File;
+import java.util.List;
+
+import static io.restassured.RestAssured.given;
+import static org.hamcrest.Matchers.hasSize;
+import static org.junit.Assert.assertThat;
+
+@RunWith(SpringRunner.class)
+@SpringBootTest(
+        // This line looks like that by intention: We want to test that the MockNakadiPublishingClient will be picked up
+        // by our starter *even if* it has been defined *after* the application itself. This has been a problem until
+        // this commit.
+        classes = { MockNakadiServerConfig.class, Application.class },
+        webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT
+)
+@ContextConfiguration(initializers=MockNakadiServerConfig.MockPropertyInitializer.class)
+public class ApplicationWithMockNakadiIT {
+    @LocalManagementPort
+    private int localManagementPort;
+
+    @ClassRule
+    public static final EnvironmentVariables environmentVariables
+            = new EnvironmentVariables();
+
+    @BeforeClass
+    public static void fakeCredentialsDir() {
+        environmentVariables.set("CREDENTIALS_DIR", new File("src/test/resources/tokens").getAbsolutePath());
+    }
+
+    @Autowired
+    EventTransmitter transmitter;
+
+    @Autowired
+    NakadiMock nakadiMock;
+
+    @Test
+    public void shouldSuccessfullyStartAndSnapshotCanBeTriggered() throws InterruptedException {
+        CollectingCallback<Application.Data> collector = new CollectingCallback<Application.Data>() {};
+        nakadiMock.eventType("eventtype").setSubmissionCallback(collector);
+
+        given().baseUri("http://localhost:" + localManagementPort)
+               .contentType("application/json")
+               .body("{'filter':'Example filter'}".replace('\'', '"'))
+        .when().post("/actuator/snapshot-event-creation/eventtype")
+        .then().statusCode(204);
+
+        Thread.sleep(500);
+
+        transmitter.sendEvents();
+
+        Thread.sleep(500);
+
+        List<Data> events = collector.getSubmittedEvents();
+        assertThat(events, hasSize(2));
+    }
+}

nakadi-producer-starter-spring-boot-2-test/src/test/java/org/zalando/nakadiproducer/tests/MockNakadiConfig.java → nakadi-producer-starter-spring-boot-2-test/src/test/java/org/zalando/nakadiproducer/tests/MockNakadiClientConfig.java

@@ -6,7 +6,7 @@
 import org.zalando.nakadiproducer.transmission.NakadiPublishingClient;
 
 @Configuration
-public class MockNakadiConfig {
+public class MockNakadiClientConfig {
     @Bean
     public NakadiPublishingClient mockNakadiPublishingClient() {
         return new MockNakadiPublishingClient();

nakadi-producer-starter-spring-boot-2-test/src/test/java/org/zalando/nakadiproducer/tests/MockNakadiServerConfig.java

@@ -0,0 +1,34 @@
+package org.zalando.nakadiproducer.tests;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.boot.test.util.TestPropertyValues;
+import org.springframework.context.ApplicationContextInitializer;
+import org.springframework.context.ConfigurableApplicationContext;
+import org.springframework.context.annotation.Configuration;
+import org.zalando.nakadi_mock.NakadiMock;
+
+import java.net.URL;
+
+@Configuration
+public class MockNakadiServerConfig {
+
+    static class MockPropertyInitializer  implements ApplicationContextInitializer<ConfigurableApplicationContext>{
+
+        private static final Logger LOG = LoggerFactory.getLogger(MockPropertyInitializer.class);
+
+        @Override
+        public void initialize(ConfigurableApplicationContext context) {
+            // TODO: setup NakadiMock, inject URL into nakadi-producer
+
+            NakadiMock mock = NakadiMock.make();
+            context.getBeanFactory().registerSingleton("nakadiMock", mock);
+            mock.start();
+            URL url = mock.getRootUrl();
+
+            LOG.info("started mock nakadi on {}", url);
+
+            TestPropertyValues.of("nakadi-producer.nakadi-base-url="+url).applyTo(context);
+        }
+    }
+}

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

@@ -55,7 +55,7 @@ public void sendEvent(EventLog eventLog) {
             log.info("Event {} locked by {} was successfully transmitted to nakadi", eventLog.getId(), eventLog.getLockedBy());
             eventLogRepository.delete(eventLog);
         } catch (Exception e) {
-            log.error("Event {} locked by {} could not be transmitted to nakadi: {}", eventLog.getId(), eventLog.getLockedBy(), e.getMessage());
+            log.error("Event {} locked by {} could not be transmitted to nakadi: {}", eventLog.getId(), eventLog.getLockedBy(), e.toString());
         }
 
     }