-
Notifications
You must be signed in to change notification settings - Fork 3
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
Showing
19 changed files
with
569 additions
and
136 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
61 changes: 61 additions & 0 deletions
61
...ava/uk/gov/justice/services/eventsourcing/publishedevent/rebuild/BatchProcessDetails.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,61 @@ | ||
| package uk.gov.justice.services.eventsourcing.publishedevent.rebuild; | ||
|
|
||
| import java.util.Objects; | ||
| import java.util.concurrent.atomic.AtomicLong; | ||
|
|
||
| public class BatchProcessDetails { | ||
|
|
||
| private final AtomicLong previousEventNumber; | ||
| private final AtomicLong currentEventNumber; | ||
| private final int processCount; | ||
| private final boolean complete; | ||
|
|
||
| public BatchProcessDetails(final AtomicLong previousEventNumber, final AtomicLong currentEventNumber, final int processCount, final boolean complete) { | ||
| this.previousEventNumber = previousEventNumber; | ||
| this.currentEventNumber = currentEventNumber; | ||
| this.processCount = processCount; | ||
| this.complete = complete; | ||
| } | ||
|
|
||
| public AtomicLong getPreviousEventNumber() { | ||
| return previousEventNumber; | ||
| } | ||
|
|
||
| public AtomicLong getCurrentEventNumber() { | ||
| return currentEventNumber; | ||
| } | ||
|
|
||
| public int getProcessCount() { | ||
| return processCount; | ||
| } | ||
|
|
||
| public boolean isComplete() { | ||
| return complete; | ||
| } | ||
|
|
||
| @Override | ||
| public boolean equals(final Object o) { | ||
| if (this == o) return true; | ||
| if (!(o instanceof BatchProcessDetails)) return false; | ||
| final BatchProcessDetails that = (BatchProcessDetails) o; | ||
| return processCount == that.processCount && | ||
| complete == that.complete && | ||
| Objects.equals(previousEventNumber, that.previousEventNumber) && | ||
| Objects.equals(currentEventNumber, that.currentEventNumber); | ||
| } | ||
|
|
||
| @Override | ||
| public int hashCode() { | ||
| return Objects.hash(previousEventNumber, currentEventNumber, processCount, complete); | ||
| } | ||
|
|
||
| @Override | ||
| public String toString() { | ||
| return "BatchProcessDetails{" + | ||
| "previousEventNumber=" + previousEventNumber + | ||
| ", currentEventNumber=" + currentEventNumber + | ||
| ", processCount=" + processCount + | ||
| ", complete=" + complete + | ||
| '}'; | ||
| } | ||
| } |
45 changes: 45 additions & 0 deletions
45
...stice/services/eventsourcing/publishedevent/rebuild/BatchProcessingDetailsCalculator.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,45 @@ | ||
| package uk.gov.justice.services.eventsourcing.publishedevent.rebuild; | ||
|
|
||
| import uk.gov.justice.services.eventsourcing.repository.jdbc.event.MissingEventNumberException; | ||
| import uk.gov.justice.services.eventsourcing.repository.jdbc.event.PublishedEvent; | ||
|
|
||
| import java.util.List; | ||
| import java.util.concurrent.atomic.AtomicLong; | ||
|
|
||
| public class BatchProcessingDetailsCalculator { | ||
|
|
||
| public BatchProcessDetails createFirstBatchProcessDetails() { | ||
|
|
||
| return new BatchProcessDetails( | ||
| new AtomicLong(0), | ||
| new AtomicLong(0), | ||
| 0, | ||
| false | ||
| ); | ||
| } | ||
|
|
||
| public BatchProcessDetails calculateNextBatchProcessDetails( | ||
| final BatchProcessDetails currentBatchProcessDetails, | ||
| final AtomicLong previousEventNumber, | ||
| final List<PublishedEvent> publishedEvents) { | ||
|
|
||
| if (publishedEvents.isEmpty()) { | ||
| return new BatchProcessDetails( | ||
| previousEventNumber, | ||
| currentBatchProcessDetails.getCurrentEventNumber(), | ||
| currentBatchProcessDetails.getProcessCount(), | ||
| true | ||
| ); | ||
| } | ||
|
|
||
| final PublishedEvent lastPublishedEvent = publishedEvents.get(publishedEvents.size() - 1); | ||
|
|
||
| final Long newCurrentEventNumber = lastPublishedEvent.getEventNumber().orElseThrow(() -> new MissingEventNumberException("")); | ||
| return new BatchProcessDetails( | ||
| previousEventNumber, | ||
| new AtomicLong(newCurrentEventNumber), | ||
| currentBatchProcessDetails.getProcessCount() + publishedEvents.size(), | ||
| false | ||
| ); | ||
| } | ||
| } |
65 changes: 65 additions & 0 deletions
65
...v/justice/services/eventsourcing/publishedevent/rebuild/BatchPublishedEventProcessor.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,65 @@ | ||
| package uk.gov.justice.services.eventsourcing.publishedevent.rebuild; | ||
|
|
||
| import static java.lang.String.format; | ||
| import static javax.transaction.Transactional.TxType.REQUIRED; | ||
|
|
||
| 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.PublishedEvent; | ||
|
|
||
| import java.util.ArrayList; | ||
| import java.util.List; | ||
| import java.util.Set; | ||
| import java.util.UUID; | ||
| import java.util.concurrent.atomic.AtomicLong; | ||
| import java.util.stream.Stream; | ||
|
|
||
| import javax.inject.Inject; | ||
| import javax.transaction.Transactional; | ||
|
|
||
| import org.slf4j.Logger; | ||
|
|
||
| public class BatchPublishedEventProcessor { | ||
|
|
||
| private static final int PAGE_SIZE = 1_000; | ||
|
|
||
| @Inject | ||
| private EventJdbcRepository eventJdbcRepository; | ||
|
|
||
| @Inject | ||
| private PublishedEventInserter publishedEventInserter; | ||
|
|
||
| @Inject | ||
| private BatchProcessingDetailsCalculator batchProcessingDetailsCalculator; | ||
|
|
||
| @Inject | ||
| private Logger logger; | ||
|
|
||
| @Transactional(REQUIRED) | ||
| public BatchProcessDetails processNextBatchOfEvents( | ||
| final BatchProcessDetails batchProcessDetails, | ||
| final Set<UUID> activeStreamIds) { | ||
|
|
||
| final AtomicLong currentEventNumber = batchProcessDetails.getCurrentEventNumber(); | ||
| final AtomicLong previousEventNumber = batchProcessDetails.getPreviousEventNumber(); | ||
|
|
||
| final List<PublishedEvent> publishedEvents = new ArrayList<>(); | ||
| try (final Stream<Event> eventStream = eventJdbcRepository.findAllFromEventNumberUptoPageSize(currentEventNumber.get(), PAGE_SIZE)) { | ||
|
|
||
| eventStream.forEach(event -> publishedEventInserter | ||
| .convertAndSave(event, previousEventNumber, activeStreamIds) | ||
| .ifPresent(publishedEvents::add)); | ||
|
|
||
| } | ||
|
|
||
| final BatchProcessDetails currentBatchProcessDetails = batchProcessingDetailsCalculator.calculateNextBatchProcessDetails( | ||
| batchProcessDetails, | ||
| previousEventNumber, | ||
| publishedEvents); | ||
|
|
||
| final int processCount = currentBatchProcessDetails.getProcessCount(); | ||
| logger.info(format("Inserted %d PublishedEvents", processCount)); | ||
|
|
||
| return currentBatchProcessDetails; | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.