Extended Integaration Tests to include View Store

framework-tools-test/framework-event-listener/pom.xml

@@ -15,6 +15,7 @@
 
     <properties>
         <cpp.service-component>EVENT_LISTENER</cpp.service-component>
+        <deltaspike.version>1.6.1</deltaspike.version>
     </properties>
 
     <dependencies>
@@ -28,7 +29,23 @@
             <artifactId>javaee-api</artifactId>
             <scope>provided</scope>
         </dependency>
-     </dependencies>
+        <dependency>
+            <groupId>uk.gov.justice.services</groupId>
+            <artifactId>persistence-deltaspike</artifactId>
+            <version>2.2.1</version>
+         </dependency>
+
+        <dependency>
+            <groupId>org.apache.deltaspike.core</groupId>
+            <artifactId>deltaspike-core-api</artifactId>
+            <version>${deltaspike.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.deltaspike.modules</groupId>
+            <artifactId>deltaspike-data-module-api</artifactId>
+            <version>${deltaspike.version}</version>
+        </dependency>
+    </dependencies>
 
     <build>
         <plugins>

framework-tools-test/framework-event-listener/src/main/java/uk/gov/justice/framework/tools/entity/TestEvent.java

@@ -0,0 +1,82 @@
+package uk.gov.justice.framework.tools.entity;
+
+import java.io.Serializable;
+import java.util.Objects;
+import java.util.UUID;
+
+import javax.persistence.Column;
+import javax.persistence.Entity;
+import javax.persistence.Id;
+import javax.persistence.Table;
+
+@Entity
+@Table(name = "test")
+public class TestEvent implements Serializable {
+
+    @Id
+    @Column(name = "stream_id")
+    private UUID streamId;
+
+    @Column(name = "version_id")
+    private Integer versionId;
+
+    @Column(name = "data")
+    private String data;
+
+
+
+    public TestEvent(UUID streamId, Integer versionId, String data) {
+        this.streamId = streamId;
+        this.versionId = versionId;
+        this.data = data;
+    }
+
+    public UUID getStreamId() {
+        return streamId;
+    }
+
+    public void setStreamId(UUID streamId) {
+        this.streamId = streamId;
+    }
+
+    public Integer getVersionId() {
+        return versionId;
+    }
+
+    public void setVersionId(int versionId) {
+        this.versionId = versionId;
+    }
+
+    public String getData() {
+        return data;
+    }
+
+    public void setData(String data) {
+        this.data = data;
+    }
+
+    @Override
+    public String toString() {
+        return "TestEvent{" +
+                "streamId=" + streamId +
+                ", versionId=" + versionId +
+                ", data='" + data + '\'' +
+                '}';
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) return true;
+        if (o == null || getClass() != o.getClass()) return false;
+        TestEvent testEvent = (TestEvent) o;
+        return versionId == testEvent.versionId &&
+                Objects.equals(streamId, testEvent.streamId) &&
+                Objects.equals(data, testEvent.data);
+    }
+
+    @Override
+    public int hashCode() {
+
+        return Objects.hash(streamId, versionId, data);
+    }
+}

framework-tools-test/framework-event-listener/src/main/java/uk/gov/justice/framework/tools/listener/FrameworkToolsTestListener.java

@@ -1,10 +1,15 @@
 package uk.gov.justice.framework.tools.listener;
 
+import uk.gov.justice.framework.tools.entity.TestEvent;
+
+import uk.gov.justice.framework.tools.repository.TestViewstoreRepository;
 import uk.gov.justice.services.core.annotation.Component;
 import uk.gov.justice.services.core.annotation.Handles;
 import uk.gov.justice.services.core.annotation.ServiceComponent;
 import uk.gov.justice.services.messaging.JsonEnvelope;
 
+import javax.inject.Inject;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -13,8 +18,19 @@ public class FrameworkToolsTestListener {
 
     private static final Logger logger = LoggerFactory.getLogger(FrameworkToolsTestListener.class);
 
+    @Inject
+    private TestViewstoreRepository testViewstoreRepository;
+
     @Handles("framework.example-test")
     public void handle(final JsonEnvelope envelope) {
-        logger.info("caught a fish!");
+        testViewstoreRepository.save(fromJsonEnvelope(envelope));
+    }
+
+    private TestEvent fromJsonEnvelope(JsonEnvelope envelope) {
+
+        return new TestEvent(
+                        envelope.metadata().id(),
+                        envelope.metadata().version().get().intValue(),
+                        envelope.payloadAsJsonObject().toString());
     }
 }

framework-tools-test/framework-event-listener/src/main/java/uk/gov/justice/framework/tools/repository/TestViewstoreRepository.java

