diff --git a/.github/workflows/validate.yml b/.github/workflows/validate.yml index cf0dcce8f0..039b70484b 100644 --- a/.github/workflows/validate.yml +++ b/.github/workflows/validate.yml @@ -126,6 +126,7 @@ jobs: (cd examples && mm.py ./src/main/java/io/dapr/examples/workflows/README.md) (cd spring-boot-examples/pubsub && mm.py README.md) (cd spring-boot-examples/workflows/patterns && mm.py README.md) + (cd spring-boot-examples/workflows/event-workflow && mm.py README.md) (cd examples && mm.py ./src/main/java/io/dapr/examples/jobs/README.md) (cd examples && mm.py ./src/main/java/io/dapr/examples/conversation/README.md) (cd examples && mm.py ./src/main/java/io/dapr/examples/invoke/http/README.md) diff --git a/spring-boot-examples/README.md b/spring-boot-examples/README.md index ca03bf9bc8..892bccc3bc 100644 --- a/spring-boot-examples/README.md +++ b/spring-boot-examples/README.md @@ -6,5 +6,6 @@ folder per Dapr API: - [PubSub](pubsub/README.md): a producer app publishing messages with Spring Messaging and a consumer app subscribing to them. - Workflows: - [Patterns](workflows/patterns/README.md): common Dapr Workflow patterns. + - [Event Workflow](workflows/event-workflow/README.md): workflows started and continued by Pub/Sub events. - [Multi App](workflows/multi-app/README.md): workflows that span multiple applications. - [Versioning](workflows/versioning/README.md): versioning workflows. diff --git a/spring-boot-examples/workflows/event-workflow/README.md b/spring-boot-examples/workflows/event-workflow/README.md new file mode 100644 index 0000000000..16ef6d7354 --- /dev/null +++ b/spring-boot-examples/workflows/event-workflow/README.md @@ -0,0 +1,182 @@ +# Dapr Spring Boot Event Driven Workflow Example + +This application shows how to combine [Dapr Workflows](https://docs.dapr.io/developing-applications/building-blocks/workflow/workflow-overview/) +with [Dapr Pub/Sub](https://docs.dapr.io/developing-applications/building-blocks/pubsub/pubsub-overview/) in a Spring Boot +application: + +- An `orders` event starts a new workflow instance. +- The workflow waits for a `payments` event, which is forwarded to it as an external event. +- The workflow publishes its progress to the `order-status` topic from an activity. + +```mermaid +graph LR + O[orders topic] -->|start workflow| SW((Start + Workflow)) + SW --> A1[Reserve Inventory] + A1 --> A2[Publish WAITING_FOR_PAYMENT] + A2 --> E{Wait for + PaymentReceived} + P[payments topic] -->|raise event| E + E --> A3[Publish COMPLETED] + A3 --> EW((End + Workflow)) + A2 --> S[order-status topic] + A3 --> S +``` + +The main classes are: + +- [`OrderEventsSubscriber`](src/main/java/io/dapr/springboot/examples/eventworkflow/OrderEventsSubscriber.java): + subscribes to the topics using the `@Topic` annotation. It starts the workflow with `DaprWorkflowClient.scheduleNewWorkflow`, + using the order id as the workflow instance id, and forwards payments with `DaprWorkflowClient.raiseEvent`. +- [`OrderProcessingWorkflow`](src/main/java/io/dapr/springboot/examples/eventworkflow/OrderProcessingWorkflow.java): + waits for the `PaymentReceived` external event. If the payment does not arrive within 5 minutes, the order is cancelled. +- [`PublishOrderStatusActivity`](src/main/java/io/dapr/springboot/examples/eventworkflow/PublishOrderStatusActivity.java): + publishes the status events using `DaprMessagingTemplate`. Publishing happens in an activity because workflow code + is replayed and must not perform I/O directly. + +## Running this example from source code + +To run this example you will need: +- Java SDK +- Maven +- Docker or a container runtime such as Podman + +From the `spring-boot-examples/workflows/event-workflow` directory you can start the service by running the following command: + + + + +```sh +../../../mvnw spring-boot:test-run +``` + + + +By running the `spring-boot:test-run` goal, the application is loaded using the [test configurations](src/test/java/io/dapr/springboot/examples/eventworkflow/DaprTestContainersConfig.java) +configured using [Testcontainers](https://testcontainers.com) to bootstrap the [Dapr](https://dapr.io) sidecar, +PostgreSQL (used as the workflow state store) and RabbitMQ (used as the Pub/Sub broker). + +## Placing an order + +Publish an order event by sending a POST request to the application: + + + + +```sh +curl -X POST localhost:8080/orders -H 'Content-Type: application/json' -d '{ "id": "order-1", "item": "the mars volta EP", "amount": 1 }' --retry 10 --max-time 20 --retry-all-errors --retry-max-time 90 +``` + + + +The subscriber of the `orders` topic starts a workflow instance. The workflow reserves the inventory and publishes +the `WAITING_FOR_PAYMENT` status. You can check the last status received from the `order-status` topic: + + + + +```sh +curl --fail localhost:8080/orders/order-1/status --retry 10 --max-time 20 --retry-all-errors --retry-max-time 90 +``` + + + +```bash +{"orderId":"order-1","status":"WAITING_FOR_PAYMENT"} +``` + +In the application output you should see: + +```bash +i.d.s.e.e.OrdersRestController : Publishing Order Event: Order{id='order-1', item='the mars volta EP', amount=1} +i.d.s.e.e.OrderEventsSubscriber : Order Event Received: Order{id='order-1', item='the mars volta EP', amount=1} +i.d.s.e.e.OrderEventsSubscriber : Workflow instance order-1 started +i.d.s.e.e.OrderProcessingWorkflow : Starting Workflow: io.dapr.springboot.examples.eventworkflow.OrderProcessingWorkflow +i.d.s.e.e.ReserveInventoryActivity : Reserving 1 unit(s) of 'the mars volta EP' for order order-1 +i.d.s.e.e.PublishOrderStatusActivity : Publishing Order Status Event: OrderStatus{orderId='order-1', status='WAITING_FOR_PAYMENT'} +i.d.s.e.e.OrderEventsSubscriber : Order Status Event Received: OrderStatus{orderId='order-1', status='WAITING_FOR_PAYMENT'} +``` + +## Paying for the order + +Publish a payment event for the order: + + + + +```sh +curl -X POST localhost:8080/orders/order-1/payment -H 'Content-Type: application/json' -d '{ "total": 10.0 }' --retry 10 --max-time 20 --retry-all-errors --retry-max-time 90 +``` + + + +The subscriber of the `payments` topic raises the `PaymentReceived` event on the workflow instance `order-1`, and the +workflow completes the order: + + + + +```sh +curl --fail localhost:8080/orders/order-1/status --retry 10 --max-time 20 --retry-all-errors --retry-max-time 90 +``` + + + +```bash +{"orderId":"order-1","status":"COMPLETED"} +``` + +In the application output you should see: + +```bash +i.d.s.e.e.OrderEventsSubscriber : Payment Event Received: Payment{orderId='order-1', total=10.0} +i.d.s.e.e.OrderProcessingWorkflow : Payment received for order order-1: Payment{orderId='order-1', total=10.0} +i.d.s.e.e.PublishOrderStatusActivity : Publishing Order Status Event: OrderStatus{orderId='order-1', status='COMPLETED'} +i.d.s.e.e.OrderEventsSubscriber : Order Status Event Received: OrderStatus{orderId='order-1', status='COMPLETED'} +``` + +## Testing event driven workflows + +The [`EventWorkflowAppIT`](src/test/java/io/dapr/springboot/examples/eventworkflow/EventWorkflowAppIT.java) test uses +Testcontainers to run the same flow end to end: it places an order, waits for the `WAITING_FOR_PAYMENT` status, +pays the order and checks that the workflow instance completed. diff --git a/spring-boot-examples/workflows/event-workflow/pom.xml b/spring-boot-examples/workflows/event-workflow/pom.xml new file mode 100644 index 0000000000..a351cefe69 --- /dev/null +++ b/spring-boot-examples/workflows/event-workflow/pom.xml @@ -0,0 +1,75 @@ + + + 4.0.0 + + + io.dapr + workflows + 1.20.0-SNAPSHOT + ../pom.xml + + + event-workflow + event-workflow + Spring Boot 4, Testcontainers and Dapr Integration Examples :: Event Driven Workflow + + + + org.springframework.boot + spring-boot-starter-actuator + + + org.springframework.boot + spring-boot-starter-webmvc + + + org.springframework.boot + spring-boot-starter-webmvc-test + test + + + io.dapr.spring + dapr-spring-boot-starter + + + io.dapr.spring + dapr-spring-boot-starter-test + test + + + org.testcontainers + testcontainers-postgresql + test + + + org.testcontainers + testcontainers-rabbitmq + test + + + + + + + org.springframework.boot + spring-boot-maven-plugin + + + org.apache.maven.plugins + maven-site-plugin + + true + + + + org.apache.maven.plugins + maven-checkstyle-plugin + + + true + + + + + diff --git a/spring-boot-examples/workflows/event-workflow/src/main/java/io/dapr/springboot/examples/eventworkflow/EventWorkflowApplication.java b/spring-boot-examples/workflows/event-workflow/src/main/java/io/dapr/springboot/examples/eventworkflow/EventWorkflowApplication.java new file mode 100644 index 0000000000..9b5884fa10 --- /dev/null +++ b/spring-boot-examples/workflows/event-workflow/src/main/java/io/dapr/springboot/examples/eventworkflow/EventWorkflowApplication.java @@ -0,0 +1,26 @@ +/* + * Copyright 2026 The Dapr Authors + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * http://www.apache.org/licenses/LICENSE-2.0 + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and +limitations under the License. +*/ + +package io.dapr.springboot.examples.eventworkflow; + +import org.springframework.boot.SpringApplication; +import org.springframework.boot.autoconfigure.SpringBootApplication; + +@SpringBootApplication +public class EventWorkflowApplication { + + public static void main(String[] args) { + SpringApplication.run(EventWorkflowApplication.class, args); + } + +} diff --git a/spring-boot-examples/workflows/event-workflow/src/main/java/io/dapr/springboot/examples/eventworkflow/EventWorkflowConfiguration.java b/spring-boot-examples/workflows/event-workflow/src/main/java/io/dapr/springboot/examples/eventworkflow/EventWorkflowConfiguration.java new file mode 100644 index 0000000000..21ebbc8da6 --- /dev/null +++ b/spring-boot-examples/workflows/event-workflow/src/main/java/io/dapr/springboot/examples/eventworkflow/EventWorkflowConfiguration.java @@ -0,0 +1,48 @@ +/* + * Copyright 2026 The Dapr Authors + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * http://www.apache.org/licenses/LICENSE-2.0 + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and +limitations under the License. +*/ + +package io.dapr.springboot.examples.eventworkflow; + +import io.dapr.client.DaprClient; +import io.dapr.spring.boot.properties.pubsub.DaprPubSubProperties; +import io.dapr.spring.messaging.DaprMessagingTemplate; +import org.springframework.boot.context.properties.EnableConfigurationProperties; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; + +@Configuration +@EnableConfigurationProperties({DaprPubSubProperties.class}) +public class EventWorkflowConfiguration { + + public static final String ORDERS_TOPIC = "orders"; + public static final String PAYMENTS_TOPIC = "payments"; + public static final String ORDER_STATUS_TOPIC = "order-status"; + + @Bean + public DaprMessagingTemplate orderMessagingTemplate(DaprClient daprClient, + DaprPubSubProperties daprPubSubProperties) { + return new DaprMessagingTemplate<>(daprClient, daprPubSubProperties.getName(), false); + } + + @Bean + public DaprMessagingTemplate paymentMessagingTemplate(DaprClient daprClient, + DaprPubSubProperties daprPubSubProperties) { + return new DaprMessagingTemplate<>(daprClient, daprPubSubProperties.getName(), false); + } + + @Bean + public DaprMessagingTemplate orderStatusMessagingTemplate(DaprClient daprClient, + DaprPubSubProperties daprPubSubProperties) { + return new DaprMessagingTemplate<>(daprClient, daprPubSubProperties.getName(), false); + } +} diff --git a/spring-boot-examples/workflows/event-workflow/src/main/java/io/dapr/springboot/examples/eventworkflow/Order.java b/spring-boot-examples/workflows/event-workflow/src/main/java/io/dapr/springboot/examples/eventworkflow/Order.java new file mode 100644 index 0000000000..a6f2f60177 --- /dev/null +++ b/spring-boot-examples/workflows/event-workflow/src/main/java/io/dapr/springboot/examples/eventworkflow/Order.java @@ -0,0 +1,65 @@ +/* + * Copyright 2026 The Dapr Authors + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * http://www.apache.org/licenses/LICENSE-2.0 + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and +limitations under the License. +*/ + +package io.dapr.springboot.examples.eventworkflow; + +public class Order { + private String id; + private String item; + private Integer amount; + + public Order() { + } + + /** + * Creates a new Order. + * + * @param id order id + * @param item item reference + * @param amount of items in the order + */ + public Order(String id, String item, Integer amount) { + this.id = id; + this.item = item; + this.amount = amount; + } + + public String getId() { + return id; + } + + public void setId(String id) { + this.id = id; + } + + public String getItem() { + return item; + } + + public void setItem(String item) { + this.item = item; + } + + public Integer getAmount() { + return amount; + } + + public void setAmount(Integer amount) { + this.amount = amount; + } + + @Override + public String toString() { + return "Order{" + "id='" + id + '\'' + ", item='" + item + '\'' + ", amount=" + amount + '}'; + } +} diff --git a/spring-boot-examples/workflows/event-workflow/src/main/java/io/dapr/springboot/examples/eventworkflow/OrderEventsSubscriber.java b/spring-boot-examples/workflows/event-workflow/src/main/java/io/dapr/springboot/examples/eventworkflow/OrderEventsSubscriber.java new file mode 100644 index 0000000000..933356e3f4 --- /dev/null +++ b/spring-boot-examples/workflows/event-workflow/src/main/java/io/dapr/springboot/examples/eventworkflow/OrderEventsSubscriber.java @@ -0,0 +1,83 @@ +/* + * Copyright 2026 The Dapr Authors + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * http://www.apache.org/licenses/LICENSE-2.0 + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and +limitations under the License. +*/ + +package io.dapr.springboot.examples.eventworkflow; + +import io.dapr.Topic; +import io.dapr.client.domain.CloudEvent; +import io.dapr.workflows.client.DaprWorkflowClient; +import io.dapr.workflows.client.WorkflowInstanceAlreadyExistsException; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.web.bind.annotation.PostMapping; +import org.springframework.web.bind.annotation.RequestBody; +import org.springframework.web.bind.annotation.RestController; + +/** + * Subscribes to the Pub/Sub topics that drive the {@link OrderProcessingWorkflow}. + */ +@RestController +public class OrderEventsSubscriber { + + private final Logger logger = LoggerFactory.getLogger(OrderEventsSubscriber.class); + + @Autowired + private DaprWorkflowClient daprWorkflowClient; + + @Autowired + private OrderStatusStore orderStatusStore; + + /** + * Start a workflow instance for each order event. The order id is used as the workflow instance id, + * so a redelivered event does not start a second instance for the same order. + * @param cloudEvent order event + */ + @PostMapping("events/orders") + @Topic(pubsubName = "pubsub", name = EventWorkflowConfiguration.ORDERS_TOPIC) + public void onOrder(@RequestBody CloudEvent cloudEvent) { + Order order = cloudEvent.getData(); + logger.info("Order Event Received: " + order); + try { + daprWorkflowClient.scheduleNewWorkflow(OrderProcessingWorkflow.class, order, order.getId()); + logger.info("Workflow instance " + order.getId() + " started"); + } catch (WorkflowInstanceAlreadyExistsException e) { + logger.info("Workflow instance " + order.getId() + " already exists, ignoring duplicated event"); + } + } + + /** + * Forward each payment event to the workflow instance waiting for it. + * @param cloudEvent payment event + */ + @PostMapping("events/payments") + @Topic(pubsubName = "pubsub", name = EventWorkflowConfiguration.PAYMENTS_TOPIC) + public void onPayment(@RequestBody CloudEvent cloudEvent) { + Payment payment = cloudEvent.getData(); + logger.info("Payment Event Received: " + payment); + daprWorkflowClient.raiseEvent(payment.getOrderId(), OrderProcessingWorkflow.PAYMENT_RECEIVED_EVENT, payment); + } + + /** + * Record the status events published by the workflow. + * @param cloudEvent order status event + */ + @PostMapping("events/order-status") + @Topic(pubsubName = "pubsub", name = EventWorkflowConfiguration.ORDER_STATUS_TOPIC) + public void onOrderStatus(@RequestBody CloudEvent cloudEvent) { + OrderStatus orderStatus = cloudEvent.getData(); + logger.info("Order Status Event Received: " + orderStatus); + orderStatusStore.put(orderStatus); + } + +} diff --git a/spring-boot-examples/workflows/event-workflow/src/main/java/io/dapr/springboot/examples/eventworkflow/OrderProcessingWorkflow.java b/spring-boot-examples/workflows/event-workflow/src/main/java/io/dapr/springboot/examples/eventworkflow/OrderProcessingWorkflow.java new file mode 100644 index 0000000000..7c7981e4dc --- /dev/null +++ b/spring-boot-examples/workflows/event-workflow/src/main/java/io/dapr/springboot/examples/eventworkflow/OrderProcessingWorkflow.java @@ -0,0 +1,59 @@ +/* + * Copyright 2026 The Dapr Authors + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * http://www.apache.org/licenses/LICENSE-2.0 + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and +limitations under the License. +*/ + +package io.dapr.springboot.examples.eventworkflow; + +import io.dapr.durabletask.TaskCanceledException; +import io.dapr.workflows.Workflow; +import io.dapr.workflows.WorkflowStub; +import org.springframework.stereotype.Component; + +import java.time.Duration; + +/** + * Workflow started by an order event and continued by a payment event, both received through Pub/Sub. + * Status changes are published back to Pub/Sub from an activity, as workflows must not perform I/O directly. + */ +@Component +public class OrderProcessingWorkflow implements Workflow { + + public static final String PAYMENT_RECEIVED_EVENT = "PaymentReceived"; + + private static final Duration PAYMENT_TIMEOUT = Duration.ofMinutes(5); + + @Override + public WorkflowStub create() { + return ctx -> { + ctx.getLogger().info("Starting Workflow: " + ctx.getName()); + Order order = ctx.getInput(Order.class); + + ctx.callActivity(ReserveInventoryActivity.class.getName(), order).await(); + + OrderStatus waiting = new OrderStatus(order.getId(), OrderStatus.WAITING_FOR_PAYMENT); + ctx.callActivity(PublishOrderStatusActivity.class.getName(), waiting).await(); + + OrderStatus result; + try { + Payment payment = ctx.waitForExternalEvent(PAYMENT_RECEIVED_EVENT, PAYMENT_TIMEOUT, Payment.class).await(); + ctx.getLogger().info("Payment received for order " + order.getId() + ": " + payment); + result = new OrderStatus(order.getId(), OrderStatus.COMPLETED); + } catch (TaskCanceledException e) { + ctx.getLogger().info("Payment not received in time for order " + order.getId()); + result = new OrderStatus(order.getId(), OrderStatus.CANCELLED); + } + + ctx.callActivity(PublishOrderStatusActivity.class.getName(), result).await(); + ctx.complete(result); + }; + } +} diff --git a/spring-boot-examples/workflows/event-workflow/src/main/java/io/dapr/springboot/examples/eventworkflow/OrderStatus.java b/spring-boot-examples/workflows/event-workflow/src/main/java/io/dapr/springboot/examples/eventworkflow/OrderStatus.java new file mode 100644 index 0000000000..287e00c2d9 --- /dev/null +++ b/spring-boot-examples/workflows/event-workflow/src/main/java/io/dapr/springboot/examples/eventworkflow/OrderStatus.java @@ -0,0 +1,59 @@ +/* + * Copyright 2026 The Dapr Authors + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * http://www.apache.org/licenses/LICENSE-2.0 + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and +limitations under the License. +*/ + +package io.dapr.springboot.examples.eventworkflow; + +public class OrderStatus { + + public static final String WAITING_FOR_PAYMENT = "WAITING_FOR_PAYMENT"; + public static final String COMPLETED = "COMPLETED"; + public static final String CANCELLED = "CANCELLED"; + + private String orderId; + private String status; + + public OrderStatus() { + } + + /** + * Creates a new OrderStatus. + * + * @param orderId id of the order + * @param status current status of the order + */ + public OrderStatus(String orderId, String status) { + this.orderId = orderId; + this.status = status; + } + + public String getOrderId() { + return orderId; + } + + public void setOrderId(String orderId) { + this.orderId = orderId; + } + + public String getStatus() { + return status; + } + + public void setStatus(String status) { + this.status = status; + } + + @Override + public String toString() { + return "OrderStatus{" + "orderId='" + orderId + '\'' + ", status='" + status + '\'' + '}'; + } +} diff --git a/spring-boot-examples/workflows/event-workflow/src/main/java/io/dapr/springboot/examples/eventworkflow/OrderStatusStore.java b/spring-boot-examples/workflows/event-workflow/src/main/java/io/dapr/springboot/examples/eventworkflow/OrderStatusStore.java new file mode 100644 index 0000000000..c542bc77a6 --- /dev/null +++ b/spring-boot-examples/workflows/event-workflow/src/main/java/io/dapr/springboot/examples/eventworkflow/OrderStatusStore.java @@ -0,0 +1,40 @@ +/* + * Copyright 2026 The Dapr Authors + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * http://www.apache.org/licenses/LICENSE-2.0 + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and +limitations under the License. +*/ + +package io.dapr.springboot.examples.eventworkflow; + +import org.springframework.stereotype.Component; + +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; + +/** + * Keeps the last status received from the order-status topic for each order. + */ +@Component +public class OrderStatusStore { + + private final Map statuses = new ConcurrentHashMap<>(); + + public void put(OrderStatus orderStatus) { + statuses.put(orderStatus.getOrderId(), orderStatus); + } + + public OrderStatus get(String orderId) { + return statuses.get(orderId); + } + + public void clear() { + statuses.clear(); + } +} diff --git a/spring-boot-examples/workflows/event-workflow/src/main/java/io/dapr/springboot/examples/eventworkflow/OrdersRestController.java b/spring-boot-examples/workflows/event-workflow/src/main/java/io/dapr/springboot/examples/eventworkflow/OrdersRestController.java new file mode 100644 index 0000000000..66d4e4457a --- /dev/null +++ b/spring-boot-examples/workflows/event-workflow/src/main/java/io/dapr/springboot/examples/eventworkflow/OrdersRestController.java @@ -0,0 +1,88 @@ +/* + * Copyright 2026 The Dapr Authors + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * http://www.apache.org/licenses/LICENSE-2.0 + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and +limitations under the License. +*/ + +package io.dapr.springboot.examples.eventworkflow; + +import io.dapr.spring.messaging.DaprMessagingTemplate; +import io.dapr.spring.workflows.config.EnableDaprWorkflows; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.http.ResponseEntity; +import org.springframework.web.bind.annotation.GetMapping; +import org.springframework.web.bind.annotation.PathVariable; +import org.springframework.web.bind.annotation.PostMapping; +import org.springframework.web.bind.annotation.RequestBody; +import org.springframework.web.bind.annotation.RestController; + +import java.util.UUID; + +@RestController +@EnableDaprWorkflows +public class OrdersRestController { + + private final Logger logger = LoggerFactory.getLogger(OrdersRestController.class); + + @Autowired + private DaprMessagingTemplate orderMessagingTemplate; + + @Autowired + private DaprMessagingTemplate paymentMessagingTemplate; + + @Autowired + private OrderStatusStore orderStatusStore; + + /** + * Publish an order event. The workflow is started by the subscriber of the orders topic. + * @param order from the customer + * @return confirmation that the order event was published + */ + @PostMapping("/orders") + public String placeOrder(@RequestBody Order order) { + if (order.getId() == null) { + order.setId(UUID.randomUUID().toString()); + } + logger.info("Publishing Order Event: " + order); + orderMessagingTemplate.send(EventWorkflowConfiguration.ORDERS_TOPIC, order); + return "Order Event Published: " + order.getId(); + } + + /** + * Publish a payment event. The subscriber of the payments topic forwards it to the waiting workflow. + * @param orderId id of the order being paid + * @param payment details of the payment + * @return confirmation that the payment event was published + */ + @PostMapping("/orders/{orderId}/payment") + public String pay(@PathVariable("orderId") String orderId, @RequestBody Payment payment) { + payment.setOrderId(orderId); + logger.info("Publishing Payment Event: " + payment); + paymentMessagingTemplate.send(EventWorkflowConfiguration.PAYMENTS_TOPIC, payment); + return "Payment Event Published: " + orderId; + } + + /** + * Get the last status published by the workflow for an order. + * @param orderId id of the order + * @return the last known status of the order + */ + @GetMapping("/orders/{orderId}/status") + public ResponseEntity status(@PathVariable("orderId") String orderId) { + OrderStatus orderStatus = orderStatusStore.get(orderId); + if (orderStatus == null) { + return ResponseEntity.notFound().build(); + } + return ResponseEntity.ok(orderStatus); + } + +} diff --git a/spring-boot-examples/workflows/event-workflow/src/main/java/io/dapr/springboot/examples/eventworkflow/Payment.java b/spring-boot-examples/workflows/event-workflow/src/main/java/io/dapr/springboot/examples/eventworkflow/Payment.java new file mode 100644 index 0000000000..a6d06e4933 --- /dev/null +++ b/spring-boot-examples/workflows/event-workflow/src/main/java/io/dapr/springboot/examples/eventworkflow/Payment.java @@ -0,0 +1,54 @@ +/* + * Copyright 2026 The Dapr Authors + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * http://www.apache.org/licenses/LICENSE-2.0 + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and +limitations under the License. +*/ + +package io.dapr.springboot.examples.eventworkflow; + +public class Payment { + private String orderId; + private Double total; + + public Payment() { + } + + /** + * Creates a new Payment. + * + * @param orderId id of the order being paid + * @param total amount paid + */ + public Payment(String orderId, Double total) { + this.orderId = orderId; + this.total = total; + } + + public String getOrderId() { + return orderId; + } + + public void setOrderId(String orderId) { + this.orderId = orderId; + } + + public Double getTotal() { + return total; + } + + public void setTotal(Double total) { + this.total = total; + } + + @Override + public String toString() { + return "Payment{" + "orderId='" + orderId + '\'' + ", total=" + total + '}'; + } +} diff --git a/spring-boot-examples/workflows/event-workflow/src/main/java/io/dapr/springboot/examples/eventworkflow/PublishOrderStatusActivity.java b/spring-boot-examples/workflows/event-workflow/src/main/java/io/dapr/springboot/examples/eventworkflow/PublishOrderStatusActivity.java new file mode 100644 index 0000000000..d1a6b00fef --- /dev/null +++ b/spring-boot-examples/workflows/event-workflow/src/main/java/io/dapr/springboot/examples/eventworkflow/PublishOrderStatusActivity.java @@ -0,0 +1,40 @@ +/* + * Copyright 2026 The Dapr Authors + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * http://www.apache.org/licenses/LICENSE-2.0 + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and +limitations under the License. +*/ + +package io.dapr.springboot.examples.eventworkflow; + +import io.dapr.spring.messaging.DaprMessagingTemplate; +import io.dapr.workflows.WorkflowActivity; +import io.dapr.workflows.WorkflowActivityContext; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.stereotype.Component; + +@Component +public class PublishOrderStatusActivity implements WorkflowActivity { + + private final Logger logger = LoggerFactory.getLogger(PublishOrderStatusActivity.class); + + @Autowired + private DaprMessagingTemplate messagingTemplate; + + @Override + public Object run(WorkflowActivityContext ctx) { + logger.info("Starting Activity: " + ctx.getName()); + OrderStatus orderStatus = ctx.getInput(OrderStatus.class); + logger.info("Publishing Order Status Event: " + orderStatus); + messagingTemplate.send(EventWorkflowConfiguration.ORDER_STATUS_TOPIC, orderStatus); + return orderStatus; + } +} diff --git a/spring-boot-examples/workflows/event-workflow/src/main/java/io/dapr/springboot/examples/eventworkflow/ReserveInventoryActivity.java b/spring-boot-examples/workflows/event-workflow/src/main/java/io/dapr/springboot/examples/eventworkflow/ReserveInventoryActivity.java new file mode 100644 index 0000000000..14ac80dded --- /dev/null +++ b/spring-boot-examples/workflows/event-workflow/src/main/java/io/dapr/springboot/examples/eventworkflow/ReserveInventoryActivity.java @@ -0,0 +1,34 @@ +/* + * Copyright 2026 The Dapr Authors + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * http://www.apache.org/licenses/LICENSE-2.0 + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and +limitations under the License. +*/ + +package io.dapr.springboot.examples.eventworkflow; + +import io.dapr.workflows.WorkflowActivity; +import io.dapr.workflows.WorkflowActivityContext; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.stereotype.Component; + +@Component +public class ReserveInventoryActivity implements WorkflowActivity { + + private final Logger logger = LoggerFactory.getLogger(ReserveInventoryActivity.class); + + @Override + public Object run(WorkflowActivityContext ctx) { + logger.info("Starting Activity: " + ctx.getName()); + Order order = ctx.getInput(Order.class); + logger.info("Reserving " + order.getAmount() + " unit(s) of '" + order.getItem() + "' for order " + order.getId()); + return order; + } +} diff --git a/spring-boot-examples/workflows/event-workflow/src/main/resources/application.properties b/spring-boot-examples/workflows/event-workflow/src/main/resources/application.properties new file mode 100644 index 0000000000..71e75dc361 --- /dev/null +++ b/spring-boot-examples/workflows/event-workflow/src/main/resources/application.properties @@ -0,0 +1,2 @@ +dapr.pubsub.name=pubsub +spring.application.name=event-workflow-app diff --git a/spring-boot-examples/workflows/event-workflow/src/test/java/io/dapr/springboot/examples/eventworkflow/DaprTestContainersConfig.java b/spring-boot-examples/workflows/event-workflow/src/test/java/io/dapr/springboot/examples/eventworkflow/DaprTestContainersConfig.java new file mode 100644 index 0000000000..12cb4afb0a --- /dev/null +++ b/spring-boot-examples/workflows/event-workflow/src/test/java/io/dapr/springboot/examples/eventworkflow/DaprTestContainersConfig.java @@ -0,0 +1,124 @@ +/* + * Copyright 2026 The Dapr Authors + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * http://www.apache.org/licenses/LICENSE-2.0 + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and +limitations under the License. +*/ + +package io.dapr.springboot.examples.eventworkflow; + +import io.dapr.testcontainers.Component; +import io.dapr.testcontainers.DaprContainer; +import io.dapr.testcontainers.DaprLogLevel; +import org.springframework.boot.test.context.TestConfiguration; +import org.springframework.boot.testcontainers.service.connection.ServiceConnection; +import org.springframework.context.annotation.Bean; +import org.springframework.core.env.Environment; +import org.testcontainers.DockerClientFactory; +import org.testcontainers.containers.Network; +import org.testcontainers.postgresql.PostgreSQLContainer; +import org.testcontainers.rabbitmq.RabbitMQContainer; +import org.testcontainers.utility.DockerImageName; + +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +import static io.dapr.testcontainers.DaprContainerConstants.DAPR_RUNTIME_IMAGE_TAG; + +/** + * Starts the Dapr sidecar with a PostgreSQL state store for the workflow state and a RabbitMQ Pub/Sub broker + * for the order, payment and order-status topics. + */ +@TestConfiguration(proxyBeanMethods = false) +public class DaprTestContainersConfig { + + @Bean + public Network getDaprNetwork(Environment env) { + boolean reuse = env.getProperty("reuse", Boolean.class, false); + if (reuse) { + Network defaultDaprNetwork = new Network() { + @Override + public String getId() { + return "dapr-network"; + } + + @Override + public void close() { + + } + }; + + List networks = DockerClientFactory.instance().client().listNetworksCmd() + .withNameFilter("dapr-network").exec(); + if (networks.isEmpty()) { + Network.builder().createNetworkCmdModifier(cmd -> cmd.withName("dapr-network")).build().getId(); + return defaultDaprNetwork; + } else { + return defaultDaprNetwork; + } + } else { + return Network.newNetwork(); + } + } + + @Bean + public PostgreSQLContainer postgreSQLContainer(Network daprNetwork) { + return new PostgreSQLContainer(DockerImageName.parse("postgres")) + .withNetworkAliases("postgresql") + .withDatabaseName("dapr") + .withUsername("postgres") + .withPassword("postgres") + .withNetwork(daprNetwork); + } + + @Bean + public RabbitMQContainer rabbitMQContainer(Network daprNetwork, Environment env) { + boolean reuse = env.getProperty("reuse", Boolean.class, false); + return new RabbitMQContainer(DockerImageName.parse("rabbitmq:3.7.25-management-alpine")) + .withExposedPorts(5672) + .withNetworkAliases("rabbitmq") + .withReuse(reuse) + .withNetwork(daprNetwork); + } + + @Bean + @ServiceConnection + public DaprContainer daprContainer(Network daprNetwork, PostgreSQLContainer postgreSQLContainer, + RabbitMQContainer rabbitMQContainer, Environment env) { + boolean reuse = env.getProperty("reuse", Boolean.class, false); + + Map postgreSQLDetails = new HashMap<>(); + postgreSQLDetails.put("host", "postgresql"); + postgreSQLDetails.put("user", "postgres"); + postgreSQLDetails.put("password", "postgres"); + postgreSQLDetails.put("database", "dapr"); + postgreSQLDetails.put("port", "5432"); + postgreSQLDetails.put("actorStateStore", String.valueOf(true)); + + Map rabbitMqProperties = new HashMap<>(); + rabbitMqProperties.put("connectionString", "amqp://guest:guest@rabbitmq:5672"); + rabbitMqProperties.put("user", "guest"); + rabbitMqProperties.put("password", "guest"); + + return new DaprContainer(DAPR_RUNTIME_IMAGE_TAG) + .withAppName("event-workflow-app") + .withNetwork(daprNetwork) + .withComponent(new Component("kvstore", "state.postgresql", "v2", postgreSQLDetails)) + .withComponent(new Component("pubsub", "pubsub.rabbitmq", "v1", rabbitMqProperties)) + .withDaprLogLevel(DaprLogLevel.INFO) + .withLogConsumer(outputFrame -> System.out.println(outputFrame.getUtf8String())) + .withAppPort(8080).withAppChannelAddress("host.testcontainers.internal") + .withReusablePlacement(reuse) + .withReusableScheduler(reuse) + .withAppHealthCheckPath("/actuator/health") + .dependsOn(postgreSQLContainer, rabbitMQContainer); + } + +} diff --git a/spring-boot-examples/workflows/event-workflow/src/test/java/io/dapr/springboot/examples/eventworkflow/EventWorkflowAppIT.java b/spring-boot-examples/workflows/event-workflow/src/test/java/io/dapr/springboot/examples/eventworkflow/EventWorkflowAppIT.java new file mode 100644 index 0000000000..f5dabd3934 --- /dev/null +++ b/spring-boot-examples/workflows/event-workflow/src/test/java/io/dapr/springboot/examples/eventworkflow/EventWorkflowAppIT.java @@ -0,0 +1,108 @@ +/* + * Copyright 2026 The Dapr Authors + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * http://www.apache.org/licenses/LICENSE-2.0 + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and +limitations under the License. +*/ + +package io.dapr.springboot.examples.eventworkflow; + +import io.dapr.springboot.DaprAutoConfiguration; +import io.dapr.testcontainers.DaprContainer; +import io.dapr.workflows.client.DaprWorkflowClient; +import io.dapr.workflows.client.WorkflowRuntimeStatus; +import io.dapr.workflows.client.WorkflowState; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.boot.test.web.server.LocalServerPort; +import org.springframework.http.MediaType; +import org.springframework.test.web.servlet.client.RestTestClient; +import org.testcontainers.containers.wait.strategy.Wait; + +import java.time.Duration; +import java.util.UUID; + +import static org.awaitility.Awaitility.await; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; + +@SpringBootTest(classes = {TestEventWorkflowApplication.class, DaprTestContainersConfig.class, + DaprAutoConfiguration.class}, + webEnvironment = SpringBootTest.WebEnvironment.DEFINED_PORT) +class EventWorkflowAppIT { + + private static final String SUBSCRIPTION_MESSAGE_PATTERN = ".*app is subscribed to the following topics.*"; + + @Autowired + private DaprContainer daprContainer; + + @Autowired + private DaprWorkflowClient daprWorkflowClient; + + @Autowired + private OrderStatusStore orderStatusStore; + + @LocalServerPort + private int port; + + private RestTestClient client; + + @BeforeAll + public static void setup() { + org.testcontainers.Testcontainers.exposeHostPorts(8080); + } + + @BeforeEach + void setUp() { + client = RestTestClient.bindToServer() + .baseUrl("http://localhost:" + port) + .build(); + Wait.forLogMessage(SUBSCRIPTION_MESSAGE_PATTERN, 1).waitUntilReady(daprContainer); + orderStatusStore.clear(); + } + + @Test + void testOrderCompletedByPaymentEvent() { + String orderId = UUID.randomUUID().toString(); + + client.post() + .uri("/orders") + .contentType(MediaType.APPLICATION_JSON) + .body(new Order(orderId, "the mars volta EP", 1)) + .exchange() + .expectStatus().isOk(); + + await().atMost(Duration.ofSeconds(30)) + .until(() -> statusOf(orderId), s -> OrderStatus.WAITING_FOR_PAYMENT.equals(s)); + + client.post() + .uri("/orders/" + orderId + "/payment") + .contentType(MediaType.APPLICATION_JSON) + .body(new Payment(orderId, 10.0)) + .exchange() + .expectStatus().isOk(); + + await().atMost(Duration.ofSeconds(30)) + .until(() -> statusOf(orderId), s -> OrderStatus.COMPLETED.equals(s)); + + WorkflowState workflowState = daprWorkflowClient.getWorkflowState(orderId, true); + assertNotNull(workflowState); + assertEquals(WorkflowRuntimeStatus.COMPLETED, workflowState.getRuntimeStatus()); + assertEquals(OrderStatus.COMPLETED, workflowState.readOutputAs(OrderStatus.class).getStatus()); + } + + private String statusOf(String orderId) { + OrderStatus orderStatus = orderStatusStore.get(orderId); + return orderStatus == null ? null : orderStatus.getStatus(); + } + +} diff --git a/spring-boot-examples/workflows/event-workflow/src/test/java/io/dapr/springboot/examples/eventworkflow/TestEventWorkflowApplication.java b/spring-boot-examples/workflows/event-workflow/src/test/java/io/dapr/springboot/examples/eventworkflow/TestEventWorkflowApplication.java new file mode 100644 index 0000000000..f70104a97a --- /dev/null +++ b/spring-boot-examples/workflows/event-workflow/src/test/java/io/dapr/springboot/examples/eventworkflow/TestEventWorkflowApplication.java @@ -0,0 +1,29 @@ +/* + * Copyright 2026 The Dapr Authors + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * http://www.apache.org/licenses/LICENSE-2.0 + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and +limitations under the License. +*/ + +package io.dapr.springboot.examples.eventworkflow; + +import org.springframework.boot.SpringApplication; +import org.springframework.boot.autoconfigure.SpringBootApplication; + +@SpringBootApplication +public class TestEventWorkflowApplication { + + public static void main(String[] args) { + SpringApplication.from(EventWorkflowApplication::main) + .with(DaprTestContainersConfig.class) + .run(args); + org.testcontainers.Testcontainers.exposeHostPorts(8080); + } + +} diff --git a/spring-boot-examples/workflows/event-workflow/src/test/resources/application.properties b/spring-boot-examples/workflows/event-workflow/src/test/resources/application.properties new file mode 100644 index 0000000000..71e75dc361 --- /dev/null +++ b/spring-boot-examples/workflows/event-workflow/src/test/resources/application.properties @@ -0,0 +1,2 @@ +dapr.pubsub.name=pubsub +spring.application.name=event-workflow-app diff --git a/spring-boot-examples/workflows/pom.xml b/spring-boot-examples/workflows/pom.xml index 5d2f182972..f37f657a0c 100644 --- a/spring-boot-examples/workflows/pom.xml +++ b/spring-boot-examples/workflows/pom.xml @@ -21,6 +21,7 @@ patterns multi-app versioning + event-workflow