Add jobstore usecase

.travis.yml

@@ -15,6 +15,7 @@ before_script:
 - psql -c 'create database frameworkviewstore;' -U postgres
 - psql -c 'create database fileservice;' -U postgres
 - psql -c 'create database frameworksystem;' -U postgres
+- psql -c 'create database frameworkjobstore;' -U postgres
 - psql -c "CREATE USER framework WITH PASSWORD 'framework';" -U postgres
 - psql -c "CREATE USER fileservice WITH PASSWORD 'fileservice';" -U postgres
 addons:

cakeshop-event/cakeshop-event-processor/pom.xml

@@ -49,6 +49,14 @@
             <groupId>uk.gov.justice.framework-generators</groupId>
             <artifactId>rest-client-core</artifactId>
         </dependency>
+        <dependency>
+            <groupId>uk.gov.justice.services</groupId>
+            <artifactId>jobstore-api</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>uk.gov.justice.services</groupId>
+            <artifactId>job-executor</artifactId>
+        </dependency>
 
         <!-- Test Dependencies -->
         <dependency>

cakeshop-event/cakeshop-event-processor/src/main/java/uk/gov/justice/services/cakeshop/event/processor/CakeMadeEventProcessor.java

@@ -1,22 +1,55 @@
 package uk.gov.justice.services.cakeshop.event.processor;
 
-import static uk.gov.justice.services.core.annotation.Component.EVENT_PROCESSOR;
-
+import javax.inject.Inject;
+import javax.json.JsonObject;
+import org.slf4j.Logger;
+import uk.gov.justice.services.cakeshop.jobstore.CakeMadeJobData;
+import uk.gov.justice.services.common.converter.ObjectToJsonObjectConverter;
+import uk.gov.justice.services.common.util.UtcClock;
 import uk.gov.justice.services.core.annotation.Handles;
 import uk.gov.justice.services.core.annotation.ServiceComponent;
 import uk.gov.justice.services.core.sender.Sender;
 import uk.gov.justice.services.messaging.JsonEnvelope;
+import uk.gov.moj.cpp.jobstore.api.ExecutionService;
+import uk.gov.moj.cpp.jobstore.api.task.ExecutionInfo;
 
-import javax.inject.Inject;
+import static uk.gov.justice.services.cakeshop.jobstore.CakeMadeNotificationTask.CAKE_MADE_NOTIFICATION_TASK;
+import static uk.gov.justice.services.core.annotation.Component.EVENT_PROCESSOR;
+import static uk.gov.moj.cpp.jobstore.api.task.ExecutionStatus.STARTED;
 
 @ServiceComponent(EVENT_PROCESSOR)
 public class CakeMadeEventProcessor {
 
     @Inject
     Sender sender;
 
+    @Inject
+    private ObjectToJsonObjectConverter objectToJsonObjectConverter;
+
+    @Inject
+    private UtcClock clock;
+
+    @Inject
+    private ExecutionService executionService;
+
+    @SuppressWarnings({"squid:S1312"})
+    @Inject
+    private Logger logger;
+
     @Handles("cakeshop.events.cake-made")
     public void handle(final JsonEnvelope event) {
+        final JsonObject cakeMadeEventPayload = event.payloadAsJsonObject();
+        final String cakeId = cakeMadeEventPayload.getString("cakeId");
+        final ExecutionInfo executionInfo = new ExecutionInfo(
+                objectToJsonObjectConverter.convert(new CakeMadeJobData(cakeId)),
+                CAKE_MADE_NOTIFICATION_TASK,
+                clock.now(),
+                STARTED);
+
+        executionService.executeWith(executionInfo);
+
+        logger.info("Cake made notification task submitted to job store");
+
         sender.send(event);
     }
 }

cakeshop-event/cakeshop-event-processor/src/main/java/uk/gov/justice/services/cakeshop/jobstore/CakeMadeJobData.java

@@ -0,0 +1,34 @@
+package uk.gov.justice.services.cakeshop.jobstore;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import org.apache.commons.lang3.builder.HashCodeBuilder;
+
+import static org.apache.commons.lang3.builder.EqualsBuilder.reflectionEquals;
+
+@SuppressWarnings("squid:S2384")
+public class CakeMadeJobData {
+
+    private final String cakeId;
+
+    public CakeMadeJobData(@JsonProperty("cakeId") final String cakeId) {
+        this.cakeId = cakeId;
+    }
+
+    public String getCakeId() {
+        return cakeId;
+    }
+
+    public String toString() {
+        return "CakeMadeJobData(cakeId=" + this.getCakeId() + ")";
+    }
+
+    @Override
+    public boolean equals(final Object o) {
+        return reflectionEquals(this, o);
+    }
+
+    @Override
+    public int hashCode() {
+        return new HashCodeBuilder(17, 37).append(cakeId).toHashCode();
+    }
+}

