integrate into worker

Signed-off-by: Tim Li <[email protected]>

Author: timl3136

cadence/worker/_decision.py

@@ -1,22 +1,22 @@
 import asyncio
 from typing import Optional
 
-from cadence.api.v1.common_pb2 import Payload
-from cadence.api.v1.service_worker_pb2 import PollForDecisionTaskRequest, PollForDecisionTaskResponse, \
-    RespondDecisionTaskFailedRequest
+from cadence.api.v1.service_worker_pb2 import PollForDecisionTaskRequest, PollForDecisionTaskResponse
 from cadence.api.v1.tasklist_pb2 import TaskList, TaskListKind
-from cadence.api.v1.workflow_pb2 import DecisionTaskFailedCause
 from cadence.client import Client
 from cadence.worker._poller import Poller
 from cadence.worker._types import WorkerOptions, _LONG_POLL_TIMEOUT
+from cadence.worker._decision_task_handler import DecisionTaskHandler
+from cadence.worker._registry import Registry
 
 
 class DecisionWorker:
-    def __init__(self, client: Client, task_list: str, options: WorkerOptions) -> None:
+    def __init__(self, client: Client, task_list: str, registry: Registry, options: WorkerOptions) -> None:
         self._client = client
         self._task_list = task_list
         self._identity = options["identity"]
         permits = asyncio.Semaphore(options["max_concurrent_decision_task_execution_size"])
+        self._decision_handler = DecisionTaskHandler(client, task_list, registry, **options)
         self._poller = Poller[PollForDecisionTaskResponse](options["decision_task_pollers"], permits, self._poll, self._execute)
         # TODO: Sticky poller, actually running workflows, etc.
 
@@ -30,17 +30,12 @@ async def _poll(self) -> Optional[PollForDecisionTaskResponse]:
             identity=self._identity,
         ), timeout=_LONG_POLL_TIMEOUT)
 
-        if task.task_token:
+        if task and task.task_token:
             return task
         else:
             return None
 
 
     async def _execute(self, task: PollForDecisionTaskResponse) -> None:
-        await self._client.worker_stub.RespondDecisionTaskFailed(RespondDecisionTaskFailedRequest(
-            task_token=task.task_token,
-            cause=DecisionTaskFailedCause.DECISION_TASK_FAILED_CAUSE_UNHANDLED_DECISION,
-            identity=self._identity,
-            details=Payload(data=b'not implemented')
-        ))
+        await self._decision_handler.handle_task(task)
 

cadence/worker/_worker.py

@@ -19,7 +19,7 @@ def __init__(self, client: Client, task_list: str, registry: Registry, **kwargs:
         _validate_and_copy_defaults(client, task_list, options)
         self._options = options
         self._activity_worker = ActivityWorker(client, task_list, registry, options)
-        self._decision_worker = DecisionWorker(client, task_list, options)
+        self._decision_worker = DecisionWorker(client, task_list, registry, options)
 
 
     async def run(self) -> None: