feature: gRPC

.codecov.yml

@@ -11,3 +11,6 @@ coverage:
 
 ignore:
 - "examples/"
+- "pynumaflow/function/generated/udfunction_pb2.py"
+- "pynumaflow/function/_udfunction_pb2.pyi"
+- "pynumaflow/function/generated/udfunction_pb2_grpc.py"

.gitignore

@@ -159,3 +159,6 @@ cython_debug/
 #  option (not recommended) you can uncomment the following to ignore the entire idea folder.
 
 .idea/
+
+# Proto files are coming from numaflow-go https://github.com/numaproj/numaflow-go
+protos/

Makefile

@@ -3,3 +3,5 @@ format:
 
 test:
 	poetry run pytest pynumaflow/tests/
+
+# TODO: proto file generate

README.md

@@ -6,29 +6,26 @@ and [UDSinks](https://numaproj.github.io/numaflow/sinks/user-defined-sinks/) in
 ## Implement a User Defined Function (UDF)
 
 ```python
-from pynumaflow.function import Message, Messages, HTTPHandler
-import random
 
+from pynumaflow.function import Messages, Message, Datum
+from pynumaflow.function.server import UserDefinedFunctionServicer
 
-def my_handler(key: bytes, value: bytes, _) -> Messages:
+
+def map_handler(key: str, datum: Datum) -> Messages:
+    val = datum.value
+    _ = datum.event_time
+    _ = datum.watermark
     messages = Messages()
-    if random.randint(0, 10) % 2 == 0:
-        messages.append(Message.to_all(value))
-    else:
-        messages.append(Message.to_drop())
+    messages.append(Message.to_vtx(key, val))
     return messages
 
 
 if __name__ == "__main__":
-    handler = HTTPHandler(my_handler)
-    handler.start()
+    grpc_server = UserDefinedFunctionServicer(map_handler)
+    grpc_server.start()
 ```
 
-### Sample Image
-
-A sample UDF [Dockerfile](examples/function/udfproj/Dockerfile) is provided 
-under [examples](examples/function/udfproj).
-
+### Sample Image (TODO)
 
 ## Implement a User Defined Sink (UDSink)
 

examples/function/forward_message/example.py

@@ -0,0 +1,16 @@
+from pynumaflow.function import Messages, Message, Datum
+from pynumaflow.function.server import UserDefinedFunctionServicer
+
+
+def map_handler(key: str, datum: Datum) -> Messages:
+    val = datum.value
+    _ = datum.event_time
+    _ = datum.watermark
+    messages = Messages()
+    messages.append(Message.to_vtx(key, val))
+    return messages
+
+
+if __name__ == "__main__":
+    grpc_server = UserDefinedFunctionServicer(map_handler)
+    grpc_server.start()