cakeshop-event/cakeshop-event-processor/src/main/java/uk/gov/justice/services/cakeshop/jobstore/CakeMadeNotificationTask.java

@@ -0,0 +1,57 @@
+package uk.gov.justice.services.cakeshop.jobstore;
+
+import java.util.UUID;
+import javax.inject.Inject;
+import org.slf4j.Logger;
+import uk.gov.justice.services.common.util.UtcClock;
+import uk.gov.justice.services.messaging.JsonEnvelope;
+import uk.gov.justice.services.messaging.jms.JmsSender;
+import uk.gov.justice.services.messaging.spi.DefaultJsonEnvelopeProvider;
+import uk.gov.justice.services.messaging.spi.DefaultJsonMetadata;
+import uk.gov.moj.cpp.jobstore.api.annotation.Task;
+import uk.gov.moj.cpp.jobstore.api.task.ExecutableTask;
+import uk.gov.moj.cpp.jobstore.api.task.ExecutionInfo;
+import uk.gov.moj.cpp.jobstore.api.task.ExecutionStatus;
+
+import static uk.gov.justice.services.cakeshop.jobstore.CakeMadeNotificationTask.CAKE_MADE_NOTIFICATION_TASK;
+
+@Task(CAKE_MADE_NOTIFICATION_TASK)
+public class CakeMadeNotificationTask implements ExecutableTask {
+
+    public static final String CAKE_MADE_NOTIFICATION_TASK = "cake-made-notification-task";
+
+    @Inject
+    private JmsSender jmsSender;
+
+    @SuppressWarnings({"squid:S1312"})
+    @Inject
+    private Logger logger;
+
+    @Inject
+    private UtcClock clock;
+
+    @Override
+    public ExecutionInfo execute(final ExecutionInfo executionInfo) {
+        try{
+            final JsonEnvelope jsonEnvelope = new DefaultJsonEnvelopeProvider().envelopeFrom(DefaultJsonMetadata.metadataBuilder()
+                            .withId(UUID.randomUUID())
+                            .createdAt(clock.now())
+                            .withName("jobstore.task.notification.cake-made").build(),
+                    executionInfo.getJobData());
+
+            jmsSender.send(jsonEnvelope, "public.event");
+
+            logger.info("Cake made notification sent successfully to public.event topic");
+
+            return ExecutionInfo.executionInfo()
+                    .withExecutionStatus(ExecutionStatus.COMPLETED)
+                    .build();
+
+        } catch (Exception e) {
+            logger.error("Error while sending cake made notification to public.event topic", e);
+            return ExecutionInfo.executionInfo()
+                    .withExecutionStatus(ExecutionStatus.COMPLETED)
+                    .build();
+        }
+    }
+}

cakeshop-integration-test/pom.xml

@@ -282,6 +282,22 @@
                             <goal>update</goal>
                         </goals>
                     </execution>
+
+                    <execution>
+                        <id>job-store-liquibase</id>
+                        <configuration>
+                            <changeLogFile>liquibase/jobstore-db-changelog.xml
+                            </changeLogFile>
+                            <url>jdbc:postgresql://localhost:5432/frameworkjobstore</url>
+                            <username>framework</username>
+                            <password>framework</password>
+                        </configuration>
+                        <phase>pre-integration-test</phase>
+                        <goals>
+                            <goal>dropAll</goal>
+                            <goal>update</goal>
+                        </goals>
+                    </execution>
                 </executions>
                 <dependencies>
                     <dependency>
@@ -309,6 +325,11 @@
                         <artifactId>file-service-liquibase</artifactId>
                         <version>${framework-libraries.version}</version>
                     </dependency>
+                    <dependency>
+                        <groupId>uk.gov.justice.services</groupId>
+                        <artifactId>jobstore-liquibase</artifactId>
+                        <version>${framework-libraries.version}</version>
+                    </dependency>
                     <dependency>
                         <groupId>uk.gov.justice.services</groupId>
                         <artifactId>framework-system-liquibase</artifactId>

cakeshop-integration-test/src/test/java/uk/gov/justice/services/cakeshop/it/CakeShopIT.java

@@ -1,5 +1,31 @@
 package uk.gov.justice.services.cakeshop.it;
 
