broker_request.cpp

#include "asapo/request/broker_request.h"
#include "asapo/request/consumer_error.h"
#include "asapo/http_client/http_error.h"
#include "asapo/http_client/http_client.h"
#include "asapo/common/internal/version.h"
#include "asapo/json_parser/json_parser.h"

#include <chrono>
#include "asapo/io/io_factory.h"
#include <memory>

namespace asapo
{

    Error GetNoDataResponseFromJson(const std::string &json_string, ConsumerErrorData *data)
    {
        JsonStringParser parser(json_string);
        Error err;
        if ((err = parser.GetUInt64("id", &data->id)) || (err = parser.GetUInt64("id_max", &data->id_max)) || (err = parser.GetString("next_stream", &data->next_stream)))
        {
            return err;
        }
        return nullptr;
    }

    Error GetPartialDataResponseFromJson(const std::string &json_string, PartialErrorData *data)
    {
        Error err;
        auto parser = JsonStringParser(json_string);
        uint64_t id, size;
        if ((err = parser.GetUInt64("size", &size)) ||
            (err = parser.GetUInt64("_id", &id)))
        {
            return err;
        }
        data->id = id;
        data->expected_size = size;
        return nullptr;
    }

    Error ConsumerErrorFromPartialDataResponse(const std::string &response)
    {
        PartialErrorData data;
        auto parse_error = GetPartialDataResponseFromJson(response, &data);
        if (parse_error)
        {
            auto err = ConsumerErrorTemplates::kInterruptedTransaction.Generate("malformed response");
            err->AddDetails("response", response);
            return err;
        }
        auto err = ConsumerErrorTemplates::kPartialData.Generate();
        PartialErrorData *error_data = new PartialErrorData{data};
        err->SetCustomData(std::unique_ptr<CustomErrorData>{error_data});
        return err;
    }

    Error ConsumerErrorFromNoDataResponse(const std::string &response)
    {
        if (response.find("get_record_by_id") != std::string::npos)
        {
            ConsumerErrorData data;
            auto parse_error = GetNoDataResponseFromJson(response, &data);
            if (parse_error)
            {
                return ConsumerErrorTemplates::kInterruptedTransaction.Generate("malformed response - " + response);
            }
            Error err;
            if (data.id >= data.id_max)
            {
                err = data.next_stream.empty() ? ConsumerErrorTemplates::kEndOfStream.Generate() : ConsumerErrorTemplates::kStreamFinished.Generate();
            }
            else
            {
                err = ConsumerErrorTemplates::kNoData.Generate();
            }
            ConsumerErrorData *error_data = new ConsumerErrorData{data};
            err->SetCustomData(std::unique_ptr<CustomErrorData>{error_data});
            return err;
        }
        return ConsumerErrorTemplates::kNoData.Generate();
    }

    Error ConsumerErrorFromHttpCode(const RequestOutput *response, const HttpCode &code)
    {
        std::cout << "http code :" << int(code) << "\n";
        switch (code)
        {
        case HttpCode::OK:
            return nullptr;
        case HttpCode::NoContent:
            return nullptr;
        case HttpCode::PartialContent:
            return ConsumerErrorFromPartialDataResponse(response->to_string());
        case HttpCode::BadRequest:
            return ConsumerErrorTemplates::kWrongInput.Generate(response->to_string());
        case HttpCode::Unauthorized:
            return ConsumerErrorTemplates::kWrongInput.Generate(response->to_string());
        case HttpCode::InternalServerError:
            return ConsumerErrorTemplates::kInterruptedTransaction.Generate(response->to_string());
        case HttpCode::ServiceUnavailable:
            return ConsumerErrorTemplates::kUnavailableService.Generate(response->to_string());
        case HttpCode::NotFound:
            return ConsumerErrorTemplates::kUnavailableService.Generate(response->to_string());
        case HttpCode::Conflict:
            return ConsumerErrorFromNoDataResponse(response->to_string());
        case HttpCode::UnsupportedMediaType:
            return ConsumerErrorTemplates::kUnsupportedClient.Generate(response->to_string());
        default:
            return ConsumerErrorTemplates::kInterruptedTransaction.Generate(response->to_string());
        }
    }
    Error ConsumerErrorFromServerError(const Error &server_err)
    {
        if (server_err == HttpErrorTemplates::kTransferError)
        {
            return ConsumerErrorTemplates::kInterruptedTransaction.Generate();
        }
        else
        {
            return ConsumerErrorTemplates::kUnavailableService.Generate();
        }
    }

