* 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 "http_pipeline_proxy.hpp"
#include <deque>
#include <memory>
#include <sstream>
#include <string>
#include <thread>
#include <vector>
#include "actor/buslog.hpp"
#include "http_iomgr.hpp"
#include "httpd/http.hpp"
#include "actor/actormgr.hpp"
#include "async/defer.hpp"
#include "async/future.hpp"
#include "evloop/evloop.hpp"
#include "iomgr/linkmgr.hpp"
#include "utils/os_utils.hpp"
using namespace std;
using std::placeholders::_1;
using namespace litebus;
using namespace http;
namespace litebus {
namespace http {
void HttpPipelineProxy::Process(const Request &request, const Future<Response> &responseFuture)
{
DataItem *item = new (std::nothrow) DataItem(request, responseFuture);
BUS_OOM_EXIT(item);
dataItem.push_back(item);
if (dataItem.size() == 1) {
HandleNextCallback();
}
}
void HttpPipelineProxy::HandleNextCallback()
{
if (dataItem.size() > 0) {
(void)dataItem.front()->responseFuture.OnComplete(
Defer(GetAID(), &HttpPipelineProxy::ReceiveHttpResponseCallback, ::_1, dataItem.front()->request));
}
}
void HttpPipelineProxy::ReceiveHttpResponseCallback(const Future<Response> &future, const Request &request)
{
BUSLOG_DEBUG("Handle response from application layer.");
if (dataItem.empty()) {
BUS_EXIT("Pipeline is empty.");
}
DataItem *item = dataItem.front();
if (future == item->responseFuture) {
bool result = HandleResponse(future, request);
dataItem.pop_front();
delete item;
item = nullptr;
if (result) {
HandleNextCallback();
}
} else {
BUS_EXIT("Pipeline is error.");
}
}
bool HttpPipelineProxy::HandleResponse(const Future<Response> &responseFuture, const Request &request) const
{
Response response = responseFuture.Get();
ostringstream output;
output << "HTTP/1.1 " << static_cast<int>(response.retCode) << " " << Response::GetStatusDescribe(response.retCode)
<< "\r\n";
auto iter = response.headers.begin();
while (iter != response.headers.end()) {
output << iter->first << ": " << iter->second << "\r\n";
++iter;
}
if (request.keepAlive) {
if (response.headers.find("Connection") != response.headers.end()) {
if (response.headers["Connection"] == "close") {
output << "Connection: close\r\n";
}
} else {
output << "Connection: Keep-Alive\r\n";
}
} else {
output << "Connection: close\r\n";
}
string body = response.body;
if (body.size() > 0) {
output << "Content-Length: " << body.size() << "\r\n";
output << "\r\n";
(void)output.write(body.data(), static_cast<std::streamsize>(body.size()));
} else {
output << "Content-Length: 0\r\n";
output << "\r\n";
}
shared_ptr<TCPMgr> tcpmgr = dynamic_pointer_cast<TCPMgr>(ActorMgr::GetIOMgrRef(string("tcp")));
if (tcpmgr != nullptr) {
MessageBase *msg = new (std::nothrow) MessageBase(MessageBase::Type::KHTTP);
BUS_OOM_EXIT(msg);
msg->body = output.str();
BUSLOG_DEBUG("Encode msg, request url,response code, body size, url:{},code:{},size:{}", request.url.path,
static_cast<std::underlying_type_t<ResponseCode>>(response.retCode), body.size());
(void)tcpmgr->Send(msg, connection, conSeq);
} else {
BUSLOG_ERROR("tcp protocol is not exist.");
}
return true;
}
}
}