#include <softbus_transport/TcpUplinkServer.h>

#include <softbus_core/logging/Logger.h>

#include <array>

namespace softbus::transport {

TcpUplinkServer::TcpUplinkServer(asio::io_context& io, std::string host, uint16_t port)
    : io_(io)
    , acceptor_(io, asio::ip::tcp::endpoint(asio::ip::make_address(host), port))
{
}

TcpUplinkServer::~TcpUplinkServer()
{
    stop();
}

void TcpUplinkServer::setRawPacketHandler(RawPacketHandler handler)
{
    rawHandler_ = std::move(handler);
}

void TcpUplinkServer::setWireMessageHandler(WireMessageHandler handler)
{
    wireHandler_ = std::move(handler);
}

void TcpUplinkServer::start()
{
    running_ = true;
    doAccept();
    SB_LOG_INFO("TcpUplinkServer", "Listening on ",
        acceptor_.local_endpoint().address().to_string(), ":",
        acceptor_.local_endpoint().port());
}

void TcpUplinkServer::stop()
{
    running_ = false;
    std::error_code ec;
    acceptor_.close(ec);
    if (clientSocket_) {
        clientSocket_->close(ec);
        clientSocket_.reset();
    }
}

bool TcpUplinkServer::sendCommand(const softbus::api::ExecuteCommand& command)
{
    if (!clientSocket_ || !clientSocket_->is_open()) {
        return false;
    }
    const auto line = encodeWireMessage(command);
    std::lock_guard<std::mutex> lock(sendMutex_);
    std::error_code ec;
    asio::write(*clientSocket_, asio::buffer(line), ec);
    return !ec;
}

void TcpUplinkServer::doAccept()
{
    if (!running_) {
        return;
    }
    auto socket = std::make_shared<asio::ip::tcp::socket>(io_);
    acceptor_.async_accept(*socket, [this, socket](const std::error_code& ec) {
        if (ec) {
            if (running_) {
                doAccept();
            }
            return;
        }
        clientSocket_ = socket;
        readBuffer_.clear();
        SB_LOG_INFO("TcpUplinkServer", "Edge connected");
        doRead(socket);
        doAccept();
    });
}

void TcpUplinkServer::doRead(const std::shared_ptr<asio::ip::tcp::socket> socket)
{
    auto buffer = std::make_shared<std::array<char, 4096>>();
    socket->async_read_some(asio::buffer(*buffer),
        [this, socket, buffer](const std::error_code& ec, std::size_t bytes) {
            if (ec) {
                SB_LOG_WARN("TcpUplinkServer", "Edge disconnected");
                if (clientSocket_ == socket) {
                    clientSocket_.reset();
                }
                return;
            }
            readBuffer_.append(buffer->data(), bytes);
            for (;;) {
                const auto pos = readBuffer_.find('\n');
                if (pos == std::string::npos) {
                    break;
                }
                const std::string line = readBuffer_.substr(0, pos);
                readBuffer_.erase(0, pos + 1);
                handleLine(line);
            }
            doRead(socket);
        });
}

void TcpUplinkServer::handleLine(const std::string& line)
{
    const auto msg = decodeWireMessage(line);
    if (!msg) {
        return;
    }
    if (wireHandler_) {
        wireHandler_(*msg);
    }
    if (msg->type == WireMessageType::RawPacket && rawHandler_) {
        rawHandler_(msg->rawPacket);
    }
}

} // namespace softbus::transport