@@ -0,0 +1,13 @@
+package uk.gov.justice.framework.tools.repository;
+
+import uk.gov.justice.framework.tools.entity.TestEvent;
+
+import java.util.UUID;
+
+import org.apache.deltaspike.data.api.EntityRepository;
+import org.apache.deltaspike.data.api.Repository;
+
+@Repository
+public interface TestViewstoreRepository extends EntityRepository<TestEvent, UUID> {
+
+}

framework-tools-test/framework-event-listener/src/main/resources/META-INF/persistence.xml

@@ -0,0 +1,13 @@
+<?xml version="1.0" encoding="UTF-8" ?>
+<persistence xmlns="http://java.sun.com/xml/ns/persistence"
+             xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+             xsi:schemaLocation="http://java.sun.com/xml/ns/persistence
+ http://java.sun.com/xml/ns/persistence/persistence_1_0.xsd" version="1.0">
+
+    <persistence-unit name="replay-test">
+        <provider>org.hibernate.ejb.HibernatePersistence</provider>
+        <jta-data-source>java:/DS.replay</jta-data-source>
+        <class>uk.gov.justice.framework.tools.entity.TestEvent</class>
+    </persistence-unit>
+
+</persistence>

framework-tools-test/framework-tools-it/src/test/java/uk/gov/justice/framework/tools/replay/ReplayIntegrationIT.java

@@ -14,12 +14,17 @@
 import java.io.BufferedReader;
 import java.io.File;
 import java.io.FileFilter;
+import java.io.IOException;
 import java.io.InputStreamReader;
+import java.sql.Connection;
 import java.sql.PreparedStatement;
+import java.sql.ResultSet;
 import java.sql.SQLException;
 import java.time.ZonedDateTime;
 import java.util.UUID;
+import java.util.concurrent.TimeUnit;
 import java.util.regex.Pattern;
+import java.util.stream.Collectors;
 
 import javax.sql.DataSource;
 
