make_request implementation (#644)

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

Author: bretambrose

awscrt/mqtt_request_response.py

@@ -6,9 +6,86 @@
 # SPDX-License-Identifier: Apache-2.0.
 
 from dataclasses import dataclass
-from typing import Callable, Union
-from awscrt import NativeResource, mqtt5, mqtt
+from typing import Union
+from awscrt import NativeResource, mqtt5, mqtt, exceptions
+from concurrent.futures import Future
 import _awscrt
+import collections.abc
+
+
+@dataclass
+class Response:
+    """
+    Encapsulates a response to an AWS IoT Core MQTT-based service request
+
+    Args:
+        topic (str):  MQTT Topic that the response was received on.
+        payload (Optional[bytes]):  The payload of the response.
+    """
+    topic: str
+    payload: 'Optional[bytes]' = None
+
+
+@dataclass
+class ResponsePath:
+    """
+    A response path is a pair of values - MQTT topic and a JSON path - that describe how a response to
+    an MQTT-based request may arrive.  For a given request type, there may be multiple response paths and each
+    one is associated with a separate JSON schema for the response body.
+
+    Args:
+        topic (str):  MQTT topic that a response may arrive on.
+        correlation_token_json_path (Optional[str]):  JSON path for finding correlation tokens within payloads that arrive on this path's topic.
+    """
+    topic: str
+    correlation_token_json_path: 'Optional[str]' = None
+
+    def validate(self):
+        assert isinstance(self.topic, str)
+        assert isinstance(self.correlation_token_json_path, str) or self.correlation_token_json_path is None
+
+
+@dataclass
+class RequestResponseOperationOptions:
+    """
+    Configuration options for an MQTT-based request-response operation.
+
+    Args:
+        subscription_topic_filters (Sequence[str]):  Set of topic filters that should be subscribed to in order to cover all possible response paths.  Sometimes using wildcards can cut down on the subscriptions needed; other times that isn't valid.
+        response_paths (Sequence[ResponsePath]):  Set of all possible response paths associated with this request type.
+        publish_topic (str): Topic to publish the request to once response subscriptions have been established.
+        payload (bytes): Payload to publish to 'publishTopic' in order to initiate the request
+        correlation_token (Optional[str]): Correlation token embedded in the request that must be found in a response message.  This can be null to support certain services which don't use correlation tokens.  In that case, the client only allows one token-less request at a time.
+    """
+    subscription_topic_filters: 'Sequence[str]'
+    response_paths: 'Sequence[ResponsePath]'
+    publish_topic: str
+    payload: bytes
+    correlation_token: 'Optional[str]' = None
+
+    def validate(self):
+        assert isinstance(self.subscription_topic_filters, collections.abc.Sequence)
+        for topic_filter in self.subscription_topic_filters:
+            assert isinstance(topic_filter, str)
+
+        assert isinstance(self.response_paths, collections.abc.Sequence)
+        for response_path in self.response_paths:
+            response_path.validate()
+
+        assert isinstance(self.publish_topic, str)
+        assert isinstance(self.payload, bytes)
+        assert isinstance(self.correlation_token, str) or self.correlation_token is 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_topic_filter: str
 
 
 @dataclass
@@ -58,3 +135,25 @@ def __init__(self, protocol_client: Union[mqtt5.Client, mqtt.Connection],
             self._binding = _awscrt.mqtt_request_response_client_new_from_5(protocol_client, client_options)
         else:
             self._binding = _awscrt.mqtt_request_response_client_new_from_311(protocol_client, client_options)
+
+    def make_request(self, options: RequestResponseOperationOptions):
+        options.validate()
+
+        future = Future()
+
+        def on_request_complete(error_code, topic, payload):
+            if error_code != 0:
+                future.set_exception(exceptions.from_code(error_code))
+            else:
+                response = Response(topic=topic, payload=payload)
+                future.set_result(response)
+
+        _awscrt.mqtt_request_response_client_make_request(self._binding,
+                                                          options.subscription_topic_filters,
+                                                          options.response_paths,
+                                                          options.publish_topic,
+                                                          options.payload,
+                                                          options.correlation_token,
+                                                          on_request_complete)
+
+        return future

source/module.c

@@ -780,6 +780,7 @@ static PyMethodDef s_module_methods[] = {
     /* MQTT Request Response Client */
     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),
 
     /* Cryptographic primitives */
     AWS_PY_METHOD_DEF(md5_new, METH_NOARGS),