Request response streams (awslabs#647)

Co-authored-by: Bret Ambrose <bambrose@amazon.com>

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),