run IT test for long values

Signed-off-by: tanvigour <[email protected]>

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

@@ -66,6 +66,8 @@ public class PubSubIT extends BaseIT {
   // Topic to test binary data
   private static final String BINARY_TOPIC_NAME = "binarytopic";
 
+   private static final String LONG_TOPIC_NAME = "testinglongvalues";
+
   /**
    * Parameters for this test.
    * Param #1: useGrpc.
@@ -402,6 +404,55 @@ public void testPubSubTTLMetadata() throws Exception {
     daprRun.stop();
   }
 
+  @Test
+  public void testlongValues() throws Exception {
+    final DaprRun daprRun = closeLater(startDaprApp(
+        this.getClass().getSimpleName(),
+        SubscriberService.SUCCESS_MESSAGE,
+        SubscriberService.class,
+        true,
+        60000));
+    // At this point, it is guaranteed that the service above is running and all ports being listened to.
+    if (this.useGrpc) {
+      daprRun.switchToGRPC();
+    } else {
+      daprRun.switchToHTTP();
+    }
+
+    ConverLong value = new ConverLong();
+    value.setVal(590518626939830271L);
+    try (DaprClient client = new DaprClientBuilder().build()) {
+      for (int i = 0; i < NUM_MESSAGES; i++) {
+        //Publishing messages
+        client.publishEvent(
+            PUBSUB_NAME,
+            LONG_TOPIC_NAME,
+            value,
+            Collections.singletonMap(Metadata.TTL_IN_SECONDS, "1")).block();
+
+        try {
+          Thread.sleep((long) (1000 * Math.random()));
+        } catch (InterruptedException e) {
+          e.printStackTrace();
+          Thread.currentThread().interrupt();
+          return;
+        }
+      }
+    }
+
+    try (DaprClient client = new DaprClientBuilder().build()) {
+      callWithRetry(() -> {
+        System.out.println("Checking results for topic " + LONG_TOPIC_NAME);
+        final List<CloudEvent> messages = client.invokeMethod(
+            daprRun.getAppName(),
+            "messages/testinglongvalues",
+            null,
+            HttpExtension.GET, CLOUD_EVENT_LIST_TYPE_REF).block();
+        assertEquals(590518626939830271L, messages.get(0).getData());
+      }, 2000);
+    }
+  }
+
   public static class MyObject {
     private String id;
 
@@ -413,4 +464,13 @@ public void setId(String id) {
       this.id = id;
     }
   }
+
+  public static class ConverLong {
+    public Long value;
+
+    public void setVal(Long value) {
+      this.value = value;
+    }
+  }
+
 }

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

@@ -115,6 +115,20 @@ public Mono<Void> handleMessageTTLTopic(@RequestBody(required = false) CloudEven
     });
   }
 
+  @Topic(name = "testinglongvalues", pubsubName = "messagebus")
+  @PostMapping(path = "/testinglongvalues")
+  public Mono<Void> handleMessageLongValues(@RequestBody(required = false) CloudEvent<PubSubIT.ConverLong> cloudEvent) {
+    return Mono.fromRunnable(() -> {
+      try {
+        Long message = cloudEvent.getData().value;
+        System.out.println("Subscriber got: " + message);
+        messagesByTopic.compute("testinglongvalues", merge(cloudEvent));
+      } catch (Exception e) {
+        throw new RuntimeException(e);
+      }
+    });
+  }
+
   private BiFunction<String, List<CloudEvent<?>>, List<CloudEvent<?>>> merge(final CloudEvent<?> item) {
     return (key, value) -> {
       final List<CloudEvent<?>> list = value == null ? new ArrayList<>() : value;