Data pipeline with subscription replies

[leandro-lucarella-frequenz]

sequenceDiagram
    participant LogicalMeter
    participant Resampler
    participant DataSource
    participant ChannelRegistry

    LogicalMeter-)Resampler: Subscribe(component_id=1, metric_id=2, ...)
    Resampler->>ChannelRegistry: get_sender(channel_name="resampler:1:2")
    ChannelRegistry-->>Resampler: sender

    Resampler-)DataSource: Subscribe(component_id=1, metric_id=2, ...)
    DataSource->>ChannelRegistry: get_sender(channel_name="data_source:1:2")
    ChannelRegistry-->>DataSource: sender

    DataSource--)Resampler: Response(channel_name="data_source:1:2")
    loop Stream
        DataSource->>Resampler: Stream on Response.channel_name starts ...
    end
    Resampler-->>ChannelRegistry: get_receiver(Response.channel_name)
    ChannelRegistry-->>Resampler: receiver
    loop Receive
        Resampler->>Resampler: Receive on receiver and start resampling
    end

    Resampler--)LogicalMeter: Response(channel_name="resampler:1:2")
    loop Stream
        Resampler-)LogicalMeter: Stream on Response.channel_name starts...
    end
    LogicalMeter-->>ChannelRegistry: get_receiver(Response.channel_name)
    ChannelRegistry-->>LogicalMeter: receiver
    loop Receive
        LogicalMeter->>LogicalMeter: Receive on receiver and start calculating
    end

Loading

@dataclass
class ComponentMetricSubscribeRequest:
    component_id: int
    metric_id: str
    start_time: Optional[datetime]

@dataclass
class ComponentMetricSubscribeResponse:
    channel_name: str


# Logical meter at some point will have to subscribe to some metric, for example:

for (component_id, metric_id) in info_we_want:
    subscribe_req = ComponentMetricSubscribeRequest(component_id, metric_id)
    resp = self.resampling_req_sender.send(subscribe_req)
    resampling_receiver = channel_registry.get_receiver(resp.channel_name)

    receivers.append(resampling_receiver)