Merge branch 'master' into actor_ttl

Author: artursouza

dapr-spring/dapr-spring-workflows/src/main/java/io/dapr/spring/workflows/config/DaprWorkflowsConfiguration.java

@@ -17,7 +17,7 @@
 public class DaprWorkflowsConfiguration implements ApplicationContextAware {
   private static final Logger LOGGER = LoggerFactory.getLogger(DaprWorkflowsConfiguration.class);
 
-  private WorkflowRuntimeBuilder workflowRuntimeBuilder;
+  private final WorkflowRuntimeBuilder workflowRuntimeBuilder;
 
   public DaprWorkflowsConfiguration(WorkflowRuntimeBuilder workflowRuntimeBuilder) {
     this.workflowRuntimeBuilder = workflowRuntimeBuilder;
@@ -29,16 +29,21 @@ public DaprWorkflowsConfiguration(WorkflowRuntimeBuilder workflowRuntimeBuilder)
    */
   private void registerWorkflowsAndActivities(ApplicationContext applicationContext) {
     LOGGER.info("Registering Dapr Workflows and Activities");
+
     Map<String, Workflow> workflowBeans = applicationContext.getBeansOfType(Workflow.class);
-    for (Workflow w :  workflowBeans.values()) {
-      LOGGER.info("Dapr Workflow: '{}' registered", w.getClass().getName());
-      workflowRuntimeBuilder.registerWorkflow(w.getClass());
+
+    for (Workflow workflow :  workflowBeans.values()) {
+      LOGGER.info("Dapr Workflow: '{}' registered", workflow.getClass().getName());
+
+      workflowRuntimeBuilder.registerWorkflow(workflow);
     }
 
     Map<String, WorkflowActivity> workflowActivitiesBeans = applicationContext.getBeansOfType(WorkflowActivity.class);
-    for (WorkflowActivity a :  workflowActivitiesBeans.values()) {
-      LOGGER.info("Dapr Workflow Activity: '{}' registered", a.getClass().getName());
-      workflowRuntimeBuilder.registerActivity(a.getClass());
+
+    for (WorkflowActivity activity :  workflowActivitiesBeans.values()) {
+      LOGGER.info("Dapr Workflow Activity: '{}' registered", activity.getClass().getName());
+
+      workflowRuntimeBuilder.registerActivity(activity);
     }
 
     try (WorkflowRuntime runtime = workflowRuntimeBuilder.build()) {

daprdocs/content/en/java-sdk-docs/spring-boot/_index.md

@@ -122,7 +122,7 @@ Besides the previous configuration (`DaprTestContainersConfig`) your tests shoul
 The Java SDK allows you to interface with all of the [Dapr building blocks]({{< ref building-blocks >}}). 
 But if you want to leverage the Spring and Spring Boot programming model you can use the `dapr-spring-boot-starter` integration. 
 This includes implementations of Spring Data (`KeyValueTemplate` and `CrudRepository`) as well as a `DaprMessagingTemplate` for producing and consuming messages 
-(similar to [Spring Kafka](https://spring.io/projects/spring-kafka), [Spring Pulsar](https://spring.io/projects/spring-pulsar) and [Spring AMQP for RabbitMQ](https://spring.io/projects/spring-amqp)).
+(similar to [Spring Kafka](https://spring.io/projects/spring-kafka), [Spring Pulsar](https://spring.io/projects/spring-pulsar) and [Spring AMQP for RabbitMQ](https://spring.io/projects/spring-amqp)) and Dapr workflows. 
 
 ## Using Spring Data `CrudRepository` and `KeyValueTemplate`
 
@@ -277,6 +277,53 @@ public static void setup(){
 
 You can check and run the [full example source code here](https://github.com/salaboy/dapr-spring-boot-docs-examples).
 
+## Using Dapr Workflows with Spring Boot
+
+Following the same approach that we used for Spring Data and Spring Messaging, the `dapr-spring-boot-starter` brings Dapr Workflow integration for Spring Boot users. 
+
+To work with Dapr Workflows you need to define and implement your workflows using code. The Dapr Spring Boot Starter makes your life easier by managing `Workflow`s and `WorkflowActivity`s as Spring beans.
+
+In order to enable the automatic bean discovery you can annotate your `@SpringBootApplication` with the `@EnableDaprWorkflows` annotation: 
+
+```
+@SpringBootApplication
+@EnableDaprWorkflows
+public class MySpringBootApplication {}
+```
+
+By adding this annotation, all the `WorkflowActivity`s will be automatically managed by Spring and registered to the workflow engine. 
+
+By having all `WorkflowActivity`s as managed beans we can use Spring `@Autowired` mechanism to inject any bean that our workflow activity might need to implement its functionality, for example the `@RestTemplate`:
+
+```
+public class MyWorkflowActivity implements WorkflowActivity {
+
+  @Autowired
+  private RestTemplate restTemplate;
+```
+
+You can also `@Autowired` the `DaprWorkflowClient` to create new instances of your workflows. 
+
+```
+@Autowired
+private DaprWorkflowClient daprWorkflowClient;
+```
+
+This enable applications to schedule new workflow instances and raise events.
+
+```
+String instanceId = daprWorkflowClient.scheduleNewWorkflow(MyWorkflow.class, payload);
+```
+
+and
+
+```
+daprWorkflowClient.raiseEvent(instanceId, "MyEvenet", event);
+```
+
+Check the [Dapr Workflow documentation](https://docs.dapr.io/developing-applications/building-blocks/workflow/workflow-overview/) for more information about how to work with Dapr Workflows.
+
+
 ## Next steps
 
 Learn more about the [Dapr Java SDK packages available to add to your Java applications](https://dapr.github.io/java-sdk/).

examples/pom.xml

@@ -134,6 +134,11 @@
       <artifactId>protobuf-java</artifactId>
       <version>${protobuf.version}</version>
     </dependency>
+    <dependency>
+      <groupId>com.squareup.okhttp3</groupId>
+      <artifactId>okhttp</artifactId>
+      <version>4.12.0</version>
+    </dependency>
   </dependencies>
 
   <build>

examples/src/main/java/io/dapr/examples/OpenTelemetryInterceptor.java

@@ -19,12 +19,13 @@
 import jakarta.servlet.DispatcherType;
 import jakarta.servlet.http.HttpServletRequest;
 import jakarta.servlet.http.HttpServletResponse;
-import org.jetbrains.annotations.Nullable;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
 import org.springframework.web.servlet.HandlerInterceptor;
 import org.springframework.web.servlet.ModelAndView;
 
+import javax.annotation.Nullable;
+
 import java.util.Collections;
 
 @Component

sdk-actors/src/test/java/io/dapr/client/DaprHttpProxy.java

@@ -21,8 +21,8 @@
 
 import java.time.Duration;
 
+import static org.assertj.core.api.Assertions.assertThat;
 import static org.junit.jupiter.api.Assertions.assertThrows;
-import static org.junit.jupiter.api.Assertions.assertTrue;
 
 /**
  * Test SDK resiliency.
@@ -43,7 +43,7 @@ public class WaitForSidecarIT extends BaseIT {
   @BeforeAll
   public static void init() throws Exception {
     daprRun = startDaprApp(WaitForSidecarIT.class.getSimpleName(), 5000);
-    daprNotRunning = startDaprApp(WaitForSidecarIT.class.getSimpleName()+"NotRunning", 5000);
+    daprNotRunning = startDaprApp(WaitForSidecarIT.class.getSimpleName() + "NotRunning", 5000);
     daprNotRunning.stop();
 
     toxiProxyRun = new ToxiProxyRun(daprRun, LATENCY, JITTER);
@@ -61,24 +61,30 @@ public void waitSucceeds() throws Exception {
   public void waitTimeout() {
     int timeoutInMillis = (int)LATENCY.minusMillis(100).toMillis();
     long started = System.currentTimeMillis();
+
     assertThrows(RuntimeException.class, () -> {
       try(var client = toxiProxyRun.newDaprClientBuilder().build()) {
         client.waitForSidecar(timeoutInMillis).block();
       }
     });
+
     long duration = System.currentTimeMillis() - started;
-    assertTrue(duration >= timeoutInMillis);
+
+    assertThat(duration).isGreaterThanOrEqualTo(timeoutInMillis);
   }
 
   @Test
   public void waitSlow() throws Exception {
     int timeoutInMillis = (int)LATENCY.plusMillis(100).toMillis();
     long started = System.currentTimeMillis();
+
     try(var client = toxiProxyRun.newDaprClientBuilder().build()) {
         client.waitForSidecar(timeoutInMillis).block();
     }
+
     long duration = System.currentTimeMillis() - started;
-    assertTrue(duration >= LATENCY.toMillis());
+
+    assertThat(duration).isGreaterThanOrEqualTo(LATENCY.toMillis());
   }
 
   @Test
@@ -87,12 +93,15 @@ public void waitNotRunningTimeout() {
     // This has to do with a previous bug in the implementation.
     int timeoutMilliseconds = 5000;
     long started = System.currentTimeMillis();
+
     assertThrows(RuntimeException.class, () -> {
       try(var client = daprNotRunning.newDaprClientBuilder().build()) {
         client.waitForSidecar(timeoutMilliseconds).block();
       }
     });
+
     long duration = System.currentTimeMillis() - started;
-    assertTrue(duration >= timeoutMilliseconds);
+
+    assertThat(duration).isGreaterThanOrEqualTo(timeoutMilliseconds);
   }
 }

sdk-tests/src/test/java/io/dapr/it/resiliency/WaitForSidecarIT.java

@@ -23,16 +23,14 @@
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Tag;
 import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.Disabled;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.test.context.SpringBootTest;
 import org.springframework.boot.test.context.SpringBootTest.WebEnvironment;
 import org.springframework.boot.testcontainers.service.connection.ServiceConnection;
-import org.springframework.test.context.DynamicPropertyRegistry;
-import org.springframework.test.context.DynamicPropertySource;
 import org.testcontainers.containers.Network;
+import org.testcontainers.containers.wait.strategy.Wait;
 import org.testcontainers.junit.jupiter.Container;
 import org.testcontainers.junit.jupiter.Testcontainers;
 
@@ -56,16 +54,18 @@ public class DaprSpringMessagingIT {
   private static final Logger logger = LoggerFactory.getLogger(DaprSpringMessagingIT.class);
 
   private static final String TOPIC = "mockTopic";
-
   private static final Network DAPR_NETWORK = Network.newNetwork();
+  private static final int APP_PORT = 8080;
+  private static final String SUBSCRIPTION_MESSAGE_PATTERN = ".*app is subscribed to the following topics.*";
 
   @Container
   @ServiceConnection
   private static final DaprContainer DAPR_CONTAINER = new DaprContainer("daprio/daprd:1.13.2")
       .withAppName("messaging-dapr-app")
       .withNetwork(DAPR_NETWORK)
       .withComponent(new Component("pubsub", "pubsub.in-memory", "v1", Collections.emptyMap()))
-      .withAppPort(8080)
+      .withAppPort(APP_PORT)
+      .withAppHealthCheckPath("/ready")
       .withDaprLogLevel(DaprLogLevel.DEBUG)
       .withLogConsumer(outputFrame -> System.out.println(outputFrame.getUtf8String()))
       .withAppChannelAddress("host.testcontainers.internal");
@@ -78,16 +78,16 @@ public class DaprSpringMessagingIT {
 
   @BeforeAll
   public static void beforeAll(){
-    org.testcontainers.Testcontainers.exposeHostPorts(8080);
+    org.testcontainers.Testcontainers.exposeHostPorts(APP_PORT);
   }
 
   @BeforeEach
-  public void beforeEach() throws InterruptedException {
-    Thread.sleep(1000);
+  public void beforeEach() {
+    // Ensure the subscriptions are registered
+    Wait.forLogMessage(SUBSCRIPTION_MESSAGE_PATTERN, 1).waitUntilReady(DAPR_CONTAINER);
   }
 
   @Test
-  @Disabled("Test is flaky due to global state in the spring test application.")
   public void testDaprMessagingTemplate() throws InterruptedException {
     for (int i = 0; i < 10; i++) {
       var msg = "ProduceAndReadWithPrimitiveMessageType:" + i;

sdk-tests/src/test/java/io/dapr/it/spring/messaging/DaprSpringMessagingIT.java

@@ -33,7 +33,7 @@ public class TestRestController {
   private static final Logger LOG = LoggerFactory.getLogger(TestRestController.class);
   private final List<CloudEvent<String>> events = new ArrayList<>();
 
-  @GetMapping("/")
+  @GetMapping("/ready")
   public String ok() {
     return "OK";
   }

sdk-tests/src/test/java/io/dapr/it/spring/messaging/TestRestController.java

@@ -23,7 +23,7 @@
 /**
  * Wrapper for Durable Task Framework task activity factory.
  */
-public class WorkflowActivityWrapper<T extends WorkflowActivity> implements TaskActivityFactory {
+public class WorkflowActivityClassWrapper<T extends WorkflowActivity> implements TaskActivityFactory {
   private final Constructor<T> activityConstructor;
   private final String name;
 
@@ -32,7 +32,7 @@ public class WorkflowActivityWrapper<T extends WorkflowActivity> implements Task
    *
    * @param clazz Class of the activity to wrap.
    */
-  public WorkflowActivityWrapper(Class<T> clazz) {
+  public WorkflowActivityClassWrapper(Class<T> clazz) {
     this.name = clazz.getCanonicalName();
     try {
       this.activityConstructor = clazz.getDeclaredConstructor();

sdk-workflows/src/main/java/io/dapr/workflows/runtime/WorkflowActivityWrapper.java renamed to sdk-workflows/src/main/java/io/dapr/workflows/runtime/WorkflowActivityClassWrapper.java

@@ -0,0 +1,46 @@
+/*
+ * Copyright 2023 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.workflows.runtime;
+
+import com.microsoft.durabletask.TaskActivity;
+import com.microsoft.durabletask.TaskActivityFactory;
+import io.dapr.workflows.WorkflowActivity;
+
+/**
+ * Wrapper for Durable Task Framework task activity factory.
+ */
+public class WorkflowActivityInstanceWrapper<T extends WorkflowActivity> implements TaskActivityFactory {
+  private final T activity;
+  private final String name;
+
+  /**
+   * Constructor for WorkflowActivityWrapper.
+   *
+   * @param instance Instance of the activity to wrap.
+   */
+  public WorkflowActivityInstanceWrapper(T instance) {
+    this.name = instance.getClass().getCanonicalName();
+    this.activity = instance;
+  }
+
+  @Override
+  public String getName() {
+    return name;
+  }
+
+  @Override
+  public TaskActivity create() {
+    return ctx -> activity.run(new DefaultWorkflowActivityContext(ctx));
+  }
+}