Request response streams (#647)

Co-authored-by: Bret Ambrose <[email protected]>

Author: bretambrose

awscrt/mqtt_request_response.py

@@ -5,12 +5,95 @@
 # Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
 # SPDX-License-Identifier: Apache-2.0.
 
+from collections.abc import Sequence
+from enum import IntEnum
 from dataclasses import dataclass
-from typing import Union
+from typing import Callable, Union, Optional
 from awscrt import NativeResource, mqtt5, mqtt, exceptions
 from concurrent.futures import Future
 import _awscrt
-import collections.abc
+
+
+class SubscriptionStatusEventType(IntEnum):
+    """
+    The type of change to the state of a streaming operation subscription
+    """
+
+    SUBSCRIPTION_ESTABLISHED = 0
+    """
+    The streaming operation is successfully subscribed to its topic (filter)
+    """
+
+    SUBSCRIPTION_LOST = 1
+    """
+    The streaming operation has temporarily lost its subscription to its topic (filter)
+    """
+
+    SUBSCRIPTION_HALTED = 2
+    """
+    The streaming operation has entered a terminal state where it has given up trying to subscribe
+    to its topic (filter).  This is always due to user error (bad topic filter or IoT Core permission policy).
+    """
+
+
+@dataclass
+class SubscriptionStatusEvent:
+    """
+    An event that describes a change in subscription status for a streaming operation.
+
+    Args:
+        type (SubscriptionStatusEventType):  The type of status change represented by the event
+        error (Optional[Exception]):  Describes an underlying reason for the event.  Only set for SubscriptionLost and SubscriptionHalted.
+    """
+    type: SubscriptionStatusEventType = None
+    error: 'Optional[Exception]' = None
+
+
+@dataclass
+class IncomingPublishEvent:
+    """
+    An event that describes an incoming message on a streaming operation.
+
+    Args:
+        topic (str):  MQTT Topic that the response was received on.
+        payload (Optional[bytes]):  The payload of the incoming message.
+    """
+    topic: str
+    payload: 'Optional[bytes]' = None
+
+
+"""
+Signature for a handler that listens to subscription status events.
+"""
+SubscriptionStatusListener = Callable[[SubscriptionStatusEvent], None]
+
+"""
+Signature for a handler that listens to incoming publish events.
+"""
+IncomingPublishListener = Callable[[IncomingPublishEvent], None]
+
+
+@dataclass
+class StreamingOperationOptions:
+    """
+    Configuration options for an MQTT-based streaming operation.
+
+    Args:
+        subscription_topic_filter (str):  Topic filter that the streaming operation should listen on
+        subscription_status_listener (SubscriptionStatusListener): function object to invoke when the operation's subscription status changes
+        incoming_publish_listener (IncomingPublishListener): function object to invoke when a publish packet arrives that matches the subscription topic filter
+    """
+    subscription_topic_filter: str
+    subscription_status_listener: 'Optional[SubscriptionStatusListener]' = None
+    incoming_publish_listener: 'Optional[IncomingPublishListener]' = None
+
+    def validate(self):
+        """
+        Stringently type-checks an instance's field values.
+        """
+        assert isinstance(self.subscription_topic_filter, str)
+        assert callable(self.subscription_status_listener) or self.subscription_status_listener is None
+        assert callable(self.incoming_publish_listener) or self.incoming_publish_listener is None
 
 
 @dataclass
@@ -41,12 +124,15 @@ class ResponsePath:
     correlation_token_json_path: 'Optional[str]' = None
 
     def validate(self):
+        """
+        Stringently type-checks an instance's field values.
+        """
         assert isinstance(self.topic, str)
         assert isinstance(self.correlation_token_json_path, str) or self.correlation_token_json_path is None
 
 
 @dataclass
-class RequestResponseOperationOptions:
+class RequestOptions:
     """
     Configuration options for an MQTT-based request-response operation.
 
@@ -64,11 +150,14 @@ class RequestResponseOperationOptions:
     correlation_token: 'Optional[str]' = None
 
     def validate(self):
-        assert isinstance(self.subscription_topic_filters, collections.abc.Sequence)
+        """
+        Stringently type-checks an instance's field values.
+        """
+        assert isinstance(self.subscription_topic_filters, Sequence)
         for topic_filter in self.subscription_topic_filters:
             assert isinstance(topic_filter, str)
 
-        assert isinstance(self.response_paths, collections.abc.Sequence)
+        assert isinstance(self.response_paths, Sequence)
         for response_path in self.response_paths:
             response_path.validate()
 
@@ -78,18 +167,7 @@ def validate(self):
 
 
 @dataclass
-class StreamingOperationOptions:
-    """
-    Configuration options for an MQTT-based streaming operation.
-
-    Args:
-        subscription_topic_filter (str):  Topic filter that the streaming operation should listen on
-    """
-    subscription_topic_filter: str
-
-
-@dataclass
-class RequestResponseClientOptions:
+class ClientOptions:
     """
     MQTT-based request-response client configuration options
 
@@ -103,6 +181,9 @@ class RequestResponseClientOptions:
     operation_timeout_in_seconds: 'Optional[int]' = 60
 
     def validate(self):
+        """
+        Stringently type-checks an instance's field values.
+        """
         assert isinstance(self.max_request_response_subscriptions, int)
         assert isinstance(self.max_streaming_subscriptions, int)
         assert isinstance(self.operation_timeout_in_seconds, int)
@@ -123,10 +204,10 @@ class Client(NativeResource):
     """
 
     def __init__(self, protocol_client: Union[mqtt5.Client, mqtt.Connection],
-                 client_options: RequestResponseClientOptions):
+                 client_options: ClientOptions):
 
         assert isinstance(protocol_client, mqtt5.Client) or isinstance(protocol_client, mqtt.Connection)
-        assert isinstance(client_options, RequestResponseClientOptions)
+        assert isinstance(client_options, ClientOptions)
         client_options.validate()
 
         super().__init__()
@@ -136,7 +217,17 @@ def __init__(self, protocol_client: Union[mqtt5.Client, mqtt.Connection],
         else:
             self._binding = _awscrt.mqtt_request_response_client_new_from_311(protocol_client, client_options)
 
-    def make_request(self, options: RequestResponseOperationOptions):
+    def make_request(self, options: RequestOptions):
+        """
+        Initiate an MQTT-based request-response async workflow
+
+        Args:
+            options (RequestOptions): Configuration options for the request to perform
+
+        Returns:
+            concurrent.futures.Future: A Future whose result will contain the topic and payload of a response
+            to the request. The future will contain an exception if the request fails.
+        """
         options.validate()
 
         future = Future()
@@ -157,3 +248,52 @@ def on_request_complete(error_code, topic, payload):
                                                           on_request_complete)
 
         return future
+
+    def create_stream(self, options: StreamingOperationOptions):
+        """
+        Creates a new streaming operation
+
+        Args:
+            options (StreamingOperationOptions): Configuration options for the streaming operation
+
+        Returns:
+            StreamingOperation: a new streaming operation.  Opening the operation triggers the client to maintain
+            an MQTT subscription for relevant events.  Matching publishes and subscription status changes are
+            communicated by invoking configuration-controlled callbacks.
+        """
+        options.validate()
+
+        def on_subscription_status_event(event_type, error_code):
+            if options.subscription_status_listener is not None:
+                event = SubscriptionStatusEvent(event_type)
+                if error_code != 0:
+                    event.error = exceptions.from_code(error_code)
+                options.subscription_status_listener(event)
+
+        def on_incoming_publish_event(topic, payload):
+            if options.incoming_publish_listener is not None:
+                event = IncomingPublishEvent(topic, payload)
+                options.incoming_publish_listener(event)
+
+        stream_binding = _awscrt.mqtt_request_response_client_create_stream(
+            self._binding, options.subscription_topic_filter, on_subscription_status_event, on_incoming_publish_event)
+
+        return StreamingOperation(stream_binding)
+
+
+class StreamingOperation(NativeResource):
+    """
+    An operation that represents a stream of events broadcast to an MQTT topic
+    """
+
+    def __init__(self, binding):
+        super().__init__()
+
+        self._binding = binding
+
+    def open(self):
+        """
+        Triggers the streaming operation to maintain an MQTT subscription for relevant events.  Until a stream is
+        opened, no events can be received.
+        """
+        _awscrt.mqtt_streaming_operation_open(self._binding)

source/module.c

@@ -781,6 +781,8 @@ static PyMethodDef s_module_methods[] = {
     AWS_PY_METHOD_DEF(mqtt_request_response_client_new_from_5, METH_VARARGS),
     AWS_PY_METHOD_DEF(mqtt_request_response_client_new_from_311, METH_VARARGS),
     AWS_PY_METHOD_DEF(mqtt_request_response_client_make_request, METH_VARARGS),
+    AWS_PY_METHOD_DEF(mqtt_request_response_client_create_stream, METH_VARARGS),
+    AWS_PY_METHOD_DEF(mqtt_streaming_operation_open, METH_VARARGS),
 
     /* Cryptographic primitives */
     AWS_PY_METHOD_DEF(md5_new, METH_NOARGS),