/* -*- Mode: C++; tab-width: 4; c-basic-offset: 4; indent-tabs-mode: nil -*- */ /* * Copyright 2020-Present Couchbase, Inc. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ #include #include #include #include "core/cluster.hxx" #include "core/operations/document_analytics.hxx" namespace couchbase { static analytics_error_context build_context(core::operations::analytics_response& resp) { return { resp.ctx.ec, resp.ctx.last_dispatched_to, resp.ctx.last_dispatched_from, resp.ctx.retry_attempts, std::move(resp.ctx.retry_reasons), resp.ctx.first_error_code, std::move(resp.ctx.first_error_message), std::move(resp.ctx.client_context_id), std::move(resp.ctx.statement), std::move(resp.ctx.parameters), std::move(resp.ctx.method), std::move(resp.ctx.path), resp.ctx.http_status, std::move(resp.ctx.http_body), std::move(resp.ctx.hostname), resp.ctx.port, }; } static analytics_status map_status(core::operations::analytics_response::analytics_status status) { switch (status) { case core::operations::analytics_response::running: return analytics_status::running; case core::operations::analytics_response::success: return analytics_status::success; case core::operations::analytics_response::errors: return analytics_status::errors; case core::operations::analytics_response::completed: return analytics_status::completed; case core::operations::analytics_response::stopped: return analytics_status::stopped; case core::operations::analytics_response::timedout: return analytics_status::timeout; case core::operations::analytics_response::closed: return analytics_status::closed; case core::operations::analytics_response::fatal: return analytics_status::fatal; case core::operations::analytics_response::aborted: return analytics_status::aborted; case core::operations::analytics_response::unknown: return analytics_status::unknown; } return analytics_status::unknown; } static std::optional map_scan_consistency(std::optional consistency) { if (consistency.has_value()) { switch (consistency.value()) { case analytics_scan_consistency::not_bounded: return couchbase::core::analytics_scan_consistency::not_bounded; case analytics_scan_consistency::request_plus: return couchbase::core::analytics_scan_consistency::request_plus; } } return {}; } static std::vector map_rows(const core::operations::analytics_response& resp) { std::vector rows; rows.reserve(resp.rows.size()); for (const auto& row : resp.rows) { rows.emplace_back(core::utils::to_binary(row)); } return rows; } static std::vector map_warnings(core::operations::analytics_response& resp) { if (resp.meta.warnings.empty()) { return {}; } std::vector warnings; warnings.reserve(resp.meta.warnings.size()); for (auto& warning : resp.meta.warnings) { warnings.emplace_back(warning.code, std::move(warning.message)); } return warnings; } static analytics_metrics map_metrics(const core::operations::analytics_response& resp) { return analytics_metrics{ resp.meta.metrics.elapsed_time, resp.meta.metrics.execution_time, resp.meta.metrics.result_count, resp.meta.metrics.result_size, resp.meta.metrics.processed_objects, resp.meta.metrics.error_count, resp.meta.metrics.warning_count, }; } static std::optional> map_signature(core::operations::analytics_response& resp) { if (!resp.meta.signature) { return {}; } return core::utils::to_binary(resp.meta.signature.value()); } static analytics_result build_result(core::operations::analytics_response& resp) { return { analytics_meta_data{ std::move(resp.meta.request_id), std::move(resp.meta.client_context_id), map_status(resp.meta.status), map_warnings(resp), map_metrics(resp), map_signature(resp), }, map_rows(resp), }; } static core::operations::analytics_request build_analytics_request(std::string statement, analytics_options::built options, std::optional bucket_name, std::optional scope_name) { core::operations::analytics_request request{ std::move(statement), options.readonly, options.priority, std::move(bucket_name), std::move(scope_name), {}, map_scan_consistency(options.scan_consistency), {}, {}, {}, {}, std::move(options.client_context_id), options.timeout, }; if (!options.raw.empty()) { for (auto& [name, value] : options.raw) { request.raw[name] = std::move(value); } } if (!options.positional_parameters.empty()) { for (auto& value : options.positional_parameters) { request.positional_parameters.emplace_back(std::move(value)); } } if (!options.named_parameters.empty()) { for (auto& [name, value] : options.named_parameters) { request.named_parameters[name] = std::move(value); } } return request; } void cluster::analytics_query(std::string statement, const analytics_options& options, analytics_handler&& handler) const { auto request = build_analytics_request(std::move(statement), options.build(), {}, {}); core_->execute(std::move(request), [handler = std::move(handler)](core::operations::analytics_response resp) mutable { auto r = std::move(resp); return handler(build_context(r), build_result(r)); }); } auto cluster::analytics_query(std::string statement, const analytics_options& options) const -> std::future> { auto barrier = std::make_shared>>(); auto future = barrier->get_future(); analytics_query(std::move(statement), options, [barrier](auto ctx, auto result) { barrier->set_value({ std::move(ctx), std::move(result) }); }); return future; } void scope::analytics_query(std::string statement, const analytics_options& options, analytics_handler&& handler) const { auto request = build_analytics_request(std::move(statement), options.build(), bucket_name_, name_); core_->execute(std::move(request), [handler = std::move(handler)](core::operations::analytics_response resp) mutable { auto r = std::move(resp); return handler(build_context(r), build_result(r)); }); } auto scope::analytics_query(std::string statement, const analytics_options& options) const -> std::future> { auto barrier = std::make_shared>>(); auto future = barrier->get_future(); analytics_query(std::move(statement), options, [barrier](auto ctx, auto result) { barrier->set_value({ std::move(ctx), std::move(result) }); }); return future; } } // namespace couchbase