chore: Java tests for catch<Any>

fixes: #2427
block · Aug 19, 2024 · dbfe253 · dbfe253
1 parent a66e555
commit dbfe253
Show file tree

Hide file tree

Showing 3 changed files with 45 additions and 18 deletions.
diff --git a/backend/controller/pubsub/integration_test.go b/backend/controller/pubsub/integration_test.go
@@ -140,8 +140,7 @@ func TestRetry(t *testing.T) {
 			1),
 
 		// check that there was one successful attempt to catchAny
-		// CatchRequest<Any> is not supported in java yet
-		in.IfLanguage("go", in.QueryRow("ftl",
+		in.QueryRow("ftl",
 			fmt.Sprintf(`
 		SELECT COUNT(*)
 		FROM async_calls
@@ -152,7 +151,7 @@ func TestRetry(t *testing.T) {
 			AND catching = true
 			AND origin = '%s'
 `, dal.AsyncOriginPubSub{Subscription: schema.RefKey{Module: "subscriber", Name: "doomedSubscription2"}}.String()),
-			1)),
+			1),
 	)
 }
 

diff --git a/...ub/testdata/java/publisher/src/main/java/xyz/block/ftl/java/test/publisher/Publisher.java b/...ub/testdata/java/publisher/src/main/java/xyz/block/ftl/java/test/publisher/Publisher.java
@@ -8,7 +8,6 @@
 
 public class Publisher {
 
-
     @Export
     @TopicDefinition("testTopic")
     interface TestTopic extends Topic<PubSubEvent> {
@@ -32,9 +31,9 @@ void publishTen(TestTopic testTopic) throws Exception {
 
     @Verb
     void publishOne(TestTopic testTopic) throws Exception {
-            var t = java.time.ZonedDateTime.now();
-            Log.infof("Publishing %s", t);
-            testTopic.publish(new PubSubEvent().setTime(t));
+        var t = java.time.ZonedDateTime.now();
+        Log.infof("Publishing %s", t);
+        testTopic.publish(new PubSubEvent().setTime(t));
     }
 
     @Verb
@@ -44,4 +43,3 @@ void publishOneToTopic2(Topic2 topic2) throws Exception {
         topic2.publish(new PubSubEvent().setTime(t));
     }
 }
-
diff --git a/...testdata/java/subscriber/src/main/java/xyz/block/ftl/java/test/subscriber/Subscriber.java b/...testdata/java/subscriber/src/main/java/xyz/block/ftl/java/test/subscriber/Subscriber.java
@@ -1,36 +1,41 @@
 package xyz.block.ftl.java.test.subscriber;
 
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+
 import ftl.builtin.CatchRequest;
 import ftl.publisher.PubSubEvent;
 import ftl.publisher.TestTopicSubscription;
-import ftl.publisher.Topic2Subscription;
 import io.quarkus.logging.Log;
 import xyz.block.ftl.Retry;
 import xyz.block.ftl.Subscription;
-import xyz.block.ftl.TopicDefinition;
 import xyz.block.ftl.Verb;
 import xyz.block.ftl.VerbName;
 
-import java.util.concurrent.atomic.AtomicInteger;
-
 public class Subscriber {
 
     private static final AtomicInteger catchCount = new AtomicInteger();
+
     @TestTopicSubscription
     void consume(PubSubEvent event) throws Exception {
         Log.infof("Subscriber is consuming %s", event.getTime());
     }
 
-    @Subscription(
-            topic = "topic2",
-            module = "publisher",
-            name = "doomedSubscription"
-    )
+    @Subscription(topic = "topic2", module = "publisher", name = "doomedSubscription")
     @Retry(count = 2, minBackoff = "1s", maxBackoff = "1s", catchVerb = "catch")
     public void consumeButFailAndRetry(PubSubEvent event) {
         throw new RuntimeException("always error: event " + event.getTime());
     }
 
+    @Subscription(topic = "topic2", module = "publisher", name = "doomedSubscription2")
+    @Retry(count = 1, minBackoff = "1s", maxBackoff = "1s", catchVerb = "catchAny")
+    public void consumeButFailAndCatchAny(PubSubEvent event) {
+        throw new RuntimeException("always error: event " + event.getTime());
+    }
+
     @Verb
     @VerbName("catch")
     public void catchVerb(CatchRequest<PubSubEvent> req) {
@@ -41,5 +46,30 @@ public void catchVerb(CatchRequest<PubSubEvent> req) {
             throw new RuntimeException("catching error");
         }
     }
-}
 
+    @Verb
+    public void catchAny(CatchRequest<Object> req) {
+        if (!"subscriber".equals(req.getVerb().getModule())) {
+            throw new IllegalArgumentException(String.format("unexpected verb module: %s", req.getVerb().getModule()));
+        }
+        if (!"consumeButFailAndCatchAny".equals(req.getVerb().getName())) {
+            throw new IllegalArgumentException(String.format("unexpected verb name: %s", req.getVerb().getName()));
+        }
+        if (!"publisher.PubSubEvent".equals(req.getRequestType())) {
+            throw new IllegalArgumentException(String.format("unexpected request type: %s", req.getRequestType()));
+        }
+        if (!(req.getRequest()instanceof Map<?,?>)) {
+            throw new IllegalArgumentException(
+                    String.format("expected request to be a Map: %s", req.getRequest().getClass().getName()));
+        }
+        var request = (Map<?, ?>) req.getRequest();
+        var time = request.get("time");
+        if (time == null) {
+            throw new IllegalArgumentException("expected request to have a time key");
+        }
+        if (!(time instanceof String)) {
+            throw new IllegalArgumentException("expected request to have a time value of type string");
+        }
+    }
+
+}