Fix handling of metadata, headers. Removes etag from getState.

.github/workflows/build.yml

@@ -22,11 +22,11 @@ jobs:
       GOARCH: amd64
       GOPROXY: https://proxy.golang.org
       JDK_VER: 13.0.x
-      DAPR_CLI_VER: 1.0.0-rc.2
-      DAPR_RUNTIME_VER: 1.0.0-rc.1
+      DAPR_CLI_VER: 1.0.0-rc.3
+      DAPR_RUNTIME_VER: 1.0.0-rc.2
       DAPR_INSTALL_URL: https://raw.githubusercontent.com/dapr/cli/3dacfb672d55f1436c249057aaebbe597e1066f3/install/install.sh
-      DAPR_CLI_REF: 3dacfb672d55f1436c249057aaebbe597e1066f3
-      DAPR_REF: 3cca3cc1567f1cb955ae69b8fd784f075f62ad42
+      DAPR_CLI_REF:
+      DAPR_REF: 4678e477562e7e35a1d6ba04a9b17b6c9c00a025
       OSSRH_USER_TOKEN: ${{ secrets.OSSRH_USER_TOKEN }}
       OSSRH_PWD_TOKEN: ${{ secrets.OSSRH_PWD_TOKEN }}
       GPG_KEY: ${{ secrets.GPG_KEY }}

sdk-tests/src/test/java/io/dapr/it/pubsub/http/PubSubIT.java

@@ -9,18 +9,22 @@
 import io.dapr.client.DaprClientBuilder;
 import io.dapr.client.DaprHttp;
 import io.dapr.client.domain.HttpExtension;
+import io.dapr.client.domain.Metadata;
 import io.dapr.it.BaseIT;
 import io.dapr.it.DaprRun;
+import org.junit.After;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.junit.runners.Parameterized;
 
+import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
 
 import static io.dapr.it.Retry.callWithRetry;
+import static io.dapr.it.TestUtils.assertThrowsDaprException;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertTrue;
 
