Run all subscription catchups in series

CHANGELOG.md

@@ -4,6 +4,10 @@ on [Keep a CHANGELOG](http://keepachangelog.com/). This project adheres to
 [Semantic Versioning](http://semver.org/).
 
 ## [Unreleased]
+
+### Changed
+- Subscriptions are no longer run asynchronously during catchup
+
 ## [2.0.22] - 2019-09-24
 ### Changed
 - Catchup verification logging now runs in MDC context 

event-store-management/src/main/java/uk/gov/justice/services/eventstore/management/catchup/process/CatchupContext.java → event-store-management/src/main/java/uk/gov/justice/services/eventstore/management/catchup/process/CatchupSubscriptionContext.java

@@ -5,13 +5,13 @@
 
 import java.util.Objects;
 
-public class CatchupContext {
+public class CatchupSubscriptionContext {
 
     private final String componentName;
     private final Subscription subscription;
     private final CatchupRequestedEvent catchupRequestedEvent;
 
-    public CatchupContext(final String componentName, final Subscription subscription, final CatchupRequestedEvent catchupRequestedEvent) {
+    public CatchupSubscriptionContext(final String componentName, final Subscription subscription, final CatchupRequestedEvent catchupRequestedEvent) {
         this.componentName = componentName;
         this.subscription = subscription;
         this.catchupRequestedEvent = catchupRequestedEvent;
@@ -32,8 +32,8 @@ public CatchupRequestedEvent getCatchupRequestedEvent() {
     @Override
     public boolean equals(final Object o) {
         if (this == o) return true;
-        if (!(o instanceof CatchupContext)) return false;
-        final CatchupContext that = (CatchupContext) o;
+        if (!(o instanceof CatchupSubscriptionContext)) return false;
+        final CatchupSubscriptionContext that = (CatchupSubscriptionContext) o;
         return Objects.equals(componentName, that.componentName) &&
                 Objects.equals(subscription, that.subscription) &&
                 Objects.equals(catchupRequestedEvent, that.catchupRequestedEvent);

event-store-management/src/main/java/uk/gov/justice/services/eventstore/management/catchup/process/EventCatchupByComponentRunner.java

@@ -14,7 +14,7 @@
 public class EventCatchupByComponentRunner {
 
     @Inject
-    private EventCatchupBySubscriptionRunner eventCatchupBySubscriptionRunner;
+    private EventCatchupProcessorBean eventCatchupProcessorBean;
 
     @Inject
     private Logger logger;
@@ -37,11 +37,11 @@ private void runEventCatchupForSubscription(
 
         logger.info(format("Running catchup for Component '%s', Subscription '%s'", componentName, subscription.getName()));
 
-        final CatchupContext catchupContext = new CatchupContext(
+        final CatchupSubscriptionContext catchupSubscriptionContext = new CatchupSubscriptionContext(
                 componentName,
                 subscription,
                 catchupRequestedEvent);
 
-        eventCatchupBySubscriptionRunner.runEventCatchupForSubscription(catchupContext);
+        eventCatchupProcessorBean.performEventCatchup(catchupSubscriptionContext);
     }
 }

event-store-management/src/main/java/uk/gov/justice/services/eventstore/management/catchup/process/EventCatchupProcessor.java

@@ -50,13 +50,13 @@ public EventCatchupProcessor(
     }
 
     @Transactional(NEVER)
-    public void performEventCatchup(final CatchupContext catchupContext) {
+    public void performEventCatchup(final CatchupSubscriptionContext catchupSubscriptionContext) {
 
-        final Subscription subscription = catchupContext.getSubscription();
+        final Subscription subscription = catchupSubscriptionContext.getSubscription();
         final String subscriptionName = subscription.getName();
         final String eventSourceName = subscription.getEventSourceName();
-        final String componentName = catchupContext.getComponentName();
-        final CatchupRequestedEvent catchupRequestedEvent = catchupContext.getCatchupRequestedEvent();
+        final String componentName = catchupSubscriptionContext.getComponentName();
+        final CatchupRequestedEvent catchupRequestedEvent = catchupSubscriptionContext.getCatchupRequestedEvent();
 
         final PublishedEventSource eventSource = publishedEventSourceProvider.getPublishedEventSource(eventSourceName);
         final Long latestProcessedEventNumber = processedEventTrackingService.getLatestProcessedEventNumber(eventSourceName, componentName);

event-store-management/src/main/java/uk/gov/justice/services/eventstore/management/catchup/process/EventCatchupProcessorBean.java

@@ -16,10 +16,10 @@ public class EventCatchupProcessorBean {
     @Inject
     EventCatchupProcessorFactory eventCatchupProcessorFactory;
 
-    public void performEventCatchup(final CatchupContext catchupContext) {
+    public void performEventCatchup(final CatchupSubscriptionContext catchupSubscriptionContext) {
 
         final EventCatchupProcessor eventCatchupProcessor = eventCatchupProcessorFactory.create();
 
-        eventCatchupProcessor.performEventCatchup(catchupContext);
+        eventCatchupProcessor.performEventCatchup(catchupSubscriptionContext);
     }
 }

event-store-management/src/test/java/uk/gov/justice/services/eventstore/management/catchup/process/EventCatchupByComponentRunnerTest.java

@@ -22,7 +22,7 @@
 public class EventCatchupByComponentRunnerTest {
 
     @Mock
-    private EventCatchupBySubscriptionRunner eventCatchupBySubscriptionRunner;
+    private EventCatchupProcessorBean eventCatchupProcessorBean;
 
     @Mock
     private Logger logger;
@@ -51,12 +51,12 @@ public void shouldGetAllSubscriptionsForTheComponentAndRunCatchupOnEach() throws
 
         eventCatchupByComponentRunner.runEventCatchupForComponent(subscriptionsDescriptor, catchupRequestedEvent);
 
-        final InOrder inOrder = inOrder(logger, eventCatchupBySubscriptionRunner);
+        final InOrder inOrder = inOrder(logger, eventCatchupProcessorBean);
 
         inOrder.verify(logger).info("Running catchup for Component 'AN_EVENT_LISTENER', Subscription 'subscriptionName_1'");
-        inOrder.verify(eventCatchupBySubscriptionRunner).runEventCatchupForSubscription(new CatchupContext(componentName, subscription_1, catchupRequestedEvent));
+        inOrder.verify(eventCatchupProcessorBean).performEventCatchup(new CatchupSubscriptionContext(componentName, subscription_1, catchupRequestedEvent));
         inOrder.verify(logger).info("Running catchup for Component 'AN_EVENT_LISTENER', Subscription 'subscriptionName_2'");
-        inOrder.verify(eventCatchupBySubscriptionRunner).runEventCatchupForSubscription(new CatchupContext(componentName, subscription_2, catchupRequestedEvent));
+        inOrder.verify(eventCatchupProcessorBean).performEventCatchup(new CatchupSubscriptionContext(componentName, subscription_2, catchupRequestedEvent));
     }
 
     @Test
@@ -70,6 +70,6 @@ public void shouldNotRunCatchupForThisComponentIfTheComponentIsNotAnEventListene
 
         eventCatchupByComponentRunner.runEventCatchupForComponent(subscriptionsDescriptor, catchupRequestedEvent);
 
-        verifyZeroInteractions(eventCatchupBySubscriptionRunner);
+        verifyZeroInteractions(eventCatchupProcessorBean);
     }
 }

event-store-management/src/test/java/uk/gov/justice/services/eventstore/management/catchup/process/EventCatchupProcessorBeanTest.java

@@ -22,13 +22,13 @@ public class EventCatchupProcessorBeanTest {
     @Test
     public void shouldPerformEventCatchup() throws Exception {
 
-        final CatchupContext catchupContext = mock(CatchupContext.class);
+        final CatchupSubscriptionContext catchupSubscriptionContext = mock(CatchupSubscriptionContext.class);
         final EventCatchupProcessor eventCatchupProcessor = mock(EventCatchupProcessor.class);
 
         when(eventCatchupProcessorFactory.create()).thenReturn(eventCatchupProcessor);
 
-        eventCatchupProcessorBean.performEventCatchup(catchupContext);
+        eventCatchupProcessorBean.performEventCatchup(catchupSubscriptionContext);
 
-        verify(eventCatchupProcessor).performEventCatchup(catchupContext);
+        verify(eventCatchupProcessor).performEventCatchup(catchupSubscriptionContext);
     }
 }

event-store-management/src/test/java/uk/gov/justice/services/eventstore/management/catchup/process/EventCatchupProcessorTest.java

@@ -92,7 +92,7 @@ public void shouldFetchAllMissingEventsAndProcess() throws Exception {
         final Subscription subscription = mock(Subscription.class);
         final PublishedEventSource publishedEventSource = mock(PublishedEventSource.class);
         final CatchupRequestedEvent catchupRequestedEvent = mock(CatchupRequestedEvent.class);
-        final CatchupContext catchupContext = new CatchupContext(componentName, subscription, catchupRequestedEvent);
+        final CatchupSubscriptionContext catchupSubscriptionContext = new CatchupSubscriptionContext(componentName, subscription, catchupRequestedEvent);
         final SystemCommand systemCommand = mock(SystemCommand.class);
 
         final PublishedEvent publishedEvent_1 = mock(PublishedEvent.class);
@@ -116,7 +116,7 @@ public void shouldFetchAllMissingEventsAndProcess() throws Exception {
         when(eventStreamConsumerManager.add(publishedEvent_3, subscriptionName)).thenReturn(1);
         when(catchupRequestedEvent.getTarget()).thenReturn(systemCommand);
 
-        eventCatchupProcessor.performEventCatchup(catchupContext);
+        eventCatchupProcessor.performEventCatchup(catchupSubscriptionContext);
 
         final InOrder inOrder = inOrder(
                 catchupStartedForSubscriptionEventFirer,
@@ -159,7 +159,7 @@ public void shouldThrowExceptionIfEventNumberIsAbsentFromPublishedEvent() throws
         final Subscription subscription = mock(Subscription.class);
         final PublishedEventSource publishedEventSource = mock(PublishedEventSource.class);
         final CatchupRequestedEvent catchupRequestedEvent = mock(CatchupRequestedEvent.class);
-        final CatchupContext catchupContext = new CatchupContext(componentName, subscription, catchupRequestedEvent);
+        final CatchupSubscriptionContext catchupSubscriptionContext = new CatchupSubscriptionContext(componentName, subscription, catchupRequestedEvent);
         final SystemCommand systemCommand = mock(SystemCommand.class);
 
         final PublishedEvent publishedEvent_1 = mock(PublishedEvent.class);
@@ -185,7 +185,7 @@ public void shouldThrowExceptionIfEventNumberIsAbsentFromPublishedEvent() throws
         when(catchupRequestedEvent.getTarget()).thenReturn(systemCommand);
 
         try {
-            eventCatchupProcessor.performEventCatchup(catchupContext);
+            eventCatchupProcessor.performEventCatchup(catchupSubscriptionContext);
             fail();
         } catch (final MissingEventNumberException expected) {
             assertThat(expected.getMessage(), is("PublishedEvent with id '937f9fd6-3679-4bc2-a73c-6a7b18a651e1' is missing its event number"));