Extend awaitEvent to waitForNext event instance (#1932)

mltheuser · ku76uh · web-flow · commit 004d5feb6cdb · 2025-12-15T11:08:21.000+01:00
* Extend awaitEvent to waitForNext event instance

* Updated documentation.

---------

Co-authored-by: ku76uh &lt;malte.heuser@ing.com&gt;
diff --git a/core/akka-runtime/src/main/protobuf/process_index.proto b/core/akka-runtime/src/main/protobuf/process_index.proto
@@ -204,6 +204,7 @@ message AwaitCompleted {
 message AwaitEvent {
   optional string recipeInstanceId = 1;
   optional string eventName = 2;
+  optional bool waitForNext = 3;
 }
 
 message ProcessSensoryEvent {
diff --git a/core/akka-runtime/src/main/protobuf/process_instance.proto b/core/akka-runtime/src/main/protobuf/process_instance.proto
@@ -283,6 +283,7 @@ message Completed {
 
 message AwaitEvent {
   optional string eventName = 1;
+  optional bool waitForNext = 2;
 }
 
 message EventOccurred {
diff --git a/core/akka-runtime/src/main/scala/com/ing/baker/runtime/akka/AkkaBaker.scala b/core/akka-runtime/src/main/scala/com/ing/baker/runtime/akka/AkkaBaker.scala
@@ -608,9 +608,9 @@ class AkkaBaker private[runtime](config: AkkaBakerConfig) extends scaladsl.Baker
     }
   }
 
-  override def awaitEvent(recipeInstanceId: String, eventName: String, timeout: FiniteDuration): Future[Unit] = {
+  override def awaitEvent(recipeInstanceId: String, eventName: String, timeout: FiniteDuration, waitForNext: Boolean = false): Future[Unit] = {
     processIndexActor
-      .ask(ProcessIndexProtocol.AwaitEvent(recipeInstanceId, eventName))(timeout)
+      .ask(ProcessIndexProtocol.AwaitEvent(recipeInstanceId, eventName, waitForNext))(timeout)
       .javaTimeoutToBakerTimeout("awaitEvent")
       .flatMap {
         case EventOccurred => Future.successful()
diff --git a/core/akka-runtime/src/main/scala/com/ing/baker/runtime/akka/actor/process_index/ProcessIndex.scala b/core/akka-runtime/src/main/scala/com/ing/baker/runtime/akka/actor/process_index/ProcessIndex.scala
@@ -634,45 +634,26 @@ class ProcessIndex(recipeInstanceIdleTimeout: Option[FiniteDuration],
         actorRef.forward(ProcessInstanceProtocol.AwaitCompleted)
       }
 
-    case ProcessIndexProtocol.AwaitEvent(recipeInstanceId, eventName) =>
+    case ProcessIndexProtocol.AwaitEvent(recipeInstanceId, eventName, waitForNext) =>
       val originalSender = sender()
 
-      // This program describes the logic as a sequence of validation steps.
-      // If any step fails (returns a Left), the entire program short-circuits
-      // and the failure is handled in unsafeRunAsync.
       val program: FireEventIO[Unit] = for {
-        // Fetch the process actor and its metadata. This will activate the actor if it's passivated.
         instanceAndMeta <- fetchInstance(recipeInstanceId)
         (processInstance, metadata) = instanceAndMeta
-
-        // Fetch the compiled recipe from the cache or manager.
         recipe <- fetchRecipe(metadata)
-
-        // Validate that the event name corresponds to a known sensory event in the recipe.
-        // We don't need the result, we just care that it succeeds.
         _ <- validateAnyEventNameIsInRecipe(recipe, eventName, recipeInstanceId)
-
-        // If all validations pass, forward the AwaitEvent message to the ProcessInstance actor.
-        // `forward` preserves the original sender, so the ProcessInstance actor's reply will go
-        // to the actor that originally sent the AwaitEvent message.
-        _ <- accept(processInstance.tell(ProcessInstanceProtocol.AwaitEvent(eventName), originalSender))
+        _ <- accept(processInstance.tell(ProcessInstanceProtocol.AwaitEvent(eventName, waitForNext), originalSender))
       } yield ()
 
-      // Asynchronously execute the program and handle the result.
       program.value.unsafeRunAsync{
         case Left(exception) =>
-          // This case represents an unexpected technical error within the IO program.
           log.error(exception, s"Unexpected error processing AwaitEvent for recipe instance '$recipeInstanceId'")
           originalSender ! akka.actor.Status.Failure(exception)
 
         case Right(Left(rejection)) =>
-          // This case represents a controlled, business-logic failure (e.g., event name not found, process doesn't exist).
-          // We send the rejection object directly back to the sender.
           originalSender ! rejection
 
         case Right(Right(())) =>
-          // The command was successfully validated and forwarded to the ProcessInstance actor.
-          // That actor is now responsible for handling the await logic. No further action is needed here.
           ()
       }(IORuntime.global)
 
diff --git a/core/akka-runtime/src/main/scala/com/ing/baker/runtime/akka/actor/process_index/ProcessIndexProto.scala b/core/akka-runtime/src/main/scala/com/ing/baker/runtime/akka/actor/process_index/ProcessIndexProto.scala
@@ -641,13 +641,14 @@ object ProcessIndexProto {
       val companion = protobuf.AwaitEvent
 
       def toProto(a: ProcessIndexProtocol.AwaitEvent): protobuf.AwaitEvent =
-        protobuf.AwaitEvent(Some(a.recipeInstanceId), Some(a.eventName))
+        protobuf.AwaitEvent(Some(a.recipeInstanceId), Some(a.eventName), Some(a.waitForNext))
 
       def fromProto(message: protobuf.AwaitEvent): Try[ProcessIndexProtocol.AwaitEvent] =
         for {
           recipeInstanceId <- versioned(message.recipeInstanceId, "recipeInstanceId")
           eventName <- versioned(message.eventName, "eventName")
-        } yield ProcessIndexProtocol.AwaitEvent(recipeInstanceId, eventName)
+          waitForNext = message.waitForNext.getOrElse(false)
+        } yield ProcessIndexProtocol.AwaitEvent(recipeInstanceId, eventName, waitForNext)
     }
 
 }
diff --git a/core/akka-runtime/src/main/scala/com/ing/baker/runtime/akka/actor/process_index/ProcessIndexProtocol.scala b/core/akka-runtime/src/main/scala/com/ing/baker/runtime/akka/actor/process_index/ProcessIndexProtocol.scala
@@ -40,7 +40,7 @@ object ProcessIndexProtocol {
 
   case class AwaitCompleted(recipeInstanceId: String) extends ProcessIndexMessage
 
-  case class AwaitEvent(recipeInstanceId: String, eventName: String) extends ProcessIndexMessage
+  case class AwaitEvent(recipeInstanceId: String, eventName: String, waitForNext: Boolean = false) extends ProcessIndexMessage
 
   /**
     * Failure when attempting to resolve a blocked interaction, the event is not of valid type according with the recipe
diff --git a/core/akka-runtime/src/main/scala/com/ing/baker/runtime/akka/actor/process_instance/ProcessInstance.scala b/core/akka-runtime/src/main/scala/com/ing/baker/runtime/akka/actor/process_instance/ProcessInstance.scala
@@ -334,16 +334,13 @@ class ProcessInstance(
         }
       }
 
-    case ProcessInstanceProtocol.AwaitEvent(eventName) =>
-      if (instance.state.eventNames.contains(eventName)) {
+    case ProcessInstanceProtocol.AwaitEvent(eventName, waitForNext) =>
+      if (!waitForNext && instance.state.eventNames.contains(eventName)) {
         sender() ! ProcessInstanceProtocol.EventOccurred
       } else {
-        // Persist the listener addition
         val event = EventListenerAdded(eventName, sender().path.toSerializationFormat)
-        // This persist is sync so no race conditions to worry about
         persistEvent(instance, event) { _ =>
           val updatedInstance = eventSource.apply(instance)(event)
-          // Update actor's state
           context.become(running(updatedInstance, scheduledRetries))
         }
       }
diff --git a/core/akka-runtime/src/main/scala/com/ing/baker/runtime/akka/actor/process_instance/ProcessInstanceProto.scala b/core/akka-runtime/src/main/scala/com/ing/baker/runtime/akka/actor/process_instance/ProcessInstanceProto.scala
@@ -372,12 +372,13 @@ object ProcessInstanceProto {
       val companion = protobuf.AwaitEvent
 
       def toProto(a: ProcessInstanceProtocol.AwaitEvent): protobuf.AwaitEvent =
-        protobuf.AwaitEvent(Some(a.eventName))
+        protobuf.AwaitEvent(Some(a.eventName), Some(a.waitForNext))
 
       def fromProto(message: protobuf.AwaitEvent): Try[ProcessInstanceProtocol.AwaitEvent] =
         for {
           eventName <- versioned(message.eventName, "eventName")
-        } yield ProcessInstanceProtocol.AwaitEvent(eventName)
+          waitForNext = message.waitForNext.getOrElse(false)
+        } yield ProcessInstanceProtocol.AwaitEvent(eventName, waitForNext)
     }
 
   implicit def processInstanceEventOccurredProto: ProtoMap[ProcessInstanceProtocol.EventOccurred.type, protobuf.EventOccurred] =
diff --git a/core/akka-runtime/src/main/scala/com/ing/baker/runtime/akka/actor/process_instance/ProcessInstanceProtocol.scala b/core/akka-runtime/src/main/scala/com/ing/baker/runtime/akka/actor/process_instance/ProcessInstanceProtocol.scala
@@ -29,7 +29,7 @@ object ProcessInstanceProtocol {
    */
   case object Completed extends BakerSerializable
 
-  case class AwaitEvent(eventName: String) extends Command
+  case class AwaitEvent(eventName: String, waitForNext: Boolean = false) extends Command
 
   /**
    * Response confirming the event has occurred.
diff --git a/core/baker-interface-kotlin/src/main/kotlin/com/ing/baker/runtime/kotlindsl/Baker.kt b/core/baker-interface-kotlin/src/main/kotlin/com/ing/baker/runtime/kotlindsl/Baker.kt
@@ -216,8 +216,9 @@ class Baker internal constructor(private val jBaker: Baker) : AutoCloseable {
         recipeInstanceId: String,
         eventName: String,
         timeout: Duration,
-    ): Unit {
-        jBaker.awaitEvent(recipeInstanceId, eventName, timeout.toJavaDuration()).await()
+        waitForNext: Boolean = false
+    ) {
+        jBaker.awaitEvent(recipeInstanceId, eventName, timeout.toJavaDuration(), waitForNext).await()
     }
 
     suspend fun awaitCompleted(
diff --git a/core/baker-interface/src/main/scala/com/ing/baker/runtime/common/Baker.scala b/core/baker-interface/src/main/scala/com/ing/baker/runtime/common/Baker.scala
@@ -161,7 +161,7 @@ trait Baker[F[_]] extends LanguageApi {
    * @throws ProcessDeletedException if the recipe instance has been deleted.
    * @throws TimeoutException        if the event does not occur within the specified `timeout`.
    */
-  def awaitEvent(recipeInstanceId: String, eventName: String, timeout: DurationType): F[Unit]
+  def awaitEvent(recipeInstanceId: String, eventName: String, timeout: DurationType, waitForNext: Boolean = false): F[Unit]
 
   /**
    * Waits until a recipe instance completes execution.
diff --git a/core/baker-interface/src/main/scala/com/ing/baker/runtime/javadsl/Baker.scala b/core/baker-interface/src/main/scala/com/ing/baker/runtime/javadsl/Baker.scala
@@ -471,8 +471,14 @@ class Baker(private val baker: scaladsl.Baker) extends common.Baker[CompletableF
   override def fireSensoryEventAndAwaitReceived(recipeInstanceId: String, event: EventInstance, correlationId: Optional[String]): CompletableFuture[SensoryEventStatus] =
     toCompletableFuture(baker.fireSensoryEventAndAwaitReceived(recipeInstanceId, event.asScala, Option.apply(correlationId.orElse(null))))
 
-  override def awaitEvent(recipeInstanceId: String, eventName: String, timeout: Duration): CompletableFuture[Unit] =
-    toCompletableFuture(baker.awaitEvent(recipeInstanceId, eventName, FiniteDuration.apply(timeout.toMillis, TimeUnit.MILLISECONDS)))
+  /**
+    * Convenience overload for Java callers that don't want to specify waitForNext.
+    */
+  def awaitEvent(recipeInstanceId: String, eventName: String, timeout: Duration): CompletableFuture[Unit] =
+    awaitEvent(recipeInstanceId, eventName, timeout, false)
+
+  override def awaitEvent(recipeInstanceId: String, eventName: String, timeout: Duration, waitForNext: Boolean): CompletableFuture[Unit] =
+    toCompletableFuture(baker.awaitEvent(recipeInstanceId, eventName, FiniteDuration.apply(timeout.toMillis, TimeUnit.MILLISECONDS), waitForNext))
 
   override def awaitCompleted(recipeInstanceId: String, timeout: Duration): CompletableFuture[SensoryEventStatus] =
     toCompletableFuture(baker.awaitCompleted(recipeInstanceId, FiniteDuration.apply(timeout.toMillis, TimeUnit.MILLISECONDS)))
diff --git a/core/baker-interface/src/main/scala/com/ing/baker/runtime/model/BakerF.scala b/core/baker-interface/src/main/scala/com/ing/baker/runtime/model/BakerF.scala
@@ -315,8 +315,8 @@ abstract class BakerF[F[_]](implicit components: BakerComponents[F], sync: Sync[
     components.recipeInstanceManager.awaitCompleted(recipeInstanceId, timeout)
       .recoverWith(javaTimeoutToBakerTimeout(s"awaitCompleted for recipe instance '$recipeInstanceId'"))
 
-  override def awaitEvent(recipeInstanceId: String, eventName: String, timeout: FiniteDuration): F[Unit] =
-    components.recipeInstanceManager.awaitEvent(recipeInstanceId, eventName, timeout)
+  override def awaitEvent(recipeInstanceId: String, eventName: String, timeout: FiniteDuration, waitForNext: Boolean = false): F[Unit] =
+    components.recipeInstanceManager.awaitEvent(recipeInstanceId, eventName, timeout, waitForNext)
       .recoverWith(javaTimeoutToBakerTimeout(s"awaitEvent for recipe instance '$recipeInstanceId'"))
 
   override def addMetaData(recipeInstanceId: String, metadata: Map[String, String]): F[Unit] =
@@ -546,8 +546,8 @@ abstract class BakerF[F[_]](implicit components: BakerComponents[F], sync: Sync[
         mapK(self.stopRetryingInteraction(recipeInstanceId, interactionName))
       override def fireSensoryEventAndAwaitReceived(recipeInstanceId: String, event: EventInstance, correlationId: Option[String]): G[SensoryEventStatus] =
         mapK(self.fireSensoryEventAndAwaitReceived(recipeInstanceId, event, correlationId))
-      override def awaitEvent(recipeInstanceId: String, eventName: String, timeout: FiniteDuration): G[Unit] =
-        mapK(self.awaitEvent(recipeInstanceId, eventName, timeout))
+      override def awaitEvent(recipeInstanceId: String, eventName: String, timeout: FiniteDuration, waitForNext: Boolean = false): G[Unit] =
+        mapK(self.awaitEvent(recipeInstanceId, eventName, timeout, waitForNext))
       override def awaitCompleted(recipeInstanceId: String, timeout: FiniteDuration): G[SensoryEventStatus] =
         mapK(self.awaitCompleted(recipeInstanceId, timeout))
     }
@@ -618,8 +618,8 @@ abstract class BakerF[F[_]](implicit components: BakerComponents[F], sync: Sync[
         mapK(self.addMetaData(recipeInstanceId, metadata))
       override def fireSensoryEventAndAwaitReceived(recipeInstanceId: String, event: EventInstance, correlationId: Option[String]): Future[SensoryEventStatus] =
         mapK(self.fireSensoryEventAndAwaitReceived(recipeInstanceId, event, correlationId))
-      override def awaitEvent(recipeInstanceId: String, eventName: String, timeout: FiniteDuration): Future[Unit] =
-        mapK(self.awaitEvent(recipeInstanceId, eventName, timeout))
+      override def awaitEvent(recipeInstanceId: String, eventName: String, timeout: FiniteDuration, waitForNext: Boolean = false): Future[Unit] =
+        mapK(self.awaitEvent(recipeInstanceId, eventName, timeout, waitForNext))
       override def awaitCompleted(recipeInstanceId: String, timeout: FiniteDuration): Future[SensoryEventStatus] =
         mapK(self.awaitCompleted(recipeInstanceId, timeout))
     }
diff --git a/core/baker-interface/src/main/scala/com/ing/baker/runtime/model/RecipeInstanceManager.scala b/core/baker-interface/src/main/scala/com/ing/baker/runtime/model/RecipeInstanceManager.scala
@@ -177,15 +177,15 @@ trait RecipeInstanceManager[F[_]] {
     }).flatten
   }
 
-  def awaitEvent(recipeInstanceId: String, eventName: String, timeout: FiniteDuration)(implicit async: Async[F]): F[Unit] =
+  def awaitEvent(recipeInstanceId: String, eventName: String, timeout: FiniteDuration, waitForNext: Boolean = false)(implicit async: Async[F]): F[Unit] =
     getExistent(recipeInstanceId).flatMap { recipeInstance =>
       Deferred[F, Unit].flatMap { listener =>
         recipeInstance.state.modify { currentState =>
-          // Check if the event has already occurred
-          if (currentState.events.exists(_.name == eventName)) {
+          // If waitForNext is false, check if the event has already occurred
+          if (!waitForNext && currentState.events.exists(_.name == eventName)) {
             (currentState, async.unit) // Already happened, resolve immediately
           } else {
-            // Not yet happened, add listener and wait on it
+            // Not yet happened (or waitForNext=true), add listener and wait on it
             (currentState.addEventListener(eventName, listener), listener.get)
           }
         }.flatten.timeoutTo(timeout, async.raiseError(new TimeoutException(s"Timed out after $timeout waiting for event '$eventName' in instance '$recipeInstanceId'")))
diff --git a/docs/sections/cookbook/fire-sensory-events.md b/docs/sections/cookbook/fire-sensory-events.md
@@ -71,8 +71,14 @@ Make sure to set a timeout smaller than the requestTimeout of your httpClient. O
 
 `awaitEvent` completes when an event with the specified name appears in the list of fired events.
 
+The method accepts an optional `waitForNext` parameter (default: `false`):
+
+*   `waitForNext = false`: Baker checks the event history first. If an event with the specified name has **already occurred**, the future completes immediately.
+    This implies that if you call `awaitEvent` multiple times for the same event name, all calls after the first occurrence will resolve immediately.
+*   `waitForNext = true`: Baker ignores the event history and registers a listener for the **next** occurrence of the event.
+
 !!!Note
-Make sure to set a timeout smaller than the requestTimeout of your httpClient. Otherwise, you will see awaitEvent() timeout prematurely.
+    Make sure to set a timeout smaller than the requestTimeout of your httpClient. Otherwise, you will see awaitEvent() timeout prematurely.
 
 === "Java"
 
@@ -92,6 +98,33 @@ Make sure to set a timeout smaller than the requestTimeout of your httpClient. O
     --8<-- "docs-code-snippets/src/main/scala/examples/scala/application/AwaitEvent.scala"
     ```
 
+#### Waiting for new events (waitForNext=true)
+
+When using `waitForNext=true`, rely on the following pattern to avoid race conditions.
+
+!!! Warning "Race Condition"
+    If you fire a sensory event *first* and then call `awaitEvent(..., waitForNext=true)`, the expected event might have already occurred by the time the listener is registered. This will cause `awaitEvent` to run until timeout, even though the event actually happened.
+
+    **Solution:** Start `awaitEvent` **before** firing the sensory event.
+
+=== "Java"
+
+    ```java
+    --8<-- "docs-code-snippets/src/main/java/examples/java/application/AwaitEventWaitForNext.java"
+    ```
+
+=== "Kotlin"
+
+    ```kotlin
+    --8<-- "docs-code-snippets/src/main/kotlin/examples/kotlin/application/AwaitEventWaitForNext.kt"
+    ```
+
+=== "Scala"
+
+    ```scala
+    --8<-- "docs-code-snippets/src/main/scala/examples/scala/application/AwaitEventWaitForNext.scala"
+    ```
+
 ## Inquiry
 Baker allows you to query the state of a recipe instance at any given moment. To this end, Baker exposes a couple
 of different methods that allow you to fetch information about events, ingredients, and interactions from a running process.
diff --git a/examples/docs-code-snippets/src/main/java/examples/java/application/AwaitEventWaitForNext.java b/examples/docs-code-snippets/src/main/java/examples/java/application/AwaitEventWaitForNext.java
@@ -0,0 +1,31 @@
+package examples.java.application;
+
+import com.ing.baker.runtime.javadsl.Baker;
+import com.ing.baker.runtime.javadsl.EventInstance;
+import examples.java.events.OrderPlaced;
+
+import java.time.Duration;
+
+public class AwaitEventWaitForNext {
+
+    private final Baker baker;
+
+    public AwaitEventWaitForNext(Baker baker) {
+        this.baker = baker;
+    }
+
+    public void example(String recipeInstanceId, OrderPlaced orderPlaced) {
+        var eventInstance = EventInstance.from(orderPlaced);
+
+        // Pattern: Start waiting BEFORE firing the sensory event
+        // waitForNext = true checks for *new* occurrences of the event
+        var awaitFuture = baker.awaitEvent(recipeInstanceId, "ExpectedEvent", Duration.ofSeconds(5), true);
+        
+        // Fire the event
+        baker.fireSensoryEventAndAwaitReceived(recipeInstanceId, eventInstance).join();
+        
+        // Now wait for the event to occur
+        // Because we started listening *before* firing, we are guaranteed not to miss it
+        awaitFuture.join(); 
+    }
+}
diff --git a/examples/docs-code-snippets/src/main/kotlin/examples/kotlin/application/AwaitEventWaitForNext.kt b/examples/docs-code-snippets/src/main/kotlin/examples/kotlin/application/AwaitEventWaitForNext.kt
@@ -0,0 +1,28 @@
+package examples.kotlin.application
+
+import com.ing.baker.runtime.javadsl.EventInstance
+import com.ing.baker.runtime.kotlindsl.Baker
+import examples.kotlin.events.OrderPlaced
+import kotlinx.coroutines.async
+import kotlinx.coroutines.coroutineScope
+import kotlin.time.Duration.Companion.seconds
+
+class AwaitEventWaitForNext(private val baker: Baker) {
+
+    suspend fun example(recipeInstanceId: String, orderPlaced: OrderPlaced) = coroutineScope {
+        val orderPlacedEvent = EventInstance.from(orderPlaced)
+        
+        // Pattern: Start waiting BEFORE firing the sensory event
+        // waitForNext = true checks for *new* occurrences of the event
+        // We use async to start the "await" concurrently 
+        val awaitDeferred = async { 
+            baker.awaitEvent(recipeInstanceId, "ExpectedEvent", timeout = 5.seconds, waitForNext = true) 
+        }
+        
+        // Fire the event
+        baker.fireSensoryEventAndAwaitReceived(recipeInstanceId, orderPlacedEvent)
+        
+        // Now wait for the event to occur
+        awaitDeferred.await()
+    }
+}
diff --git a/examples/docs-code-snippets/src/main/scala/examples/scala/application/AwaitEventWaitForNext.scala b/examples/docs-code-snippets/src/main/scala/examples/scala/application/AwaitEventWaitForNext.scala
@@ -0,0 +1,27 @@
+package examples.scala.application
+
+import com.ing.baker.runtime.scaladsl.{Baker, EventInstance}
+import examples.scala.events.OrderPlaced
+
+import scala.concurrent.ExecutionContext.Implicits.global
+import scala.concurrent.duration.DurationInt
+
+class AwaitEventWaitForNext(val baker: Baker) {
+
+  def example(recipeInstanceId: String, orderPlaced: OrderPlaced): Unit = {
+    val eventInstance = EventInstance.unsafeFrom(orderPlaced)
+    
+    // Pattern: Start waiting BEFORE firing the sensory event
+    // waitForNext = true checks for *new* occurrences of the event
+    val awaitFuture = baker.awaitEvent(recipeInstanceId, "ExpectedEvent", timeout = 5.seconds, waitForNext = true)
+    
+    // Fire the event
+    // We combine the futures to ensure we wait for both:
+    // 1. The event to be received
+    // 2. The expected event to occur
+    val result = for {
+      _ <- baker.fireSensoryEventAndAwaitReceived(recipeInstanceId, eventInstance)
+      _ <- awaitFuture 
+    } yield ()
+  }
+}
diff --git a/http/baker-http-client/src/main/scala/com/ing/baker/http/client/scaladsl/BakerClient.scala b/http/baker-http-client/src/main/scala/com/ing/baker/http/client/scaladsl/BakerClient.scala
@@ -442,10 +442,11 @@ final class BakerClient( client: Client[IO],
       result
     }
 
-  override def awaitEvent(recipeInstanceId: String, eventName: String, timeout: FiniteDuration): Future[Unit] =
+  override def awaitEvent(recipeInstanceId: String, eventName: String, timeout: FiniteDuration, waitForNext: Boolean = false): Future[Unit] =
     callRemoteBakerServiceFallbackAware[Unit](
       (host, prefix) => IO(POST((root(host, prefix) / "instances" / recipeInstanceId / "await-event" / eventName)
-        .withQueryParam("timeout", timeout.toMillis))), fallbackEndpoint).map { _ =>
+        .withQueryParam("timeout", timeout.toMillis)
+        .withQueryParam("waitForNext", waitForNext))), fallbackEndpoint).map { _ =>
       logger.info(s"For recipe instance '$recipeInstanceId', awaited event '$eventName'")
     }
 
diff --git a/http/baker-http-server/src/main/scala/com/ing/baker/http/server/javadsl/BakerWithHttpResponse.scala b/http/baker-http-server/src/main/scala/com/ing/baker/http/server/javadsl/BakerWithHttpResponse.scala

Original file line number	Diff line number	Diff line change
`@@ -204,6 +204,7 @@ message AwaitCompleted {`
`204`	`204`	`message AwaitEvent {`
`205`	`205`	`optional string recipeInstanceId = 1;`
`206`	`206`	`optional string eventName = 2;`
	`207`	`+ optional bool waitForNext = 3;`
`207`	`208`	`}`
`208`	`209`
`209`	`210`	`message ProcessSensoryEvent {`
Original file line number	Diff line number	Diff line change
`@@ -283,6 +283,7 @@ message Completed {`
`283`	`283`
`284`	`284`	`message AwaitEvent {`
`285`	`285`	`optional string eventName = 1;`
	`286`	`+ optional bool waitForNext = 2;`
`286`	`287`	`}`
`287`	`288`
`288`	`289`	`message EventOccurred {`
Original file line number	Diff line number	Diff line change
`@@ -608,9 +608,9 @@ class AkkaBaker private[runtime](config: AkkaBakerConfig) extends scaladsl.Baker`
`608`	`608`	`}`
`609`	`609`	`}`
`610`	`610`
`611`		`- override def awaitEvent(recipeInstanceId: String, eventName: String, timeout: FiniteDuration): Future[Unit] = {`
	`611`	`+ override def awaitEvent(recipeInstanceId: String, eventName: String, timeout: FiniteDuration, waitForNext: Boolean = false): Future[Unit] = {`
`612`	`612`	`processIndexActor`
`613`		`- .ask(ProcessIndexProtocol.AwaitEvent(recipeInstanceId, eventName))(timeout)`
	`613`	`+ .ask(ProcessIndexProtocol.AwaitEvent(recipeInstanceId, eventName, waitForNext))(timeout)`
`614`	`614`	`.javaTimeoutToBakerTimeout("awaitEvent")`
`615`	`615`	`.flatMap {`
`616`	`616`	`case EventOccurred => Future.successful()`
Original file line number	Diff line number	Diff line change
`@@ -641,13 +641,14 @@ object ProcessIndexProto {`
`641`	`641`	`val companion = protobuf.AwaitEvent`
`642`	`642`
`643`	`643`	`def toProto(a: ProcessIndexProtocol.AwaitEvent): protobuf.AwaitEvent =`
`644`		`- protobuf.AwaitEvent(Some(a.recipeInstanceId), Some(a.eventName))`
	`644`	`+ protobuf.AwaitEvent(Some(a.recipeInstanceId), Some(a.eventName), Some(a.waitForNext))`
`645`	`645`
`646`	`646`	`def fromProto(message: protobuf.AwaitEvent): Try[ProcessIndexProtocol.AwaitEvent] =`
`647`	`647`	`for {`
`648`	`648`	`recipeInstanceId <- versioned(message.recipeInstanceId, "recipeInstanceId")`
`649`	`649`	`eventName <- versioned(message.eventName, "eventName")`
`650`		`- } yield ProcessIndexProtocol.AwaitEvent(recipeInstanceId, eventName)`
	`650`	`+ waitForNext = message.waitForNext.getOrElse(false)`
	`651`	`+ } yield ProcessIndexProtocol.AwaitEvent(recipeInstanceId, eventName, waitForNext)`
`651`	`652`	`}`
`652`	`653`
`653`	`654`	`}`
Original file line number	Diff line number	Diff line change
`@@ -334,16 +334,13 @@ class ProcessInstance(`
`334`	`334`	`}`
`335`	`335`	`}`
`336`	`336`
`337`		`- case ProcessInstanceProtocol.AwaitEvent(eventName) =>`
`338`		`- if (instance.state.eventNames.contains(eventName)) {`
	`337`	`+ case ProcessInstanceProtocol.AwaitEvent(eventName, waitForNext) =>`
	`338`	`+ if (!waitForNext && instance.state.eventNames.contains(eventName)) {`
`339`	`339`	`sender() ! ProcessInstanceProtocol.EventOccurred`
`340`	`340`	`} else {`
`341`		`- // Persist the listener addition`
`342`	`341`	`val event = EventListenerAdded(eventName, sender().path.toSerializationFormat)`
`343`		`- // This persist is sync so no race conditions to worry about`
`344`	`342`	`persistEvent(instance, event) { _ =>`
`345`	`343`	`val updatedInstance = eventSource.apply(instance)(event)`
`346`		`- // Update actor's state`
`347`	`344`	`context.become(running(updatedInstance, scheduledRetries))`
`348`	`345`	`}`
`349`	`346`	`}`