Add StreamLayerPartitions API (#1427)

New API is using a SAX parser and a JSON byte stream to parse the
layer partitions while they are downloading. Using this approach we
can process large responses without any memory impact, since we do not
wait for the full response, and parse it afterwards.

Relates-To: OLPEDGE-2832

Signed-off-by: Mykhailo Kuchma <[email protected]>

Author: mykhailo-kuchma

olp-cpp-sdk-core/include/olp/core/client/OlpClient.h

@@ -1,5 +1,5 @@
 /*
- * Copyright (C) 2019-2020 HERE Europe B.V.
+ * Copyright (C) 2019-2023 HERE Europe B.V.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -48,7 +48,7 @@ class CORE_API OlpClient {
 
   /**
    * @brief Creates the `OlpClient` instance.
-   * 
+   *
    * @param settings The `OlpClientSettings` instance.
    * @param base_url The base URL to be used for all outgoing requests.
    */
@@ -160,6 +160,33 @@ class CORE_API OlpClient {
                        RequestBodyType post_body, std::string content_type,
                        CancellationContext context) const;
 
+  /**
+   * @brief Executes the HTTP request through the network stack in a blocking
+   * way. The response content is consumed via data callback.
+   *
+   * @param path The path that is appended to the base URL.
+   * @param method Select one of the following methods: `GET`, `POST`, `DELETE`,
+   * or `PUT`.
+   * @param query_params The parameters that are appended to the URL path.
+   * @param header_params The headers used to customize the request.
+   * @param data_callback The network data callback to retrieve content.
+   * @param post_body For the `POST` request, populate `form_params` or
+   * `post_body`, but not both. This data must not be modified until
+   * the request is completed.
+   * @param content_type The content type for the `post_body` or `form_params`.
+   * @param context The `CancellationContext` instance that is used to cancel
+   * the request.
+   *
+   * @return The `HttpResponse` instance.
+   */
+  HttpResponse CallApiStream(std::string path, std::string method,
+                             ParametersType query_params,
+                             ParametersType header_params,
+                             http::Network::DataCallback data_callback,
+                             RequestBodyType post_body,
+                             std::string content_type,
+                             CancellationContext context) const;
+
  private:
   class OlpClientImpl;
   std::shared_ptr<OlpClientImpl> impl_;

olp-cpp-sdk-core/src/client/OlpClient.cpp

@@ -254,6 +254,7 @@ http::NetworkRequest::HttpVerb GetHttpVerb(const std::string& verb) {
 }
 
 HttpResponse SendRequest(const http::NetworkRequest& request,
+                         const http::Network::DataCallback& data_callback,
                          const olp::client::OlpClientSettings& settings,
                          const olp::client::RetrySettings& retry_settings,
                          client::CancellationContext context) {
@@ -264,7 +265,11 @@ HttpResponse SendRequest(const http::NetworkRequest& request,
   };
 
   auto response_data = std::make_shared<ResponseData>();
-  auto response_body = std::make_shared<std::stringstream>();
+
+  // We dont need a response body in case we want a stream
+  auto response_body =
+      data_callback ? nullptr : std::make_shared<std::stringstream>();
+
   http::SendOutcome outcome{http::ErrorCode::CANCELLED_ERROR};
   const auto timeout = std::chrono::seconds(retry_settings.timeout);
 
@@ -279,7 +284,8 @@ HttpResponse SendRequest(const http::NetworkRequest& request,
             [response_data](std::string key, std::string value) {
               response_data->headers.emplace_back(std::move(key),
                                                   std::move(value));
-            });
+            },
+            data_callback);
 
         if (!outcome.IsSuccessful()) {
           OLP_SDK_LOG_WARNING_F(kLogTag,
@@ -321,9 +327,12 @@ HttpResponse SendRequest(const http::NetworkRequest& request,
     const auto status = response_data->response.GetStatus();
     if (status < 0) {
       return HttpResponse{status, response_data->response.GetError()};
-    } else {
+    } else if (response_body) {
       return HttpResponse{status, std::move(*response_body),
                           std::move(response_data->headers)};
+    } else {
+      return HttpResponse{status, std::stringstream(),
+                          std::move(response_data->headers)};
     }
   }();
 
@@ -362,7 +371,8 @@ class OlpClient::OlpClientImpl {
 
   HttpResponse CallApi(std::string path, std::string method,
                        ParametersType query_params,
-                       ParametersType header_params, ParametersType form_params,
+                       ParametersType header_params,
+                       http::Network::DataCallback data_callback,
                        RequestBodyType post_body, std::string content_type,
                        CancellationContext context) const;
 
@@ -595,7 +605,7 @@ HttpResponse OlpClient::OlpClientImpl::CallApi(
     std::string path, std::string method,
     OlpClient::ParametersType query_params,
     OlpClient::ParametersType header_params,
-    OlpClient::ParametersType /*forms_params*/,
+    http::Network::DataCallback data_callback,
     OlpClient::RequestBodyType post_body, std::string content_type,
     CancellationContext context) const {
   if (!settings_.network_request_handler) {
@@ -663,8 +673,8 @@ HttpResponse OlpClient::OlpClientImpl::CallApi(
     return {status, optional_error->GetMessage()};
   }
 
-  auto response =
-      SendRequest(network_request, settings_, retry_settings, context);
+  auto response = SendRequest(network_request, data_callback, settings_,
+                              retry_settings, context);
 
   NetworkStatistics accumulated_statistics = response.GetNetworkStatistics();
 
@@ -697,7 +707,8 @@ HttpResponse OlpClient::OlpClientImpl::CallApi(
     }
 
     backdown_period = CalculateNextWaitTime(retry_settings, i);
-    response = SendRequest(network_request, settings_, retry_settings, context);
+    response = SendRequest(network_request, data_callback, settings_,
+                           retry_settings, context);
 
     // In case we retry, accumulate the stats
     accumulated_statistics += response.GetNetworkStatistics();
@@ -744,13 +755,26 @@ CancellationToken OlpClient::CallApi(
 HttpResponse OlpClient::CallApi(std::string path, std::string method,
                                 ParametersType query_params,
                                 ParametersType header_params,
-                                ParametersType form_params,
+                                ParametersType /*form_params*/,
                                 RequestBodyType post_body,
                                 std::string content_type,
                                 CancellationContext context) const {
   return impl_->CallApi(std::move(path), std::move(method),
                         std::move(query_params), std::move(header_params),
-                        std::move(form_params), std::move(post_body),
+                        nullptr, std::move(post_body), std::move(content_type),
+                        std::move(context));
+}
+
+HttpResponse OlpClient::CallApiStream(std::string path, std::string method,
+                                      ParametersType query_params,
+                                      ParametersType header_params,
+                                      http::Network::DataCallback data_callback,
+                                      RequestBodyType post_body,
+                                      std::string content_type,
+                                      CancellationContext context) const {
+  return impl_->CallApi(std::move(path), std::move(method),
+                        std::move(query_params), std::move(header_params),
+                        std::move(data_callback), std::move(post_body),
                         std::move(content_type), std::move(context));
 }
 

olp-cpp-sdk-dataservice-read/include/olp/dataservice/read/Types.h

@@ -1,5 +1,5 @@
 /*
- * Copyright (C) 2019-2022 HERE Europe B.V.
+ * Copyright (C) 2019-2023 HERE Europe B.V.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -153,6 +153,12 @@ using CompatibleVersionsCallback = Callback<CompatibleVersionsResult>;
 /// The list of tile keys.
 using TileKeys = std::vector<geo::TileKey>;
 
+/// A callback type for partitions stream.
+using PartitionsStreamCallback = std::function<void(model::Partition)>;
+
+/// A type of callback that has no result, or an error.
+using CallbackNoResult = Callback<client::ApiNoResult>;
+
 }  // namespace read
 }  // namespace dataservice
 }  // namespace olp

olp-cpp-sdk-dataservice-read/include/olp/dataservice/read/VersionedLayerClient.h

@@ -1,5 +1,5 @@
 /*
- * Copyright (C) 2019-2022 HERE Europe B.V.
+ * Copyright (C) 2019-2023 HERE Europe B.V.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -282,6 +282,28 @@ class DATASERVICE_READ_API VersionedLayerClient final {
   client::CancellationToken GetPartitions(PartitionsRequest partitions_request,
                                           PartitionsResponseCallback callback);
 
+  /**
+   * @brief Fetches a list of partitions of the given generic layer
+   * asynchronously. Client does not cache the partitions, instead every
+   * partition is passed to the provided callback.
+   * 
+   * @note API is considered experimental and a subject to change.
+   *
+   * @param partitions_request The `PartitionsRequest` instance that contains
+   * a complete set of request parameters.
+   * @note Fetch option and partition list are not supported.
+   * @param partition_stream_callback The `PartitionsStreamCallback` that
+   * receives every fetched partition.
+   * @param callback The `CallbackNoResult` object that is invoked when
+   * operation is complete or an error is encountered.
+   *
+   * @return A token that can be used to cancel this request.
+   */
+  client::CancellationToken StreamLayerPartitions(
+      PartitionsRequest partitions_request,
+      PartitionsStreamCallback partition_stream_callback,
+      CallbackNoResult callback);
+
   /**
    * @brief Fetches a list of partitions of the given generic layer
    * asynchronously.

olp-cpp-sdk-dataservice-read/src/VersionedLayerClient.cpp

@@ -1,5 +1,5 @@
 /*
- * Copyright (C) 2019-2022 HERE Europe B.V.
+ * Copyright (C) 2019-2023 HERE Europe B.V.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -62,6 +62,15 @@ client::CancellationToken VersionedLayerClient::GetPartitions(
                               std::move(callback));
 }
 
+client::CancellationToken VersionedLayerClient::StreamLayerPartitions(
+    PartitionsRequest partitions_request,
+    PartitionsStreamCallback partition_stream_callback,
+    CallbackNoResult callback) {
+  return impl_->StreamLayerPartitions(std::move(partitions_request),
+                                      std::move(partition_stream_callback),
+                                      std::move(callback));
+}
+
 client::CancellableFuture<PartitionsResponse>
 VersionedLayerClient::GetPartitions(PartitionsRequest partitions_request) {
   return impl_->GetPartitions(std::move(partitions_request));

olp-cpp-sdk-dataservice-read/src/VersionedLayerClientImpl.cpp

@@ -1,5 +1,5 @@
 /*
- * Copyright (C) 2019-2022 HERE Europe B.V.
+ * Copyright (C) 2019-2023 HERE Europe B.V.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -48,6 +48,9 @@
 #include "repositories/PartitionsRepository.h"
 #include "repositories/PrefetchTilesRepository.h"
 
+#include "repositories/AsyncJsonStream.h"
+#include "repositories/PartitionsSaxHandler.h"
+
 namespace olp {
 namespace dataservice {
 namespace read {
@@ -110,6 +113,53 @@ client::CancellationToken VersionedLayerClientImpl::GetPartitions(
       std::move(callback), thread::NORMAL);
 }
 
+client::CancellationToken VersionedLayerClientImpl::StreamLayerPartitions(
+    PartitionsRequest request,
+    PartitionsStreamCallback partition_stream_callback,
+    CallbackNoResult callback) {
+  auto async_stream = std::make_shared<repository::AsyncJsonStream>();
+
+  auto request_task =
+      [=](client::CancellationContext context) -> client::ApiNoResponse {
+    auto version_response =
+        GetVersion(boost::none, FetchOptions::OnlineIfNotFound, context);
+    if (!version_response.IsSuccessful()) {
+      return version_response.GetError();
+    }
+
+    const auto version = version_response.GetResult().GetVersion();
+
+    repository::PartitionsRepository repository(catalog_, layer_id_, settings_,
+                                                lookup_client_, mutex_storage_);
+
+    repository.StreamPartitions(async_stream, version,
+                                request.GetAdditionalFields(),
+                                request.GetBillingTag(), context);
+
+    return client::ApiNoResult{};
+  };
+
+  auto request_task_token = task_sink_.AddTask(
+      std::bind(request_task, std::placeholders::_1), nullptr, thread::NORMAL);
+
+  auto parse_task = [=](client::CancellationContext context) {
+    repository::PartitionsRepository repository(catalog_, layer_id_, settings_,
+                                                lookup_client_, mutex_storage_);
+
+    return repository.ParsePartitionsStream(async_stream,
+                                            partition_stream_callback, context);
+  };
+
+  auto parse_task_token =
+      task_sink_.AddTask(std::bind(parse_task, std::placeholders::_1),
+                         std::move(callback), thread::NORMAL);
+
+  return client::CancellationToken([request_task_token, parse_task_token]() {
+    request_task_token.Cancel();
+    parse_task_token.Cancel();
+  });
+}
+
 client::CancellableFuture<PartitionsResponse>
 VersionedLayerClientImpl::GetPartitions(PartitionsRequest partitions_request) {
   auto promise = std::make_shared<std::promise<PartitionsResponse>>();

olp-cpp-sdk-dataservice-read/src/VersionedLayerClientImpl.h

@@ -1,5 +1,5 @@
 /*
- * Copyright (C) 2019-2022 HERE Europe B.V.
+ * Copyright (C) 2019-2023 HERE Europe B.V.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -74,6 +74,11 @@ class VersionedLayerClientImpl {
   virtual client::CancellationToken GetPartitions(
       PartitionsRequest request, PartitionsResponseCallback callback);
 
+  virtual client::CancellationToken StreamLayerPartitions(
+      PartitionsRequest request,
+      PartitionsStreamCallback partition_stream_callback,
+      CallbackNoResult callback);
+
   virtual client::CancellableFuture<PartitionsResponse> GetPartitions(
       PartitionsRequest partitions_request);