+import com.fasterxml.jackson.databind.ObjectMapper;
+import java.util.Optional;
+import javax.jms.MessageConsumer;
+import javax.jms.Session;
+import javax.sql.DataSource;
+import javax.ws.rs.client.Client;
+import javax.ws.rs.core.Response;
+import org.apache.http.message.BasicNameValuePair;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import uk.gov.justice.services.cakeshop.it.helpers.ApiResponse;
+import uk.gov.justice.services.cakeshop.it.helpers.CommandFactory;
+import uk.gov.justice.services.cakeshop.it.helpers.CommandSender;
+import uk.gov.justice.services.cakeshop.it.helpers.DatabaseManager;
+import uk.gov.justice.services.cakeshop.it.helpers.EventFactory;
+import uk.gov.justice.services.cakeshop.it.helpers.EventFinder;
+import uk.gov.justice.services.cakeshop.it.helpers.JmsBootstrapper;
+import uk.gov.justice.services.cakeshop.it.helpers.Querier;
+import uk.gov.justice.services.cakeshop.it.helpers.RestEasyClientFactory;
+import uk.gov.justice.services.common.converter.ObjectToJsonObjectConverter;
+import uk.gov.justice.services.eventsourcing.repository.jdbc.event.Event;
+import uk.gov.justice.services.eventsourcing.repository.jdbc.event.EventJdbcRepository;
+import uk.gov.justice.services.eventsourcing.repository.jdbc.event.EventRepositoryFactory;
+import uk.gov.justice.services.test.utils.core.messaging.Poller;
+
 import static com.jayway.jsonassert.JsonAssert.with;
 import static java.util.Collections.singletonList;
 import static java.util.UUID.randomUUID;
@@ -13,6 +39,7 @@
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.core.Is.is;
 import static org.hamcrest.core.IsCollectionContaining.hasItem;
+import static org.junit.jupiter.api.Assertions.assertTrue;
 import static uk.gov.justice.services.cakeshop.it.params.CakeShopMediaTypes.ADD_RECIPE_MEDIA_TYPE;
 import static uk.gov.justice.services.cakeshop.it.params.CakeShopMediaTypes.POST_RECIPES_QUERY_MEDIA_TYPE;
 import static uk.gov.justice.services.cakeshop.it.params.CakeShopMediaTypes.QUERY_RECIPES_MEDIA_TYPE;
@@ -22,31 +49,12 @@
 import static uk.gov.justice.services.cakeshop.it.params.CakeShopUris.RECIPES_RESOURCE_URI;
 import static uk.gov.justice.services.test.utils.core.matchers.HttpStatusCodeMatcher.isStatus;
 