    Error ProcessRequestResponce(const RequestInfo &request,
                                 Error server_err,
                                 const RequestOutput *response,
                                 const HttpCode &code)
    {
        Error err;
        if (server_err != nullptr)
        {
            err = ConsumerErrorFromServerError(server_err);
            err->SetCause(std::move(server_err));
        }
        else
        {
            err = ConsumerErrorFromHttpCode(response, code);
        }

        if (err != nullptr)
        {
            err->AddDetails("host", request.host)->AddDetails("api", request.api);
        }
        return err;
    }

    ServiceRequest::ServiceRequest(std::string server_uri,
                                   SourceCredentials source) : io__{GenerateDefaultIO()}, httpclient__{DefaultHttpClient()},
                                                               endpoint_{std::move(server_uri)},
                                                               source_credentials_(std::move(source))
    {

        // net_client__ will be lazy initialized

        if (source_credentials_.data_source.empty())
        {
            source_credentials_.data_source = SourceCredentials::kDefaultDataSource;
        }
        if (source_credentials_.instance_id.empty())
        {
            source_credentials_.instance_id = SourceCredentials::kDefaultInstanceId;
        }
        if (source_credentials_.pipeline_step.empty())
        {
            source_credentials_.pipeline_step = SourceCredentials::kDefaultPipelineStep;
        }

        if (source_credentials_.instance_id == SourceCredentials::kDefaultInstanceId)
        {
            Error err;
            std::string hostname = io__->GetHostName(&err);

            if (err)
            {
                hostname = "hostnameerror";
            }

            source_credentials_.instance_id = hostname + "_" + std::to_string(io__->GetCurrentPid());
        }
        if (source_credentials_.pipeline_step == SourceCredentials::kDefaultPipelineStep)
        {
            source_credentials_.pipeline_step = "DefaultStep";
        }

        data_source_encoded_ = httpclient__->UrlEscape(source_credentials_.data_source);

        request_sender_details_prefix_ =
            source_credentials_.instance_id + "§" +
            source_credentials_.pipeline_step + "§" +
            source_credentials_.beamtime_id + "§" +
            source_credentials_.data_source + "§";
    }

    std::string ServiceRequest::RequestWithToken(std::string uri)
    {
        return std::move(uri) + "?token=" + source_credentials_.user_token;
    }

    Error ServiceRequest::ProcessPostRequest(const RequestInfo &request, RequestOutput *response, HttpCode *code)
    {
        Error err;
        switch (request.output_mode)
        {
        case OutputDataMode::string:
            response->string_output =
                httpclient__->Post(RequestWithToken(request.host + request.api) + request.extra_params,
                                   request.cookie,
                                   request.body,
                                   code,
                                   &err);
            break;
        case OutputDataMode::array:
            err =
                httpclient__->Post(RequestWithToken(request.host + request.api) + request.extra_params, request.cookie,
                                   request.body, &response->data_output, response->data_output_size, code);
            break;
        default:
            break;
        }
        return err;
    }

    Error ServiceRequest::ProcessGetRequest(const RequestInfo &request, RequestOutput *response, HttpCode *code)
    {
        Error err;
        response->string_output =
            httpclient__->Get(RequestWithToken(request.host + request.api) + request.extra_params, code, &err);
        return err;
    }

    Error ServiceRequest::ProcessRequest(RequestOutput *response, const RequestInfo &request, std::string *service_uri)
    {
        Error err;
        HttpCode code;
        if (request.post)
        {
            err = ProcessPostRequest(request, response, &code);
        }
        else
        {
            err = ProcessGetRequest(request, response, &code);
        }
        if (err && service_uri)
        {
            service_uri->clear();
        }
        return ProcessRequestResponce(request, std::move(err), response, code);
    }

    RequestInfo ServiceRequest::GetDiscoveryRequest(const std::string &service_name) const
    {
        RequestInfo ri;
        ri.host = endpoint_;
        ri.api = "/asapo-discovery/" + kConsumerProtocol.GetDiscoveryVersion() + "/" + service_name;
        ri.extra_params = "&protocol=" + kConsumerProtocol.GetVersion();
        return ri;
    }

