Make RequestHandler a functional interface (#8)

* Use TypeResolver to resolve generic types * Make Pulsar Function a functional interface * address comments
sijie · Mar 4, 2018 · 7c871b0 · 7c871b0
1 parent 9e7d30c
commit 7c871b0
Show file tree

Hide file tree

Showing 3 changed files with 22 additions and 1 deletion.
diff --git a/pulsar-functions/src/main/java/org/apache/pulsar/functions/api/RawRequestHandler.java b/pulsar-functions/src/main/java/org/apache/pulsar/functions/api/RawRequestHandler.java
@@ -29,6 +29,7 @@
  * deserialize this in any way they want before processing. Any output goes into
  * OutputStream.
  */
+@FunctionalInterface
 public interface RawRequestHandler {
     /**
      * Process the input coming as an input stream. The process function

diff --git a/pulsar-functions/src/main/java/org/apache/pulsar/functions/api/RequestHandler.java b/pulsar-functions/src/main/java/org/apache/pulsar/functions/api/RequestHandler.java
@@ -25,6 +25,7 @@
  * Map, and List types) and for org.Json type. If this serialization approach does not
  * meet your needs, you can use the byte stream handler defined in RawRequestHandler.
  */
+@FunctionalInterface
 public interface RequestHandler<I, O> {
     /**
      * Process the input.

diff --git a/pulsar-functions/src/test/java/org/apache/pulsar/functions/instance/JavaInstanceTest.java b/pulsar-functions/src/test/java/org/apache/pulsar/functions/instance/JavaInstanceTest.java
@@ -24,6 +24,7 @@
 import org.slf4j.LoggerFactory;
 import org.testng.annotations.Test;
 
+import static com.google.common.base.Charsets.UTF_8;
 import static org.testng.Assert.*;
 
 public class JavaInstanceTest {
@@ -81,12 +82,30 @@ public void testLongRunningFunction() throws Exception {
         JavaInstance instance = new JavaInstance(config, new LongRunningHandler(), log);
         String testString = "ABC123";
         JavaInstance.ExecutionResult result =
-                instance.handleMessage("1", "random", testString.getBytes());
+                instance.handleMessage("1", "random", testString.getBytes(UTF_8));
 
         assertNotNull(result.getTimeoutException());
         assertNull(result.getUserException());
     }
 
+    /**
+     * Verify that be able to run lambda functions.
+     * @throws Exception
+     */
+    @Test
+    public void testLambda() throws Exception {
+        JavaInstanceConfig config = new JavaInstanceConfig();
+        config.setTimeBudgetInMs(2000);
+        JavaInstance instance = new JavaInstance(
+            config,
+            (RequestHandler<String, String>) (input, context) -> input + "-lambda",
+            log);
+        String testString = "ABC123";
+        JavaInstance.ExecutionResult result =
+            instance.handleMessage("1", "random", testString.getBytes(UTF_8));
+        assertEquals(testString + "-lambda", result.getResultValue());
+    }
+
     /**
      * Verify that JavaInstance does not support functions that are not native Java types
      * @throws Exception