Re-enable IdleStateMonitor

src/main/java/io/camunda/zeebe/bpmnassert/testengine/EngineFactory.java

@@ -129,7 +129,7 @@ private static StreamProcessor createStreamProcessor(
         .streamProcessorFactory(
             context ->
                 EngineProcessors.createEngineProcessors(
-                    context,
+                    context.listener(idleStateMonitor),
                     partitionCount,
                     subscriptionCommandSenderFactory.createSender(),
                     new SinglePartitionDeploymentDistributor(),

src/main/java/io/camunda/zeebe/bpmnassert/testengine/IdleStateMonitor.java

@@ -7,13 +7,19 @@
 import io.camunda.zeebe.logstreams.storage.LogStorage;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Timer;
+import java.util.TimerTask;
 
 final class IdleStateMonitor implements LogStorage.CommitListener, StreamProcessorListener {
 
+  private static final Timer TIMER = new Timer();
   private final List<Runnable> callbacks = new ArrayList<>();
   private final LogStreamReader reader;
-  private long lastEventPosition = -1L;
-  private long lastProcessedPosition = -1;
+  private volatile long lastEventPosition = -1L;
+  private volatile long lastProcessedPosition = -1L;
+
+  private volatile TimerTask task =
+      createTask(); // must never be null for the synchronization to work
 
   IdleStateMonitor(final InMemoryLogStorage logStorage, final LogStreamReader logStreamReader) {
     logStorage.addCommitListener(this);
@@ -30,9 +36,18 @@ public void addCallback(final Runnable callback) {
 
   private void checkIdleState() {
     if (isInIdleState()) {
-      synchronized (callbacks) {
-        callbacks.forEach(Runnable::run);
-        callbacks.clear();
+      synchronized (task) {
+        task = createTask();
+        try {
+          TIMER.schedule(task, 10);
+        } catch (IllegalStateException e) {
+          // thrown - among others - if task was cancelled before it could be scheduled
+          // do nothing in this case
+        }
+      }
+    } else {
+      synchronized (task) {
+        task.cancel();
       }
     }
   }
@@ -43,8 +58,10 @@ private boolean isInIdleState() {
   }
 
   private void forwardToLastEvent() {
-    while (reader.hasNext()) {
-      lastEventPosition = reader.next().getPosition();
+    synchronized (reader) {
+      while (reader.hasNext()) {
+        lastEventPosition = reader.next().getPosition();
+      }
     }
   }
 
@@ -55,19 +72,33 @@ public void onCommit() {
 
   @Override
   public void onProcessed(final TypedRecord<?> processedCommand) {
-    lastProcessedPosition = processedCommand.getPosition();
+    lastProcessedPosition = Math.max(lastProcessedPosition, processedCommand.getPosition());
     checkIdleState();
   }
 
   @Override
   public void onSkipped(final LoggedEvent skippedRecord) {
-    lastProcessedPosition = skippedRecord.getPosition();
+    lastProcessedPosition = Math.max(lastProcessedPosition, skippedRecord.getPosition());
     checkIdleState();
   }
 
   @Override
   public void onReplayed(final long lastReplayedEventPosition, final long lastReadRecordPosition) {
-    lastProcessedPosition = lastReplayedEventPosition;
+    lastProcessedPosition = Math.max(lastProcessedPosition, lastReplayedEventPosition);
     checkIdleState();
   }
+
+  private TimerTask createTask() {
+    return new TimerTask() {
+      @Override
+      public void run() {
+        if (isInIdleState()) {
+          synchronized (callbacks) {
+            callbacks.forEach(Runnable::run);
+            callbacks.clear();
+          }
+        }
+      }
+    };
+  }
 }

src/main/java/io/camunda/zeebe/bpmnassert/testengine/InMemoryEngineImpl.java

@@ -10,7 +10,7 @@
 import io.grpc.Server;
 import java.io.IOException;
 import java.time.Duration;
-import org.apache.commons.lang3.NotImplementedException;
+import java.util.concurrent.CompletableFuture;
 
 public class InMemoryEngineImpl implements InMemoryEngine {
 
@@ -95,17 +95,15 @@ public void increaseTime(final Duration timeToAdd) {
 
   @Override
   public void runOnIdleState(final Runnable callback) {
-    throw new NotImplementedException("This feature is coming soon");
-    //    idleStateMonitor.addCallback(callback);
+    idleStateMonitor.addCallback(callback);
   }
 
   @Override
   public void waitForIdleState() {
-    throw new NotImplementedException("This feature is coming soon");
-    //    final CompletableFuture<Void> idleState = new CompletableFuture<>();
-    //
-    //    runOnIdleState(() -> idleState.complete(null));
-    //
-    //    idleState.join();
+    final CompletableFuture<Void> idleState = new CompletableFuture<>();
+
+    runOnIdleState(() -> idleState.complete(null));
+
+    idleState.join();
   }
 }

src/test/java/io/camunda/zeebe/bpmnassert/assertions/DeploymentAssertTest.java

@@ -4,7 +4,6 @@
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 import io.camunda.zeebe.bpmnassert.extensions.ZeebeProcessTest;
-import io.camunda.zeebe.bpmnassert.testengine.RecordStreamSource;
 import io.camunda.zeebe.bpmnassert.util.Utilities;
 import io.camunda.zeebe.bpmnassert.util.Utilities.ProcessPackLoopingServiceTask;
 import io.camunda.zeebe.bpmnassert.util.Utilities.ProcessPackMultipleTasks;

src/test/java/io/camunda/zeebe/bpmnassert/assertions/MessageAssertTest.java

@@ -11,7 +11,6 @@
 
 import io.camunda.zeebe.bpmnassert.extensions.ZeebeProcessTest;
 import io.camunda.zeebe.bpmnassert.testengine.InMemoryEngine;
-import io.camunda.zeebe.bpmnassert.testengine.RecordStreamSource;
 import io.camunda.zeebe.client.ZeebeClient;
 import io.camunda.zeebe.client.api.response.ProcessInstanceEvent;
 import io.camunda.zeebe.client.api.response.PublishMessageResponse;

src/test/java/io/camunda/zeebe/bpmnassert/assertions/ProcessAssertTest.java

@@ -7,7 +7,6 @@
 
 import io.camunda.zeebe.bpmnassert.extensions.ZeebeProcessTest;
 import io.camunda.zeebe.bpmnassert.testengine.InMemoryEngine;
-import io.camunda.zeebe.bpmnassert.testengine.RecordStreamSource;
 import io.camunda.zeebe.bpmnassert.util.Utilities.ProcessPackLoopingServiceTask;
 import io.camunda.zeebe.client.ZeebeClient;
 import io.camunda.zeebe.client.api.response.DeploymentEvent;

src/test/java/io/camunda/zeebe/bpmnassert/assertions/ProcessInstanceAssertTest.java

@@ -16,7 +16,6 @@
 
 import io.camunda.zeebe.bpmnassert.extensions.ZeebeProcessTest;
 import io.camunda.zeebe.bpmnassert.testengine.InMemoryEngine;
-import io.camunda.zeebe.bpmnassert.testengine.RecordStreamSource;
 import io.camunda.zeebe.client.ZeebeClient;
 import io.camunda.zeebe.client.api.response.ProcessInstanceEvent;
 import io.camunda.zeebe.protocol.record.value.ErrorType;

src/test/java/io/camunda/zeebe/bpmnassert/inspections/ProcessEventInspectionsTest.java

@@ -8,7 +8,6 @@
 import io.camunda.zeebe.bpmnassert.extensions.ZeebeProcessTest;
 import io.camunda.zeebe.bpmnassert.inspections.model.InspectedProcessInstance;
 import io.camunda.zeebe.bpmnassert.testengine.InMemoryEngine;
-import io.camunda.zeebe.bpmnassert.testengine.RecordStreamSource;
 import io.camunda.zeebe.bpmnassert.util.Utilities.ProcessPackTimerStartEvent;
 import io.camunda.zeebe.client.ZeebeClient;
 import io.camunda.zeebe.client.api.response.DeploymentEvent;

src/test/java/io/camunda/zeebe/bpmnassert/inspections/ProcessInstanceInspectionsTest.java

@@ -8,7 +8,6 @@
 import io.camunda.zeebe.bpmnassert.extensions.ZeebeProcessTest;
 import io.camunda.zeebe.bpmnassert.inspections.model.InspectedProcessInstance;
 import io.camunda.zeebe.bpmnassert.testengine.InMemoryEngine;
-import io.camunda.zeebe.bpmnassert.testengine.RecordStreamSource;
 import io.camunda.zeebe.bpmnassert.util.Utilities.ProcessPackCallActivity;
 import io.camunda.zeebe.client.ZeebeClient;
 import io.camunda.zeebe.client.api.response.ProcessInstanceEvent;

src/test/java/io/camunda/zeebe/bpmnassert/util/Utilities.java

@@ -113,14 +113,8 @@ public static ActivateJobsResponse activateSingleJob(
     return client.newActivateJobsCommand().jobType(jobType).maxJobsToActivate(1).send().join();
   }
 
-  // TODO find a better solution for this
   public static void waitForIdleState(final InMemoryEngine engine) {
-    try {
-      Thread.sleep(100);
-    } catch (final InterruptedException e) {
-      e.printStackTrace();
-      throw new IllegalStateException("Sleep was interrupted");
-    }
+    engine.waitForIdleState();
   }
 
   public static PublishMessageResponse sendMessage(
@@ -151,6 +145,12 @@ public static PublishMessageResponse sendMessage(
 
   public static void increaseTime(final InMemoryEngine engine, final Duration duration) {
     engine.increaseTime(duration);
+    try {
+      // we need to wait some physical time so that InMemoryEngine has a chance to fire the timers
+      Thread.sleep(50);
+    } catch (InterruptedException e) {
+      // do nothing
+    }
     waitForIdleState(engine);
   }