*/
public class InMemoryJournal extends AsyncWriteJournal {
+ private static class WriteMessagesComplete {
+ final CountDownLatch latch;
+ final Class<?> ofType;
+
+ public WriteMessagesComplete(int count, Class<?> ofType) {
+ this.latch = new CountDownLatch(count);
+ this.ofType = ofType;
+ }
+ }
+
static final Logger LOG = LoggerFactory.getLogger(InMemoryJournal.class);
private static final Map<String, Map<Long, Object>> journals = new ConcurrentHashMap<>();
private static final Map<String, CountDownLatch> deleteMessagesCompleteLatches = new ConcurrentHashMap<>();
- private static final Map<String, CountDownLatch> writeMessagesCompleteLatches = new ConcurrentHashMap<>();
+ private static final Map<String, WriteMessagesComplete> writeMessagesComplete = new ConcurrentHashMap<>();
private static final Map<String, CountDownLatch> blockReadMessagesLatches = new ConcurrentHashMap<>();
}
public static void waitForWriteMessagesComplete(String persistenceId) {
- if(!Uninterruptibles.awaitUninterruptibly(writeMessagesCompleteLatches.get(persistenceId), 5, TimeUnit.SECONDS)) {
+ if(!Uninterruptibles.awaitUninterruptibly(writeMessagesComplete.get(persistenceId).latch, 5, TimeUnit.SECONDS)) {
throw new AssertionError("Journal write messages did not complete");
}
}
}
public static void addWriteMessagesCompleteLatch(String persistenceId, int count) {
- writeMessagesCompleteLatches.put(persistenceId, new CountDownLatch(count));
+ writeMessagesComplete.put(persistenceId, new WriteMessagesComplete(count, null));
+ }
+
+ public static void addWriteMessagesCompleteLatch(String persistenceId, int count, Class<?> ofType) {
+ writeMessagesComplete.put(persistenceId, new WriteMessagesComplete(count, ofType));
}
public static void addBlockReadMessagesLatch(String persistenceId, CountDownLatch latch) {
journal.put(repr.sequenceNr(), repr.payload());
}
- CountDownLatch latch = writeMessagesCompleteLatches.get(repr.persistenceId());
- if(latch != null) {
- latch.countDown();
+ WriteMessagesComplete complete = writeMessagesComplete.get(repr.persistenceId());
+ if(complete != null) {
+ if(complete.ofType == null || complete.ofType.equals(repr.payload().getClass())) {
+ complete.latch.countDown();
+ }
}
}
@Override
public Future<Void> doAsyncDeleteMessagesTo(String persistenceId, long toSequenceNr, boolean permanent) {
+ LOG.trace("doAsyncDeleteMessagesTo: {}", toSequenceNr);
Map<Long, Object> journal = journals.get(persistenceId);
if(journal != null) {
synchronized (journal) {