* Copyright (c) Huawei Technologies Co., Ltd. 2025. All rights reserved.
*
* 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 "httpd/http_connect.hpp"
#include "httpd/http_client.hpp"
#include "actor/actor.hpp"
#include "actor/actormgr.hpp"
#include "actor/buslog.hpp"
#include "async/defer.hpp"
#include "async/asyncafter.hpp"
#include "async/future.hpp"
#include "async/option.hpp"
#include "litebus.hpp"
#include "utils/os_utils.hpp"
#include "securec.h"
#include <list>
static const std::string CONNECT_PREFIX = "CONNECT_";
static uint64_t g_requestTimeout = 90000;
using std::placeholders::_1;
using litebus::ActorBase;
using litebus::AID;
using litebus::AsyncAfter;
using litebus::Defer;
using litebus::Future;
using litebus::Option;
using litebus::Promise;
using litebus::Timer;
namespace litebus {
namespace http {
using Pipeline = std::queue<std::shared_ptr<litebus::Promise<litebus::http::Response>>>;
class HttpConnectionActor : public ActorBase {
public:
~HttpConnectionActor() override
{
delete pipeline;
pipeline = nullptr;
}
HttpConnectionActor(const HttpConnectionActor &) = delete;
HttpConnectionActor &operator=(const HttpConnectionActor &) = delete;
HttpConnectionActor(const int conSeq, const URL &url)
: ActorBase(CONNECT_PREFIX + std::to_string(conSeq)),
connectSeq(conSeq),
pipeline(new(std::nothrow) Pipeline()),
connectUrl(url),
timeOut(false),
responseTimer(Timer())
{
BUS_OOM_EXIT(pipeline);
}
Future<Response> LaunchRequest(const Request &request, const bool &timeOutFlag)
{
if (!CompareUrlDomain(request.url, connectUrl)) {
return Status(static_cast<int32_t>(INVALID_REQUEST));
}
if (pipeline == nullptr) {
BUSLOG_WARN("Connection has been closed,conSeq={}", connectSeq);
return Status(static_cast<int32_t>(CONNECTION_REFUSED));
}
if (disconnection.GetFuture().IsOK()) {
BUSLOG_WARN("Connection is closing,conSeq={}", connectSeq);
return Status(static_cast<int32_t>(CONNECTION_REFUSED));
}
if (SendResult.GetFuture().IsError()) {
BUSLOG_WARN("Launch request failed before,conSeq={}", connectSeq);
return Status(static_cast<int32_t>(CONNECTION_REFUSED));
}
timeOut = timeOutFlag;
std::shared_ptr<litebus::Promise<Response>> promise(std::make_shared<litebus::Promise<Response>>());
BUS_OOM_EXIT(promise);
pipeline->push(promise);
int tempSeq = connectSeq;
Future<bool> sendResult = HttpClient::GetInstance()->LaunchRequest(request, tempSeq);
(void)sendResult.OnComplete(
Defer(GetAID(), &HttpConnectionActor::HandleRequestCompleted, std::placeholders::_1, request));
return promise->GetFuture();
}
void HandleResponse(const Response &response)
{
if (pipeline == nullptr) {
BUSLOG_WARN("Connection has been closed,conSeq={}", connectSeq);
return;
}
bool closeFlag = false;
if (pipeline->empty()) {
(void)Disconnect();
return;
}
if (timeOut) {
(void)TimerTools::Cancel(responseTimer);
}
std::shared_ptr<litebus::Promise<Response>> promise = pipeline->front();
promise->SetValue(response);
pipeline->pop();
HeaderMap headers = response.headers;
auto headerIter = headers.find("Connection");
if ((headerIter == headers.end()) || (headers["Connection"] == "close")) {
BUSLOG_DEBUG("This is the last response, close the connection, conSeq:{}", connectSeq);
closeFlag = true;
while (!pipeline->empty()) {
promise = pipeline->front();
promise->SetFailed(static_cast<int32_t>(CANNOT_SEND_AFTER_SHUTDOWN));
pipeline->pop();
}
}
if (closeFlag && pipeline->empty()) {
(void)Disconnect();
}
return;
}
Future<bool> Disconnected()
{
return disconnection.GetFuture();
}
Future<bool> Disconnect()
{
return HttpClient::GetInstance()
->Disconnect(connectSeq)
.Then(Defer(GetAID(), &HttpConnectionActor::HandleDisconnect, CANNOT_SEND_AFTER_SHUTDOWN));
}
Future<bool> HandleDisconnect(const int errCode)
{
if (pipeline == nullptr) {
BUSLOG_DEBUG("Connection has been closed,conSeq={}", connectSeq);
return true;
}
if (timeOut) {
(void)TimerTools::Cancel(responseTimer);
}
while (!pipeline->empty()) {
std::shared_ptr<litebus::Promise<Response>> promise = pipeline->front();
promise->SetFailed((errCode == 0) ? static_cast<int>(CONNECTION_REFUSED) : errCode);
pipeline->pop();
}
delete pipeline;
pipeline = nullptr;
disconnection.SetValue(true);
return true;
}
void HandleRequestCompleted(const Future<bool> &ret, const Request &request)
{
if (ret.IsError()) {
BUSLOG_WARN("Request send failed,conSeq={}", connectSeq);
SendResult.SetFailed(static_cast<int32_t>(CONNECTION_REFUSED));
(void)HandleDisconnect(static_cast<int>(CONNECTION_REFUSED));
} else {
if (timeOut) {
responseTimer = AsyncAfter(request.timeout.IsSome() ? request.timeout.Get() : g_requestTimeout,
GetAID(), &HttpConnectionActor::HandleRequestTimeOut);
}
}
return;
}
void HandleRequestTimeOut()
{
BUSLOG_WARN("Launch request timeout,conSeq={}", connectSeq);
HttpClient::GetInstance()
->Disconnect(connectSeq)
.Then(Defer(GetAID(), &HttpConnectionActor::HandleDisconnect, CONNECTION_TIMEOUT));
return;
}
protected:
void Finalize() override
{
}
private:
int connectSeq;
Pipeline *pipeline;
Promise<bool> SendResult;
Promise<bool> disconnection;
URL connectUrl;
bool timeOut;
Timer responseTimer;
};
struct HttpConnect::HttpConnection {
HttpConnection(const int conSeq, const URL &url)
: conActorId(Spawn(std::make_shared<HttpConnectionActor>(conSeq, url)))
{
}
~HttpConnection()
{
try {
BUSLOG_DEBUG("HttpConnect is destroying, aid={}", std::string(conActorId));
Terminate(conActorId);
} catch (...) {
}
}
AID conActorId;
};
HttpConnect::HttpConnect(const int conSeq, const URL &url)
: connection(std::make_shared<HttpConnect::HttpConnection>(conSeq, url))
{
}
Future<HttpConnect> HttpConnect::ConnectEstablishedCallback(const int &conSeq, const URL &url)
{
if (conSeq >= 0) {
BUSLOG_DEBUG("Connect succeed, conSeq={}", conSeq);
return HttpConnect(conSeq, url);
} else {
int retCode = 0;
retCode = retCode - conSeq;
BUSLOG_WARN("Connect failed, errCode={}", retCode);
return Status(retCode);
}
}
Future<Response> HttpConnect::ConnectAndLaunchReqCallback(const int conSeq, const Request &request,
const bool timeOutFlag)
{
if (conSeq >= 0) {
BUSLOG_DEBUG("Connect succeed, conSeq={}", conSeq);
HttpConnect connect = HttpConnect(conSeq, request.url);
Future<Response> response = connect.LaunchRequest(request, timeOutFlag);
(void)connect.Disconnected().OnComplete([connect]() {});
return response;
} else {
int retCode = 0;
retCode = retCode - conSeq;
BUSLOG_WARN("Connect failed, errCode={}", retCode);
return Status(retCode);
}
}
void HttpConnect::ConnectClosedCallback(const int conSeq, const int errCode)
{
std::string actorName = "CONNECT_" + std::to_string(conSeq);
(void)Async(actorName, &HttpConnectionActor::HandleDisconnect, errCode);
return;
}
void HttpConnect::ResponseCompletedCallback(const int conSeq, Response *responsePtr)
{
std::string actorName = "CONNECT_" + std::to_string(conSeq);
Response responseObj = *responsePtr;
delete responsePtr;
Async(actorName, &HttpConnectionActor::HandleResponse, responseObj);
return;
}
Future<bool> HttpConnect::Disconnect() const
{
return Async(connection->conActorId, &HttpConnectionActor::Disconnect);
}
Future<bool> HttpConnect::Disconnected() const
{
return Async(connection->conActorId, &HttpConnectionActor::Disconnected);
}
Future<Response> HttpConnect::LaunchRequest(const Request &request, const bool &timeOutFlag)
{
return Async(connection->conActorId, &HttpConnectionActor::LaunchRequest, request, timeOutFlag);
}
Future<HttpConnect> Connect(const URL &url, litebus::Option<std::string> credencial)
{
if (url.ip.IsNone() || url.port.IsNone() || url.scheme.IsNone()) {
BUSLOG_WARN("Couldn't connect with no ip,port,scheme.");
return Status(static_cast<int32_t>(INVALID_REQUEST));
}
if (url.scheme.Get() != HTTPS_SCHEME && url.scheme.Get() != HTTP_SCHEME) {
BUSLOG_WARN("Only support 'http' and 'https'");
return Status(static_cast<int32_t>(INVALID_REQUEST));
}
#ifndef SSL_ENABLED
if (url.scheme.Get() == HTTPS_SCHEME) {
BUSLOG_WARN("Couldn't connect to url with 'https' while ssl isnot enable.");
return Status(static_cast<int32_t>(INVALID_REQUEST));
}
#endif
return HttpClient::GetInstance()->Connect(url, credencial).Then(
std::bind(&HttpConnect::ConnectEstablishedCallback, ::_1, url));
}
Future<Response> LaunchRequest(const Request &request)
{
if (!CheckReqUrl(request.url)) {
return Status(static_cast<int32_t>(INVALID_REQUEST));
}
if (request.keepAlive) {
BUSLOG_WARN("Couldn't create keep-alive request normally.");
return Status(static_cast<int32_t>(INVALID_REQUEST));
}
HeaderMap headerMap = request.headers;
auto conItem = headerMap.find("Connection");
if (headerMap.end() != conItem && headerMap["Connection"] != "close") {
BUSLOG_WARN("Only 'Connection:close' are allowed in headers.");
return Status(static_cast<int32_t>(INVALID_REQUEST));
}
if (!CheckReqType(request.method)) {
BUSLOG_WARN("Only 'POST GET PUT DELETE PATCH' are allowed.");
return Status(static_cast<int32_t>(INVALID_REQUEST));
}
BUSLOG_DEBUG("Launch request, ip:{},port:{},path:{}", request.url.ip.Get(), request.url.port.Get(),
request.url.path);
return HttpClient::GetInstance()
->Connect(request.url, request.credential)
.Then(std::bind(&HttpConnect::ConnectAndLaunchReqCallback, ::_1, request, true));
}
Future<Response> LaunchRequest(const Request &request, const ResponseCallback &callback)
{
if (!callback) {
BUSLOG_WARN("Callback is empty");
return Status(static_cast<int32_t>(INVALID_REQUEST));
}
if (!CheckReqUrl(request.url)) {
return Status(static_cast<int32_t>(INVALID_REQUEST));
}
if (!CheckReqType(request.method)) {
BUSLOG_WARN("Only 'POST GET PUT DELETE PATCH' are allowed.");
return Status(static_cast<int32_t>(INVALID_REQUEST));
}
BUSLOG_DEBUG("Launch request, ip:{},port:{},path:{}", request.url.ip.Get(), request.url.port.Get(),
request.url.path);
return HttpClient::GetInstance()
->Connect(request.url, request.credential)
.Then(std::bind(&HttpClient::RegisterResponseCallBack, ::_1, callback))
.Then(std::bind(&HttpConnect::ConnectAndLaunchReqCallback, ::_1, request, false));
}
Future<Response> Post(const URL &url, const Option<std::unordered_map<std::string, std::string>> &headers,
const Option<std::string> &body, const Option<std::string> &contentType,
const Option<uint64_t> &reqTimeout)
{
if (body.IsNone() && contentType.IsSome()) {
BUSLOG_WARN("Couldn't create post request with a content-type but no body.");
return Status(static_cast<int32_t>(INVALID_REQUEST));
}
Request request("POST", false, url);
if (headers.IsSome()) {
for (const auto &headerItem : headers.Get()) {
request.headers[headerItem.first] = headerItem.second;
}
}
if (body.IsSome()) {
request.body = body.Get();
}
if (contentType.IsSome()) {
request.headers["Content-Type"] = contentType.Get();
}
if (reqTimeout.IsSome()) {
request.timeout = reqTimeout.Get();
} else {
request.timeout = g_requestTimeout;
}
return LaunchRequest(request);
}
Future<Response> Get(const URL &url, const Option<std::unordered_map<std::string, std::string>> &headers,
const Option<uint64_t> &reqTimeout)
{
Request request("GET", false, url);
if (headers.IsSome()) {
for (const auto &headerItem : headers.Get()) {
request.headers[headerItem.first] = headerItem.second;
}
}
if (reqTimeout.IsSome()) {
request.timeout = reqTimeout.Get();
} else {
request.timeout = g_requestTimeout;
}
return LaunchRequest(request);
}
bool CheckReqType(const std::string &method)
{
return (ALLOW_METHOD.find(method) != ALLOW_METHOD.end());
}
bool CheckReqUrl(const URL &url)
{
if (url.scheme.IsNone()) {
BUSLOG_WARN("Couldn't create http request with no scheme.");
return false;
}
if (url.scheme.Get() != HTTPS_SCHEME && url.scheme.Get() != HTTP_SCHEME) {
BUSLOG_WARN("Only support 'http' and 'https'");
return false;
}
#ifndef SSL_ENABLED
if (url.scheme.Get() == HTTPS_SCHEME) {
BUSLOG_WARN("Couldn't create http request with 'https' while ssl isnot enable.");
return false;
}
#endif
if (url.ip.IsNone()) {
BUSLOG_WARN("Couldn't create http request with no ip.");
return false;
}
if (url.port.IsNone()) {
BUSLOG_WARN("Couldn't create http request with no port.");
return false;
}
return true;
}
std::string GetHttpError(int32_t httpErrCode)
{
BUSLOG_WARN("Http error:{}", httpErrCode);
if (httpErrCode == 0) {
return std::string("Unknown error.");
}
return litebus::os::Strerror(httpErrCode);
}
void SetHttpRequestTimeOut(uint64_t duration)
{
BUSLOG_INFO("Http timeout:{}", duration);
g_requestTimeout = duration;
}
}
}