polling: Support service handler for ADR003#103
Conversation
…et_handler() NewDataDetectionPipeline previously required result_handler via __init__, which Temporal cannot satisfy when instantiating workflow classes. Extract polling, change-detection, and handler coordination to PollDataUseCase. Subclasses override get_handler() to supply a handler without constructor injection. PollingManager.start_polling gains a workflow_name parameter (replacing the unused downstream_pipeline arg) to enable subclass-based routing.
| info: list[str] = Field( | ||
| default_factory=list, | ||
| description="Informational messages about handler processing", | ||
| ) |
There was a problem hiding this comment.
I couldn't see a reason for the other 3 string arrays that are on here in the ADR (debug, warning, error). It read as though we expected to gather all log messages during the handler's execution or something, but that clearly isn't the case (we just start it - so the only info we can return is related to starting, or why we didn't start it (no-comply) etc.). So I've just used info only. Let me know if there's some other reason I'm missing.
There was a problem hiding this comment.
YAGNI - this is fine.
It was like that because I didn't want to have logic "if error messages not empty, then UNABLE". rather I wanted to decouple the "what she said" impartial message relay from the semantics of roger/wilco/unable. potential over-complication, no objection to slimming it down until we know we need it.
2 participants
The polling contrib module needed to trigger domain-specific downstream workflows when new data was detected, but couldn't do this without knowing about those workflows — a cross-BC coupling problem.
The previous approach used "bridge pipelines" (
PollingDataPreparationPipeline), an extra intermediate workflow for each integration point. This worked but required solution providers to maintain a dedicated pipeline just to hand off between the polling system and their processing workflows.This PR implements the ADR 003 handler pattern for polling, replacing bridge pipelines with injected handler services. The polling system recognises a condition (new data detected) and hands off to whatever the solution provider has configured — without knowing what that is.
Temporal instantiation bug fixed
NewDataDetectionPipeline.__init__previously required aresult_handlerargument that Temporal could never provide — the worker registers the workflow class and Temporal instantiates it with no arguments. The handler was structurally unreachable.The fix moves the handler lookup to
get_handler(), an instance method that subclassing pipelines can override. Temporal instantiates the base class (no-op, returnsNone) or a subclass (returns the appropriate handler). No constructor injection required.Business logic extracted to use case
The polling, change-detection, and handler coordination logic that lived directly in the workflow
run()method has been extracted toPollDataUseCase. The pipeline is now pure Temporal orchestration. The use case can be tested standalone without Temporal.PollingManager.start_pollingfixedThe manager was passing
args=[config, downstream_pipeline]to the scheduled workflow, butrun()only acceptsconfig— a silent mismatch. Thedownstream_pipelinearg is replaced byworkflow_name, which lets solution providers schedule a specific subclass for a polling change detection pipeline.