add missing grpc option in worker grpc client

Signed-off-by: Filinto Duran <[email protected]>

Author: filintod

durabletask/worker.py

@@ -223,13 +223,15 @@ def __init__(
         secure_channel: bool = False,
         interceptors: Optional[Sequence[shared.ClientInterceptor]] = None,
         concurrency_options: Optional[ConcurrencyOptions] = None,
+        channel_options: Optional[Sequence[tuple[str, Any]]] = None,
     ):
         self._registry = _Registry()
         self._host_address = host_address if host_address else shared.get_default_host_address()
         self._logger = shared.get_logger("worker", log_handler, log_formatter)
         self._shutdown = Event()
         self._is_running = False
         self._secure_channel = secure_channel
+        self._channel_options = channel_options
 
         # Use provided concurrency options or create default ones
         self._concurrency_options = (
@@ -306,7 +308,7 @@ def create_fresh_connection():
             current_stub = None
             try:
                 current_channel = shared.get_grpc_channel(
-                    self._host_address, self._secure_channel, self._interceptors
+                    self._host_address, self._secure_channel, self._interceptors, options=self._channel_options
                 )
                 current_stub = stubs.TaskHubSidecarServiceStub(current_channel)
                 current_stub.Hello(empty_pb2.Empty())

tests/durabletask/test_orchestration_e2e.py

@@ -11,7 +11,8 @@
 from durabletask import client, task, worker
 
 # NOTE: These tests assume a sidecar process is running. Example command:
-#       docker run --name durabletask-sidecar -p 4001:4001 --env 'DURABLETASK_SIDECAR_LOGLEVEL=Debug' --rm cgillum/durabletask-sidecar:latest start --backend Emulator
+#       dapr init || true
+#       dapr run --app-id test-app --dapr-grpc-port  4001
 pytestmark = pytest.mark.e2e
 
 
@@ -22,16 +23,18 @@ def empty_orchestrator(ctx: task.OrchestrationContext, _):
         nonlocal invoked  # don't do this in a real app!
         invoked = True
 
+    channel_options = [
+        ("grpc.max_send_message_length", 1024 * 1024),  # 1MB
+    ]
+
     # Start a worker, which will connect to the sidecar in a background thread
-    with worker.TaskHubGrpcWorker() as w:
+    with worker.TaskHubGrpcWorker(channel_options=channel_options) as w:
         w.add_orchestrator(empty_orchestrator)
         w.start()
 
         # set a custom max send length option
         c = client.TaskHubGrpcClient(
-            channel_options=[
-                ("grpc.max_send_message_length", 1024 * 1024),  # 1MB
-            ]
+            channel_options=channel_options
         )
         id = c.schedule_new_orchestration(empty_orchestrator)
         state = c.wait_for_orchestration_completion(id, timeout=30)

tests/durabletask/test_orchestration_e2e_async.py

@@ -13,7 +13,7 @@
 from durabletask.client import OrchestrationStatus
 
 # NOTE: These tests assume a sidecar process is running. Example command:
-#       go install github.com/microsoft/durabletask-go@main
+#       go install github.com/dapr/durabletask-go@main
 #       durabletask-go --port 4001
 pytestmark = [pytest.mark.e2e, pytest.mark.asyncio]
 
@@ -25,12 +25,16 @@ def empty_orchestrator(ctx: task.OrchestrationContext, _):
         nonlocal invoked  # don't do this in a real app!
         invoked = True
 
+    channel_options = [
+        ("grpc.max_send_message_length", 1024 * 1024),  # 1MB
+    ]
+
     # Start a worker, which will connect to the sidecar in a background thread
-    with worker.TaskHubGrpcWorker() as w:
+    with worker.TaskHubGrpcWorker(channel_options=channel_options) as w:
         w.add_orchestrator(empty_orchestrator)
         w.start()
 
-        c = AsyncTaskHubGrpcClient()
+        c = AsyncTaskHubGrpcClient(channel_options=channel_options)
         id = await c.schedule_new_orchestration(empty_orchestrator)
         state = await c.wait_for_orchestration_completion(id, timeout=30)
         await c.aclose()
@@ -58,13 +62,18 @@ def sequence(ctx: task.OrchestrationContext, start_val: int):
             numbers.append(current)
         return numbers
 
+    channel_options =[
+            ("grpc.max_send_message_length", 1024 * 1024),  # 1MB
+        ]
     # Start a worker, which will connect to the sidecar in a background thread
-    with worker.TaskHubGrpcWorker() as w:
+    with worker.TaskHubGrpcWorker(
+        channel_options=channel_options
+    ) as w:
         w.add_orchestrator(sequence)
         w.add_activity(plus_one)
         w.start()
 
-        client = AsyncTaskHubGrpcClient()
+        client = AsyncTaskHubGrpcClient(channel_options=channel_options)
         id = await client.schedule_new_orchestration(sequence, input=1)
         state = await client.wait_for_orchestration_completion(id, timeout=30)
         await client.aclose()