@@ -36,39 +41,82 @@ public class ReplayIntegrationIT {
 
     private static final TestProperties TEST_PROPERTIES = new TestProperties("test.properties");
 
+    private static final int EVENT_COUNT = 5;
+
     private static final UUID STREAM_ID = randomUUID();
 
     private static TestEventLogRepository EVENT_LOG_REPOSITORY;
 
+    private static DataSource viewStoreDataSource;
 
     @Before
     public void setUpDB() throws Exception {
         EVENT_LOG_REPOSITORY = new TestEventLogRepository(initEventStoreDb());
+        viewStoreDataSource = initViewStoreDb();
     }
 
     @Test
     public void runReplayTool() throws Exception {
         insertEventLogData();
-        assertTrue(runCommand(createCommandToExecuteReplay()));
+        runCommand(createCommandToExecuteReplay());
+        assertTrue(viewStoreEventsPresent());
     }
 
+    public boolean viewStoreEventsPresent() throws SQLException {
+
+        boolean rc = false;
+
+        try (final Connection connection = viewStoreDataSource.getConnection();
+             final PreparedStatement ps = connection.prepareStatement("SELECT * FROM test")) {
+
+            int count = 0;
+            final ResultSet rs = ps.executeQuery();
+
+            while (rs.next()) {
+                count++;
+            }
+
+            if (count == EVENT_COUNT) {
+                rc = true;
+            }
+        }
+
+        return rc;
+    }
+
+
     @After
     public void tearDown() throws SQLException {
+
         final PreparedStatement preparedStatement = EVENT_LOG_REPOSITORY.getDataSource().getConnection().prepareStatement("delete from event_log");
         preparedStatement.executeUpdate();
         EVENT_LOG_REPOSITORY.getDataSource().getConnection().close();
+
+        final PreparedStatement viewStorePreparedStatement = viewStoreDataSource.getConnection().prepareStatement("delete from test");
+        viewStorePreparedStatement.executeUpdate();
+        viewStorePreparedStatement.getConnection().close();
     }
 
     private static DataSource initEventStoreDb() throws Exception {
-        return initDatabase("db.eventstore.url", "db.eventstore.userName",
-                "db.eventstore.password", "liquibase/event-store-db-changelog.xml", "liquibase/snapshot-store-db-changelog.xml", "liquibase/event-buffer-changelog.xml");
+        return initDatabase("db.eventstore.url",
+                       "db.eventstore.userName",
+                        "db.eventstore.password",
+                        "liquibase/event-store-db-changelog.xml", "liquibase/snapshot-store-db-changelog.xml");
     }
 
-    private EventLog eventLogFrom(final String eventName) {
+    private static DataSource initViewStoreDb() throws Exception {
+        return initDatabase("db.viewstore.url",
+                "db.eventstore.userName",
+                "db.eventstore.password",
+                "liquibase/viewstore-db-changelog.xml", "liquibase/event-buffer-changelog.xml", "liquibase/snapshot-store-db-changelog.xml");
+    }
+
+    private EventLog eventLogFrom(final String eventName, final Long sequenceId) {
 
         final JsonEnvelope jsonEnvelope = envelope()
                 .with(metadataWithRandomUUID(eventName)
                         .createdAt(ZonedDateTime.now())
+                        .withVersion(sequenceId)
                         .withStreamId(STREAM_ID).withVersion(1L))
                 .withPayloadOf("test", "a string")
                 .build();
@@ -77,7 +125,6 @@ private EventLog eventLogFrom(final String eventName) {
         final UUID id = metadata.id();
 
         final UUID streamId = metadata.streamId().get();
-        final Long sequenceId = 1L;
         final String name = metadata.name();
         final String payload = jsonEnvelope.payloadAsJsonObject().toString();
         final ZonedDateTime createdAt = metadata.createdAt().get();
@@ -145,27 +192,67 @@ private String getResource(final String pattern) {
         return dir.listFiles(fileFilter)[0].getAbsolutePath();
     }
 
-    public boolean runCommand(final String command) throws Exception {
+    public void runCommand(final String command) throws Exception {
 
         final Process exec = Runtime.getRuntime().exec(command);
-        final BufferedReader reader =
-                new BufferedReader(new InputStreamReader(exec.getInputStream()));
 
-        final Pattern p = Pattern.compile(".*caught a fish.*", Pattern.MULTILINE | Pattern.DOTALL);
-        boolean matches = false;
-        String line = "";
-        while ((line = reader.readLine()) != null) {
+        new Thread(() -> {
+            System.out.println("Redirecting output...");
+            try (final BufferedReader reader =
+                    new BufferedReader(new InputStreamReader(exec.getInputStream()))) {
+
+                final Pattern p = Pattern.compile(".*WFSWARM99999: WildFly Swarm is Ready.*", Pattern.MULTILINE | Pattern.DOTALL);
+                String line = "";
+                while ((line = reader.readLine()) != null) {
+
+                    System.out.println(line);
 
-            if (p.matcher(line).matches()) {
-                matches = true;
+                    if (p.matcher(line).matches()) {
+                        // Fraction has run so kill server now
+                        exec.destroyForcibly();
+                        break;
+                    }
+
+                }
+            }
+            catch (IOException ioEx) {
+                System.out.println("IOException occurred reading process input stream");
             }
-            System.out.println(line);
+
+            }).start();
+
+        System.out.println("Process started, waiting for completion..");
+
+        // Give the process 60 seconds to complete and then kill it. Successful test will be
+        // determined by querying the ViewStore for associated records later. The above Thread should
+        // kill the process inside 60 seconds but wait here and handle shutdown if things take
+        // too long for some reason
+        boolean processTerminated = exec.waitFor(60L, TimeUnit.SECONDS);
+
+        if (!processTerminated) {
+            System.err.println("WildFly Swarm process failed to terminate after 60 seconds!");
+            Process terminating = exec.destroyForcibly();
+
+            processTerminated = terminating.waitFor(10L, TimeUnit.SECONDS);
+            if (!processTerminated) {
+                System.err.println("Failed to forcibly terminate WildFly Swarm process!");
+            }
+            else {
+                System.err.println("WildFly Swarm process forcibly terminated.");
+            }
+        }
+        else {
+            System.out.println("WildFly Swarm process terminated by Test.");
         }
-        exec.destroyForcibly();
-        return matches;
+
     }
 
     private void insertEventLogData() throws SQLException, InvalidSequenceIdException {
-        EVENT_LOG_REPOSITORY.insert(eventLogFrom("framework.example-test"));
+        Long sequenceId = 0L;
+        EVENT_LOG_REPOSITORY.insert(eventLogFrom("framework.example-test", ++sequenceId));
+        EVENT_LOG_REPOSITORY.insert(eventLogFrom("framework.example-test", ++sequenceId));
+        EVENT_LOG_REPOSITORY.insert(eventLogFrom("framework.example-test", ++sequenceId));
+        EVENT_LOG_REPOSITORY.insert(eventLogFrom("framework.example-test", ++sequenceId));
+        EVENT_LOG_REPOSITORY.insert(eventLogFrom("framework.example-test", ++sequenceId));
     }
 }

framework-tools-test/framework-tools-it/src/test/resources/liquibase/liquibase.properties

@@ -0,0 +1 @@
+changeLogFile=liquibase/viewstore-db-changelog.xml

framework-tools-test/framework-tools-it/src/test/resources/liquibase/viewstore-db-changelog.xml

@@ -0,0 +1,9 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<databaseChangeLog xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+                   xmlns="http://www.liquibase.org/xml/ns/dbchangelog"
+                   xsi:schemaLocation="http://www.liquibase.org/xml/ns/dbchangelog
+         http://www.liquibase.org/xml/ns/dbchangelog/dbchangelog-3.3.xsd">
+
+    <includeAll relativeToChangelogFile="true" path="viewstore-db-changesets"/>
+
+</databaseChangeLog>