made USubscription async with InvokeProtoFuture return type

Author: MaximilianToe

include/up-cpp/client/usubscription/v3/Consumer.h

@@ -16,9 +16,7 @@
 #include <up-cpp/communication/RpcClient.h>
 #include <up-cpp/communication/Subscriber.h>
 #include <up-cpp/datamodel/builder/Payload.h>
-#include <up-cpp/utils/ProtoConverter.h>
 #include <uprotocol/core/usubscription/v3/usubscription.pb.h>
-#include <uprotocol/v1/umessage.pb.h>
 
 #include "RequestBuilder.h"
 #include "USubscriptionUUriBuilder.h"

include/up-cpp/client/usubscription/v3/RpcClientUSubscription.h

@@ -52,31 +52,29 @@ struct RpcClientUSubscription : USubscription {
 	using ListenCallback = transport::UTransport::ListenCallback;
 	using ListenHandle = transport::UTransport::ListenHandle;
 
-	template <typename Response>
-	Response invokeResponse(communication::RpcClient rpc_client);
-
 	/// @brief Subscribes from a given topic
 	///
 	/// @param subscription_request The request object containing the topic to
 	/// subscribe to
-	/// @return Returns a SubscriptionResponse on success and a UStatus else
-	utils::Expected<SubscriptionResponse, v1::UStatus> subscribe(
+	/// @return Returns a future that reslves to a SubscriptionResponse on
+	/// success and a UStatus else
+	communication::RpcClient::InvokeProtoFuture<SubscriptionResponse> subscribe(
 	    const SubscriptionRequest& subscription_request) override;
 
 	/// @brief Unsubscribes from a given topic
 	///
 	/// @param unsubscribe_request The request object containing the topic to
 	/// unsubscribe from
 	/// @return Returns an UnsubscribeResponse on success and a UStatus else
-	utils::Expected<UnsubscribeResponse, v1::UStatus> unsubscribe(
-	    const UnsubscribeRequest& unsubscribe_request) override;
+	communication::RpcClient::InvokeProtoFuture<UnsubscribeResponse>
+	unsubscribe(const UnsubscribeRequest& unsubscribe_request) override;
 
 	/// @brief Fetches the list of topics the client is subscribed to
 	///
 	/// @param fetch_subscriptions_request The request object
-	/// @return Returns a FetchSubscriptionsResponse on success and a UStatus
-	/// else
-	utils::Expected<FetchSubscriptionsResponse, v1::UStatus>
+	/// @return Returns a future that reslves to a FetchSubscriptionsResponse on
+	/// success and a UStatus else
+	communication::RpcClient::InvokeProtoFuture<FetchSubscriptionsResponse>
 	fetch_subscriptions(
 	    const FetchSubscriptionsRequest& fetch_subscriptions_request) override;
 
@@ -85,38 +83,47 @@ struct RpcClientUSubscription : USubscription {
 	/// @param fetch_subscribers_request The request object containing the topic
 	/// for which the subscribers are to be fetched
 	/// @return Returns a FetchSubscribersResponse on success and a UStatus else
-	utils::Expected<FetchSubscribersResponse, v1::UStatus> fetch_subscribers(
+	communication::RpcClient::InvokeProtoFuture<FetchSubscribersResponse>
+	fetch_subscribers(
 	    const FetchSubscribersRequest& fetch_subscribers_request) override;
 
 	/// @brief Registers to receive notifications
 	///
 	/// @param register_notifications_request The request object containing
 	///	       the details to register for notifications
-	/// @return Returns a NotificationResponse on success and a UStatus else
-	utils::Expected<NotificationsResponse, v1::UStatus>
+	/// @return Returns a future that resolves to a NotificationResponse on
+	/// success and a UStatus else
+	communication::RpcClient::InvokeProtoFuture<NotificationsResponse>
 	register_for_notifications(
 	    const NotificationsRequest& register_notifications_request) override;
 
 	/// @brief Unregisters from receiving notifications.
 	///
 	/// @param unregister_notifications_request The request object containing
 	///        the details needed to stop receiving notifications.
-	/// @return Returns a NotificationResponse on success and a UStatus else
-	utils::Expected<NotificationsResponse, v1::UStatus>
+	/// @return Returns future that resolves to a NotificationResponse on
+	/// success and a UStatus else
+	communication::RpcClient::InvokeProtoFuture<NotificationsResponse>
 	unregister_for_notifications(
 	    const NotificationsRequest& unregister_notifications_request) override;
 
 	/// @brief Constructor
 	///
 	/// @param transport Transport used to send messages
 	explicit RpcClientUSubscription(
-	    std::shared_ptr<transport::UTransport> transport)
-	    : transport_(std::move(transport)) {}
+	    std::shared_ptr<transport::UTransport> transport);
 
 	~RpcClientUSubscription() override = default;
 
 private:
 	std::shared_ptr<transport::UTransport> transport_;
+	std::shared_ptr<communication::RpcClient> subscribe_client_;
+	std::shared_ptr<communication::RpcClient> unsubscribe_client_;
+	std::shared_ptr<communication::RpcClient> fetch_subscriptions_client_;
+	std::shared_ptr<communication::RpcClient> fetch_subscribers_client_;
+	std::shared_ptr<communication::RpcClient> register_for_notification_client_;
+	std::shared_ptr<communication::RpcClient>
+	    unregister_for_notification_client_;
 
 	USubscriptionUUriBuilder uuri_builder_;
 };

include/up-cpp/client/usubscription/v3/USubscription.h

@@ -12,9 +12,9 @@
 #ifndef UP_CPP_CLIENT_USUBSCRIPTION_V3_USUBSCRIPTION_H
 #define UP_CPP_CLIENT_USUBSCRIPTION_V3_USUBSCRIPTION_H
 #include <uprotocol/core/usubscription/v3/usubscription.pb.h>
-#include <uprotocol/v1/umessage.pb.h>
 #include <uprotocol/v1/ustatus.pb.h>
 
+#include "up-cpp/communication/RpcClient.h"
 #include "up-cpp/utils/Expected.h"
 
 namespace uprotocol::core::usubscription::v3 {
@@ -33,46 +33,57 @@ struct USubscription {
 	/// response on success or else a status code
 	///
 	/// @param subscription_request containing a topic to subscribe to
-	/// @return SubscriptionReponse on success and UStatus else
-	virtual ResponseOrStatus<SubscriptionResponse> subscribe(
-	    const SubscriptionRequest& subscription_request) = 0;
+	/// @return future that resolves to a SubscriptionReponse on success and
+	/// UStatus else
+	virtual communication::RpcClient::InvokeProtoFuture<SubscriptionResponse>
+	subscribe(const SubscriptionRequest& subscription_request) = 0;
 
 	/// @brief sends an unsubscribe request to a USubscription backend and a
 	/// response on success or else a status code
 	///
 	/// @param unsubscribe_request containing a topic to unsubscribe
-	/// @return UnsubscribeResponse on success and UStatus else
-	virtual ResponseOrStatus<UnsubscribeResponse> unsubscribe(
-	    const UnsubscribeRequest& unsubscribe_request) = 0;
+	/// @return future that resolves to UnsubscribeResponse on success and
+	/// UStatus else
+	virtual communication::RpcClient::InvokeProtoFuture<UnsubscribeResponse>
+	unsubscribe(const UnsubscribeRequest& unsubscribe_request) = 0;
 
 	/// @brief fetches all topics the client is subscribed to from the backend
 	///
 	/// @param fetch_subscriptions_request
-	/// @return FetchSubscriptionsResponse on success and UStatus else
-	virtual ResponseOrStatus<FetchSubscriptionsResponse> fetch_subscriptions(
+	/// @return future that resolves to FetchSubscriptionsResponse on success
+	/// and UStatus else
+	virtual communication::RpcClient::InvokeProtoFuture<
+	    FetchSubscriptionsResponse>
+	fetch_subscriptions(
 	    const FetchSubscriptionsRequest& fetch_subscriptions_request) = 0;
 
 	/// @brief registers for notifications to a USubscription backend
 	///
 	/// @param register_notifications_request
-	/// @return NotificationResponse on success and UStatus else
-	virtual ResponseOrStatus<NotificationsResponse> register_for_notifications(
+	/// @return future that resolves to NotificationResponse on success and
+	/// UStatus else
+	virtual communication::RpcClient::InvokeProtoFuture<NotificationsResponse>
+	register_for_notifications(
 	    const NotificationsRequest& register_notifications_request) = 0;
 
 	/// @brief unregisters for notifications to a USubscription backend
 	///
 	/// @param unregister_notifications_request
-	/// @return NotificationResponse on success and UStatus else
-	virtual ResponseOrStatus<NotificationsResponse>
+	/// @return future that resolves to NotificationResponse on success and
+	/// UStatus else
+	virtual communication::RpcClient::InvokeProtoFuture<NotificationsResponse>
 	unregister_for_notifications(
 	    const NotificationsRequest& unregister_notifications_request) = 0;
 
 	/// @brief fetches all subscribers for a given topic from the backend
 	///
 	/// @param fetch_subscriptions_request containing the topic for which the
 	/// subscribers are fetched
-	/// @return FetchSubscriptionsResponse on success and UStatus else
-	virtual ResponseOrStatus<FetchSubscribersResponse> fetch_subscribers(
+	/// @return future that resolves to FetchSubscriptionsResponse on success
+	/// and UStatus else
+	virtual communication::RpcClient::InvokeProtoFuture<
+	    FetchSubscribersResponse>
+	fetch_subscribers(
 	    const FetchSubscribersRequest& fetch_subscribers_request) = 0;
 };
 

include/up-cpp/communication/RpcClient.h

@@ -76,17 +76,22 @@ struct RpcClient {
 	///        for the duration of an RPC call.
 	using InvokeHandle = Connection::Handle;
 
-	/// @brief Extension to std::future that also holds a callback handle
-	class InvokeFuture {
+	/// @brief Extension to std::future with template type that also holds a
+	/// callback handle
+	template <typename T>
+	class InvokeProtoFuture {
 		InvokeHandle callback_handle_;
-		std::future<MessageOrStatus> future_;
+		std::future<utils::Expected<T, v1::UStatus>> future_;
 
 	public:
-		InvokeFuture();
-		InvokeFuture(InvokeFuture&&) noexcept;
-		InvokeFuture(std::future<MessageOrStatus>&&, InvokeHandle&&) noexcept;
+		InvokeProtoFuture() = default;
+		InvokeProtoFuture(InvokeProtoFuture&& other) noexcept = default;
+		InvokeProtoFuture& operator=(InvokeProtoFuture&& other) noexcept =
+		    default;
 
-		InvokeFuture& operator=(InvokeFuture&&) noexcept;
+		InvokeProtoFuture(std::future<utils::Expected<T, v1::UStatus>>&& future,
+		                  InvokeHandle&& handle) noexcept
+		    : callback_handle_(std::move(handle)), future_(std::move(future)) {}
 
 		/// @name Passthroughs for std::future
 		/// @{
@@ -104,6 +109,8 @@ struct RpcClient {
 		/// @}
 	};
 
+	using InvokeFuture = InvokeProtoFuture<v1::UMessage>;
+
 	/// @brief Invokes an RPC method by sending a request message.
 	///
 	/// @param A Payload builder containing the payload to be sent with the
@@ -169,36 +176,48 @@ struct RpcClient {
 	[[nodiscard]] InvokeFuture invokeMethod();
 
 	template <typename T, typename R>
-	ResponseOrStatus<T> invokeProtoMethod(const R& request_message) {
+	InvokeProtoFuture<T> invokeProtoMethod(const R& request_message) {
+		auto result_promise =
+		    std::make_shared<std::promise<ResponseOrStatus<T>>>();
+		auto future = result_promise->get_future();
+
 		auto payload_or_status =
 		    uprotocol::utils::ProtoConverter::protoToPayload(request_message);
 
 		if (!payload_or_status.has_value()) {
-			return ResponseOrStatus<T>(
-			    UnexpectedStatus(payload_or_status.error()));
+			result_promise->set_value(ResponseOrStatus<T>(
+			    UnexpectedStatus(payload_or_status.error())));
+			return {std::move(future), InvokeHandle()};
 		}
 
 		datamodel::builder::Payload tmp_payload(payload_or_status.value());
 
-		auto message_or_status =
-		    this->invokeMethod(std::move(tmp_payload)).get();
-
-		if (!message_or_status.has_value()) {
-			return ResponseOrStatus<T>(
-			    UnexpectedStatus(message_or_status.error()));
-		}
-
-		auto response_or_status = utils::ProtoConverter::extractFromProtobuf<T>(
-		    message_or_status.value());
-
-		if (!response_or_status.has_value()) {
-			spdlog::error(
-			    "invokeProtoMethod: Error when extracting response from "
-			    "protobuf.");
-			return response_or_status;
-		}
-
-		return ResponseOrStatus<T>(response_or_status.value());
+		auto handle = invokeMethod(
+		    builder_.build(std::move(tmp_payload)),
+		    [result_promise](const MessageOrStatus& message_or_status) {
+			    if (!message_or_status.has_value()) {
+				    result_promise->set_value(ResponseOrStatus<T>(
+				        UnexpectedStatus(message_or_status.error())));
+				    return;
+			    }
+			    auto response_or_status =
+			        utils::ProtoConverter::extractFromProtobuf<T>(
+			            message_or_status.value());
+
+			    if (!response_or_status.has_value()) {
+				    spdlog::error(
+				        "invokeProtoMethod: Error when extracting response "
+				        "from "
+				        "protobuf.");
+				    result_promise->set_value(response_or_status);
+				    return;
+			    }
+
+			    result_promise->set_value(
+			        ResponseOrStatus<T>(response_or_status.value()));
+		    });
+
+		return {std::move(future), std::move(handle)};
 	}
 
 	/// @brief Default move constructor (defined in RpcClient.cpp)
@@ -224,4 +243,4 @@ struct RpcClient {
 
 }  // namespace uprotocol::communication
 
-#endif  // UP_CPP_COMMUNICATION_RPCCLIENT_H
+#endif  // UP_CPP_COMMUNICATION_RPCCLIENT_H