-import uk.gov.justice.services.cakeshop.it.helpers.ApiResponse;
-import uk.gov.justice.services.cakeshop.it.helpers.CommandFactory;
-import uk.gov.justice.services.cakeshop.it.helpers.EventFactory;
-import uk.gov.justice.services.cakeshop.it.helpers.EventFinder;
-import uk.gov.justice.services.cakeshop.it.helpers.Querier;
-import uk.gov.justice.services.cakeshop.it.helpers.RestEasyClientFactory;
-import uk.gov.justice.services.eventsourcing.repository.jdbc.event.Event;
-import uk.gov.justice.services.eventsourcing.repository.jdbc.event.EventJdbcRepository;
-import uk.gov.justice.services.eventsourcing.repository.jdbc.event.EventRepositoryFactory;
-import uk.gov.justice.services.cakeshop.it.helpers.CommandSender;
-import uk.gov.justice.services.cakeshop.it.helpers.DatabaseManager;
-
-import javax.sql.DataSource;
-import javax.ws.rs.client.Client;
-import javax.ws.rs.core.Response;
-
-import org.apache.http.message.BasicNameValuePair;
-import org.junit.jupiter.api.AfterEach;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
-
 public class CakeShopIT {
 
     private final DataSource eventStoreDataSource = new DatabaseManager().initEventStoreDb();
     private final EventJdbcRepository eventJdbcRepository = new EventRepositoryFactory().getEventJdbcRepository(eventStoreDataSource);
+    private final ObjectToJsonObjectConverter objectToJsonObjectConverter = new ObjectToJsonObjectConverter(new ObjectMapper());
+    private final JmsBootstrapper jmsBootstrapper = new JmsBootstrapper();
 
     private final EventFactory eventFactory = new EventFactory();
     private final EventFinder eventFinder = new EventFinder(eventJdbcRepository);
@@ -55,6 +63,7 @@ public class CakeShopIT {
     private Client client;
     private Querier querier;
     private CommandSender commandSender;
+    private final Poller poller = new Poller();
 
     @BeforeEach
     public void before() throws Exception {
@@ -80,6 +89,29 @@ public void shouldAcceptAddRecipeCommand() throws Exception {
         assertThat(response.getStatus(), isStatus(ACCEPTED));
     }
 
+    @Test
+    public void shouldSendNotificationThroughJobStoreTaskOnProcessingCakeMadeEvent() throws Exception {
+        try (final Session jmsSession = jmsBootstrapper.jmsSession()) {
+            try (final MessageConsumer publicTopicConsumer = jmsBootstrapper.topicConsumerOf("public.event", jmsSession)) {
+                jmsBootstrapper.clear(publicTopicConsumer);
+
+                final String recipeId = randomUUID().toString();
+                final String cakeId = randomUUID().toString();
+                final String cakeName = "Super cake";
+
+                commandSender.addRecipe(recipeId, cakeName);
+                await().until(() -> querier.recipesQueryResult().body().contains(recipeId));
+                final ApiResponse apiResponse = commandSender.makeCake(recipeId, cakeId);
+                assertThat(apiResponse.httpCode(), isStatus(ACCEPTED));
+
+                final Optional<String> eventPayload = jmsBootstrapper.getPayloadByEventName(publicTopicConsumer, "jobstore.task.notification.cake-made");
+                assertTrue(eventPayload.isPresent());
+                with(eventPayload.get())
+                        .assertThat("$.cakeId", equalTo(cakeId));
+            }
+        }
+    }
+
     @Test
     public void shouldAcceptRemoveRecipeCommand() throws Exception {
         final String recipeId = randomUUID().toString();

cakeshop-integration-test/src/test/java/uk/gov/justice/services/cakeshop/it/helpers/JmsBootstrapper.java

@@ -1,23 +1,28 @@
 package uk.gov.justice.services.cakeshop.it.helpers;
 
-import static javax.jms.Session.AUTO_ACKNOWLEDGE;
-
+import com.jayway.jsonpath.JsonPath;
+import java.util.Optional;
 import javax.jms.Connection;
 import javax.jms.JMSException;
+import javax.jms.Message;
 import javax.jms.MessageConsumer;
 import javax.jms.Queue;
 import javax.jms.QueueBrowser;
 import javax.jms.Session;
+import javax.jms.TextMessage;
 import javax.jms.Topic;
-
 import org.apache.activemq.artemis.jms.client.ActiveMQConnectionFactory;
+import uk.gov.justice.services.test.utils.core.messaging.Poller;
+
+import static javax.jms.Session.AUTO_ACKNOWLEDGE;
 
 public class JmsBootstrapper {
 
     private static final String JMS_USERNAME = SystemPropertyFinder.findJmsUserName();
     private static final String JMS_PASSWORD = SystemPropertyFinder.findJmsUserPassword();
     private static final String JMS_PORT = SystemPropertyFinder.findJmsPort();
     private static final String JMS_BROKER_URL = "tcp://localhost:" + JMS_PORT;
+    private final Poller poller = new Poller();
 
     private final ActiveMQConnectionFactory jmsConnectionFactory = new ActiveMQConnectionFactory(JMS_BROKER_URL);
 
@@ -50,8 +55,30 @@ public QueueBrowser queueBrowserOf(final String queueName, final Session session
         return session.createBrowser(queue);
     }
 
-    private void clear(MessageConsumer msgConsumer) throws JMSException {
+    public void clear(MessageConsumer msgConsumer) throws JMSException {
         while (msgConsumer.receiveNoWait() != null) {
         }
     }
+
+    public Optional<String> getPayloadByEventName(final MessageConsumer messageConsumerClient, final String expectedEventName) {
+        return poller.pollUntilFound(() -> {
+                    final Optional<String> eventPayload = retrieveEventPayload(messageConsumerClient);
+                    final boolean eventMatched = eventPayload
+                            .map(payload -> JsonPath.parse(payload).read("$._metadata.name", String.class))
+                            .filter(eventName -> eventName.equals(expectedEventName))
+                            .isPresent();
+
+                    return eventMatched ? eventPayload : Optional.empty();
+        });
+    }
+
+    private Optional<String> retrieveEventPayload(final MessageConsumer messageConsumerClient) {
+        try {
+            final Message message = messageConsumerClient.receive(1000*5);
+
+            return message != null ? Optional.ofNullable(((TextMessage) message).getText()) : Optional.empty();
+        } catch (JMSException e) {
+            return Optional.empty();
+        }
+    }
 }