    Error ServiceRequest::ProcessDiscoverServiceResult(Error err, std::string *uri_to_set)
    {
        if (err != nullptr || uri_to_set->empty())
        {
            uri_to_set->clear();
            if (err == ConsumerErrorTemplates::kUnsupportedClient)
            {
                return err;
            }
            auto ret_err = ConsumerErrorTemplates::kUnavailableService.Generate(std::move(err));
            ret_err->AddDetails("destination", endpoint_);
            return ret_err;
        }
        return nullptr;
    }

    std::string ServiceRequest::BrokerRequestWithTimeout(RequestInfo request, Error *err, uint64_t timeout_ms)
    {
        RequestOutput response;
        *err = ServiceRequestWithTimeout(kBrokerServiceName, &current_broker_uri_, request, &response, timeout_ms);
        return std::move(response.string_output);
    }

    std::string ServiceRequest::BrokerRequestWithTimeout(RequestInfo request, Error *err)
    {
        return BrokerRequestWithTimeout(request, err, timeout_ms_);
    }

    Error ServiceRequest::ServiceRequestWithTimeout(const std::string &service_name,
                                                    std::string *service_uri,
                                                    RequestInfo request,
                                                    RequestOutput *response)
    {
        return ServiceRequestWithTimeout(service_name,
                                         service_uri,
                                         request,
                                         response,
                                         timeout_ms_);
    }

    Error ServiceRequest::ServiceRequestWithTimeout(const std::string &service_name,
                                                    std::string *service_uri,
                                                    RequestInfo request,
                                                    RequestOutput *response,
                                                    uint64_t timeout_ms)
    {
        interrupt_flag_ = false;
        uint64_t elapsed_ms = 0;

        Error err;
        while (elapsed_ms <= timeout_ms)
        {
            if (interrupt_flag_)
            {
                err = ConsumerErrorTemplates::kInterruptedTransaction.Generate("interrupted by user request");
                break;
            }
            auto start = std::chrono::steady_clock::now();
            err = DiscoverService(service_name, service_uri);
            if (err == nullptr)
            {
                request.host = *service_uri;
                err = ProcessRequest(response, request, service_uri);
                if (err == nullptr || err == ConsumerErrorTemplates::kWrongInput)
                {
                    break;
                }
            }
            std::this_thread::sleep_for(std::chrono::milliseconds(100));
            elapsed_ms += static_cast<uint64_t>(std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::steady_clock::now() - start).count());
        }
        return err;
    }

    Error ServiceRequest::DiscoverService(const std::string &service_name, std::string *uri_to_set)
    {
        if (!uri_to_set->empty())
        {
            return nullptr;
        }
        auto ri = GetDiscoveryRequest(service_name);
        RequestOutput output;
        auto err = ProcessRequest(&output, ri, nullptr);
        *uri_to_set = std::move(output.string_output);
        return ProcessDiscoverServiceResult(std::move(err), uri_to_set);
    }

    RequestInfo ServiceRequest::CreateBrokerApiRequest(std::string stream, std::string group, std::string suffix, int substream_dataset) const
    {
        auto stream_encoded = httpclient__->UrlEscape(std::move(stream));
        auto group_encoded = group.size() > 0 ? httpclient__->UrlEscape(std::move(group)) : "";
        auto uri = "/" + kConsumerProtocol.GetBrokerVersion() + "/beamtime/" + source_credentials_.beamtime_id + "/" + data_source_encoded_ + "/" + stream_encoded + "/" + std::to_string(substream_dataset);
        if (group_encoded.size() > 0)
        {
            uri = uri + "/" + group_encoded;
        }
        if (suffix.size() > 0)
        {
            uri = uri + "/" + suffix;
        }

        RequestInfo ri;
        ri.api = uri;

        if (use_new_api_format_)
        {
            ri.extra_params += "&instanceid=" + httpclient__->UrlEscape(source_credentials_.instance_id);
            ri.extra_params += "&pipelinestep=" + httpclient__->UrlEscape(source_credentials_.pipeline_step);
        }
        if (substream_dataset >= 0)
        {
            ri.extra_params += "&substream_dataset=" + std::to_string(substream_dataset);
        }
        return ri;
    }

}