@@ -34,6 +38,8 @@ public class PubSubIT extends BaseIT {
   //The title of the topic to be used for publishing
   private static final String TOPIC_NAME = "testingtopic";
   private static final String ANOTHER_TOPIC_NAME = "anothertopic";
+  // Topic used for TTL test
+  private static final String TTL_TOPIC_NAME = "ttltopic";
 
   /**
    * Parameters for this test.
@@ -49,16 +55,57 @@ public static Collection<Object[]> data() {
   @Parameterized.Parameter
   public boolean useGrpc;
 
+  private final List<DaprRun> runs = new ArrayList<>();
+
+  private DaprRun closeLater(DaprRun run) {
+    this.runs.add(run);
+    return run;
+  }
+
+  @After
+  public void tearDown() throws Exception {
+    for (DaprRun run : runs) {
+      run.stop();
+    }
+  }
+
+  @Test
+  public void publishPubSubNotFound() throws Exception {
+    DaprRun daprRun = closeLater(startDaprApp(
+        this.getClass().getSimpleName(),
+        60000));
+    if (this.useGrpc) {
+      daprRun.switchToGRPC();
+    } else {
+      daprRun.switchToHTTP();
+    }
+
+    try (DaprClient client = new DaprClientBuilder().build()) {
+
+      if (this.useGrpc) {
+        assertThrowsDaprException(
+            "INVALID_ARGUMENT",
+            "INVALID_ARGUMENT: pubsub unknown pubsub not found",
+            () -> client.publishEvent("unknown pubsub", "mytopic", "payload").block());
+      } else {
+        assertThrowsDaprException(
+            "ERR_PUBSUB_NOT_FOUND",
+            "ERR_PUBSUB_NOT_FOUND: pubsub unknown pubsub not found",
+            () -> client.publishEvent("unknown pubsub", "mytopic", "payload").block());
+      }
+    }
+  }
+
   @Test
   public void testPubSub() throws Exception {
     System.out.println("Working Directory = " + System.getProperty("user.dir"));
 
-    final DaprRun daprRun = startDaprApp(
+    final DaprRun daprRun = closeLater(startDaprApp(
         this.getClass().getSimpleName(),
         SubscriberService.SUCCESS_MESSAGE,
         SubscriberService.class,
         true,
-        60000);
+        60000));
     // At this point, it is guaranteed that the service above is running and all ports being listened to.
     if (this.useGrpc) {
       daprRun.switchToGRPC();
@@ -123,4 +170,62 @@ public void testPubSub() throws Exception {
     }
   }
 
+  @Test
+  public void testPubSubTTLMetadata() throws Exception {
+    System.out.println("Working Directory = " + System.getProperty("user.dir"));
+
+    DaprRun daprRun = closeLater(startDaprApp(
+        this.getClass().getSimpleName(),
+        60000));
+    if (this.useGrpc) {
+      daprRun.switchToGRPC();
+    } else {
+      daprRun.switchToHTTP();
+    }
+
+    // Send a batch of messages on one topic, all to be expired in 1 second.
+    try (DaprClient client = new DaprClientBuilder().build()) {
+      for (int i = 0; i < NUM_MESSAGES; i++) {
+        String message = String.format("This is message #%d on topic %s", i, TTL_TOPIC_NAME);
+        //Publishing messages
+        client.publishEvent(
+            PUBSUB_NAME,
+            TTL_TOPIC_NAME,
+            message,
+            Collections.singletonMap(Metadata.TTL_IN_SECONDS, "1")).block();
+        System.out.println(String.format("Published message: '%s' to topic '%s' pubsub_name '%s'", message, TOPIC_NAME, PUBSUB_NAME));
+      }
+    }
+
+    daprRun.stop();
+
+    // Sleeps for two seconds to let them expire.
+    Thread.sleep(2000);
+
+    daprRun = closeLater(startDaprApp(
+        this.getClass().getSimpleName(),
+        SubscriberService.SUCCESS_MESSAGE,
+        SubscriberService.class,
+        true,
+        60000));
+    if (this.useGrpc) {
+      daprRun.switchToGRPC();
+    } else {
+      daprRun.switchToHTTP();
+    }
+
+    // Sleeps for five seconds to give subscriber a chance to receive messages.
+    Thread.sleep(5000);
+
+    final String appId = daprRun.getAppName();
+    try (DaprClient client = new DaprClientBuilder().build()) {
+      callWithRetry(() -> {
+        System.out.println("Checking results for topic " + TTL_TOPIC_NAME);
+        final List<String> messages = client.invokeMethod(appId, "messages/" + TTL_TOPIC_NAME, null, HttpExtension.GET, List.class).block();
+        assertEquals(0, messages.size());
+      }, 2000);
+    }
+
+    daprRun.stop();
+  }
 }

sdk-tests/src/test/java/io/dapr/it/pubsub/http/SubscriberController.java

@@ -23,6 +23,7 @@ public class SubscriberController {
 
   private static final List<Object> messagesReceivedTestingTopic = new ArrayList();
   private static final List<Object> messagesReceivedAnotherTopic = new ArrayList();
+  private static final List<Object> messagesReceivedTTLTopic = new ArrayList();
 
   @GetMapping(path = "/messages/testingtopic")
   public List<Object> getMessagesReceivedTestingTopic() {
@@ -34,6 +35,11 @@ public List<Object> getMessagesReceivedAnotherTopic() {
     return messagesReceivedAnotherTopic;
   }
 
+  @GetMapping(path = "/messages/ttltopic")
+  public List<Object> getMessagesReceivedTTLTopic() {
+    return messagesReceivedTTLTopic;
+  }
+
   @Topic(name = "testingtopic", pubsubName = "messagebus")
   @PostMapping(path = "/route1")
   public Mono<Void> handleMessage(@RequestBody(required = false) CloudEvent envelope) {
@@ -62,4 +68,18 @@ public Mono<Void> handleMessageAnotherTopic(@RequestBody(required = false) Cloud
     });
   }
 
+  @Topic(name = "ttltopic", pubsubName = "messagebus")
+  @PostMapping(path = "/route3")
+  public Mono<Void> handleMessageTTLTopic(@RequestBody(required = false) CloudEvent envelope) {
+    return Mono.fromRunnable(() -> {
+      try {
+        String message = envelope.getData() == null ? "" : envelope.getData().toString();
+        System.out.println("TTL topic Subscriber got message: " + message);
+        messagesReceivedTTLTopic.add(envelope.getData());
+      } catch (Exception e) {
+        throw new RuntimeException(e);
+      }
+    });
+  }
+
 }

sdk-tests/src/test/java/io/dapr/it/state/GRPCStateClientIT.java

@@ -79,14 +79,4 @@ public void getStatesStoreNotFound() {
             byte[].class).block());
   }
 
-  @Test
-  public void publishPubSubNotFound() {
-    DaprClient daprClient = buildDaprClient();
-
-    // DaprException is guaranteed in the Dapr SDK but getCause() is null in HTTP while present in GRPC implementation.
-    assertThrowsDaprException(
-        "NOT_FOUND",
-        "NOT_FOUND: pubsub 'unknown pubsub' not found",
-        () -> daprClient.publishEvent("unknown pubsub", "mytopic", "payload").block());
-  }
 }

sdk-tests/src/test/java/io/dapr/it/state/HttpStateClientIT.java

@@ -58,7 +58,7 @@ public void getStateStoreNotFound() {
     // DaprException is guaranteed in the Dapr SDK but getCause() is null in HTTP while present in GRPC implementation.
     assertThrowsDaprException(
         "ERR_STATE_STORE_NOT_FOUND",
-        "ERR_STATE_STORE_NOT_FOUND: state store unknown%20state%20store is not found",
+        "ERR_STATE_STORE_NOT_FOUND: state store unknown state store is not found",
         () -> daprClient.getState("unknown state store", new State(stateKey), byte[].class).block());
   }
 
@@ -71,7 +71,7 @@ public void getStatesStoreNotFound() {
     // DaprException is guaranteed in the Dapr SDK but getCause() is null in HTTP while present in GRPC implementation.
     assertThrowsDaprException(
         "ERR_STATE_STORE_NOT_FOUND",
-        "ERR_STATE_STORE_NOT_FOUND: state store unknown%20state%20store is not found",
+        "ERR_STATE_STORE_NOT_FOUND: state store unknown state store is not found",
         () -> daprClient.getBulkState(
             "unknown state store",
             Collections.singletonList(stateKey),
@@ -82,10 +82,6 @@ public void getStatesStoreNotFound() {
   public void publishPubSubNotFound() {
     DaprClient daprClient = buildDaprClient();
 
-    // DaprException is guaranteed in the Dapr SDK but getCause() is null in HTTP while present in GRPC implementation.
-    assertThrowsDaprException(
-        "ERR_PUBSUB_NOT_FOUND",
-        "ERR_PUBSUB_NOT_FOUND: pubsub 'unknown%20pubsub' not found",
-        () -> daprClient.publishEvent("unknown pubsub", "mytopic", "payload").block());
+
   }
 }

sdk/src/main/java/io/dapr/client/AbstractDaprClient.java

@@ -100,7 +100,6 @@ public <T> Mono<T> invokeMethod(
     InvokeServiceRequest req = builder
         .withBody(data)
         .withHttpExtension(httpExtension)
-        .withMetadata(metadata)
         .withContentType(objectSerializer.getContentType())
         .build();
 
@@ -252,41 +251,40 @@ public <T> Mono<T> invokeBinding(
    */
   @Override
   public <T> Mono<State<T>> getState(String storeName, State<T> state, TypeRef<T> type) {
-    return this.getState(storeName, state.getKey(), state.getEtag(), state.getOptions(), type);
+    return this.getState(storeName, state.getKey(), state.getOptions(), type);
   }
 
   /**
    * {@inheritDoc}
    */
   @Override
   public <T> Mono<State<T>> getState(String storeName, State<T> state, Class<T> clazz) {
-    return this.getState(storeName, state.getKey(), state.getEtag(), state.getOptions(), TypeRef.get(clazz));
+    return this.getState(storeName, state.getKey(), state.getOptions(), TypeRef.get(clazz));
   }
 
   /**
    * {@inheritDoc}
    */
   @Override
   public <T> Mono<State<T>> getState(String storeName, String key, TypeRef<T> type) {
-    return this.getState(storeName, key, null, null, type);
+    return this.getState(storeName, key, null, type);
   }
 
   /**
    * {@inheritDoc}
    */
   @Override
   public <T> Mono<State<T>> getState(String storeName, String key, Class<T> clazz) {
-    return this.getState(storeName, key, null, null, TypeRef.get(clazz));
+    return this.getState(storeName, key, null, TypeRef.get(clazz));
   }
 
   /**
    * {@inheritDoc}
    */
   @Override
   public <T> Mono<State<T>> getState(
-          String storeName, String key, String etag, StateOptions options, TypeRef<T> type) {
+          String storeName, String key, StateOptions options, TypeRef<T> type) {
     GetStateRequest request = new GetStateRequestBuilder(storeName, key)
-        .withEtag(etag)
         .withStateOptions(options)
         .build();
     return this.getState(request, type).map(r -> r.getObject());
@@ -298,8 +296,8 @@ public <T> Mono<State<T>> getState(
    */
   @Override
   public <T> Mono<State<T>> getState(
-          String storeName, String key, String etag, StateOptions options, Class<T> clazz) {
-    return this.getState(storeName, key, etag, options, TypeRef.get(clazz));
+          String storeName, String key, StateOptions options, Class<T> clazz) {
+    return this.getState(storeName, key, options, TypeRef.get(clazz));
   }
 
   /**

sdk/src/main/java/io/dapr/client/DaprClient.java

@@ -354,26 +354,24 @@ <T> Mono<T> invokeBinding(String bindingName, String operation, Object data, Map
    *
    * @param storeName The name of the state store.
    * @param key            The key of the State to be retrieved.
-   * @param etag           Optional etag for conditional get
    * @param options        Optional settings for retrieve operation.
    * @param type           The Type of State needed as return.
    * @param <T>            The Type of the return.
    * @return A Mono Plan for the requested State.
    */
-  <T> Mono<State<T>> getState(String storeName, String key, String etag, StateOptions options, TypeRef<T> type);
+  <T> Mono<State<T>> getState(String storeName, String key, StateOptions options, TypeRef<T> type);
 
   /**
    * Retrieve a State based on their key.
    *
    * @param storeName The name of the state store.
    * @param key            The key of the State to be retrieved.
-   * @param etag           Optional etag for conditional get
    * @param options        Optional settings for retrieve operation.
    * @param clazz          The Type of State needed as return.
    * @param <T>            The Type of the return.
    * @return A Mono Plan for the requested State.
    */
-  <T> Mono<State<T>> getState(String storeName, String key, String etag, StateOptions options, Class<T> clazz);
+  <T> Mono<State<T>> getState(String storeName, String key, StateOptions options, Class<T> clazz);
 
   /**
    * Retrieve a State based on their key.