main.cpp

#include <iostream>
#include <thread>
#include <coroutine>
#include <chrono>
#include <functional>
#include <exception>
#include <queue>
#include <memory>
#include <tuple>
#include <variant>

#include <fstream>
#include <iomanip>

#include "anio/alloc.h"
#include "anio/io_context.h"
#include <map>

#include "anio/link_list.h"

#include "anio/intrusive_ptr.h"
#include "anio/fixed_string.h"

#include "media/rtmp.h"
#include "media/http_flv.h"

#include "anio/http/http.h"

#define JSON_NOEXCEPTION
#include "json.hpp"

#include <filesystem>
#include <sstream>
#include <string>
#include <bitset>
#include <bitset>

struct acceptor {
  private:
    anio::stream_socket listen_;
    anio::endpoint ep;
    fixed_string<64> name_{};

  public:
    acceptor(anio::io_context* loop_, anio::endpoint endpoint_, const char* name)
        : listen_(loop_), ep(endpoint_), name_(name) {};

    auto accept(anio::endpoint& peer_ep) { return listen_.accept(peer_ep); }

    bool start() {
        std::error_code ec;
        listen_.assign(anio::socket_type::ipv4_tcp(ec));
        if (ec.value() != 0) {
            log_error("acceptor create listen socket, error {}", name_.c_str(), ec.message());
            return false;
        }
        listen_.set_option(anio::reuse_address(true));
        ec = anio::socket_ops::bind(listen_.native(), ep);
        if (ec.value() != 0) {
            log_error("{} acceptor bind error,{}", name_.c_str(), ec.message());
            return false;
        }
        ec = anio::socket_ops::listen(listen_.native(), 64);
        if (ec.value() != 0) {
            log_error("{} acceptor listen error,{}", name_.c_str(), ec.message());
            return false;
        }

        log_info("{} acceptor::listen() {}", name_.c_str(), ep.to_string());
        return true;
    }
};

// ffmpeg -re -i  /mnt/e/迅雷下载/s出g没.2023.mp4 -c copy -f flv
// rtmp://127.0.0.1:1935/test/aaa
//  ffmpeg -re -i E:/test.mp4 -c copy -f flv rtmp://127.0.0.1:1935/test/aaa
//  ffmpeg -re -i E:/迅雷下载/s出g没.2023.mp4 -c copy -f flv
//  rtmp://127.0.0.1:1935/test/aaa
class resource {
  public:
    int x = 123;
    resource(int x) : x(x) { std::cout << "resource()" << std::endl; }
    resource() { std::cout << "resource()" << std::endl; }
    ~resource() {
        x = 0;
        std::cout << "~resource()" << std::endl;
    }
};

void http_not_find(http::request& request, http::response& response) {
    response.set_status(http::status_404);
    response.set_content_type(http::content_type_html);
    response.content = "<html><body><h1>404 NOT_FOUND</h1></body></html>";
};

void http_test(http::request& request, http::response& response) {
    response.set_status(http::status_200);
    response.set_content_type(http::content_type_html);

    anio::buffer buf(1024 * 3);
    buf.write_string(
        "<html><style>body {font-family: \"Consolas\",  \"Courier "
        "New\", monospace;line-height: 1.6;}  </style></head><body>");
    buf.write_string("method:");
    buf.write_string(request.method);
    buf.write_string("<br>");
    buf.write_string("url:");
    buf.write_string(request.uri);
    buf.write_string("  version:");
    buf.write_string(request.version);
    buf.write_string("<br>");
    buf.write_string("<br>");
    buf.write_string("header::<br>");
    for (auto& header : request.headers) {
        buf.write_string(header.first);
        buf.write_string(":");
        buf.write_string(header.second);
        buf.write_string("<br>");
    }
    buf.write_string("<br>");
    buf.write_string("</body></html>");
    buf.write_string(http::http_crlf);
    buf.write_string(http::http_crlf);

    response.content.assign(reinterpret_cast<char*>(buf.read_begin()), buf.data_size());
};

void http_home(http::request& request, http::response& response) {
    response.set_status(http::status_302);
    response.set_content_type(http::content_type_html);
    response.headers[std::string(http::header_location)] = "/index.html";
}

bool http_filter(http::request& request, http::response& response) {
    std::string_view url(request.uri);
    if (url.starts_with("/live/")) {
        size_t flv_p = url.find(".flv");
        if (flv_p != std::string::npos) {
            size_t flv_s = url.rfind('/', flv_p - 1);
            if (flv_s != std::string::npos) {
                std::string_view name = url.substr(flv_s + 1, flv_p - flv_s - 1);

                media::media_stream_local* stream
                    = media::media_factory::instance().find_stream(name);

                if (stream != nullptr) {
                    auto* loop = request.conn.context();
                    anio::spawn(loop, media::live_flv::launch(std::move(request.conn), stream));
                    return true;
                } else {
                    response.set_status(http::status_204);
                    response.set_content_type(http::content_type_html);
                    response.content
                        = "<html><body><h1>not find flv "
                          "stream!</h1></body></html>";

                    return true;
                }
            }
        }
    }
    return false;
};

void http_livelist(http::request& request, http::response& response) {
    media::media_stream_local* head = media::media_factory::instance().stream_head();
    nlohmann::json json_;
    while (head != nullptr) {
        nlohmann::json j;
        j["stream_name"] = head->stream->stream_name;
        j["url"] = head->stream->url;
        j["have_video"] = head->stream->have_video();
        j["have_audio"] = head->stream->have_audio();
        json_.push_back(j);

        head = head->next();
    }

    std::string val = json_.dump();
    response.set_status(http::status_200);
    response.set_content_type(http::content_type_json);
    response.content.assign(val.c_str(), val.size());
};

void http_dump_alloc(http::request& request, http::response& response) {
    std::string val;
    anio::dump_allocinfo(val);
    response.set_status(http::status_200);
    response.set_content_type(http::content_type_html);
    response.content.append("<html><body><pre><code>");
    response.content.append(val.c_str(), val.size());
    response.content.append("</code></pre></html></body>");
};

auto task_hello() -> anio::task<std::tuple<int, std::string>> {
    log_info("task_hello~");
    co_return std::make_tuple(42, std::string("hello"));
};

// anio::task<void> echo_client(anio::stream_socket handle) {
//     anio::io_context *loop = co_await anio::await_executor{};
//     anio::stream_socket sock{loop, handle};
//
//     anio::buffer recv_buf{1024 * 2};
//     while (true) {
//         auto [error, len]
//             = co_await anio::async_recv(sock, recv_buf.write_begin(), recv_buf.write_cap());
//         if (!error && len <= 0) {
//             break;
//         }
//         recv_buf.commit_write(len);
//
//         auto [error1, len1]
//             = co_await anio::async_send(sock, recv_buf.read_begin(), recv_buf.data_size());
//         recv_buf.commit_read(len1);
//         recv_buf.fix();
//     }
//
//     co_return;
// };

auto task_connect() -> anio::task<void> {
    anio::io_context* loop = co_await anio::await_executor{};
    anio::ipv4_address ipv4_any;
    ipv4_any.from_string("127.0.0.1");

    std::error_code ec;
    anio::stream_socket connsock{loop, anio::socket_type::ipv4_tcp(ec)};
    if (ec.value() != 0) {
        log_info("socket create error {}", ec.message());
        co_return;
    }
    log_info("socket create error {}", ec.message());
    anio::endpoint server_ep(ipv4_any, 9095);
    ec = co_await connsock.connect(server_ep);
    if (ec.value() != 0) {
        log_info("async_connect error {}", ec.message());
        co_return;
    }

    anio::buffer recv_buf{1024 * 2};
    while (true) {
        auto [error, len] = co_await connsock.recv(recv_buf.write_begin(), recv_buf.write_cap());
        if (!error && len <= 0) {
            break;
        }
        recv_buf.commit_write(len);

        auto [error1, len1] = co_await connsock.send(recv_buf.read_begin(), recv_buf.data_size());
        recv_buf.commit_read(len1);
        recv_buf.fix();
    }

    log_info("async_connect down ~");
};

// auto task_void() -> anio::task<void> {
//     anio::io_context *loop = co_await anio::await_executor{};
//     // std::cout << "thread id:" << std::this_thread::get_id() << std::endl;
//     // anio::io_context *new_context = anio::context_pool::context();
//     // co_await anio::switch_context{new_context};
//     // std::cout << "thread id:" << std::this_thread::get_id() << std::endl;
//
//     auto [x, y] = co_await anio::spawn(anio::context_pool::context(), task_hello());
//     log_info("task_void~ {} {}", x, y);
//     anio::ipv4_address ipv4_any;
//     ipv4_any.from_string("::");
//     anio::endpoint ep_echo(ipv4_any, 9091);
//     acceptor echo_{loop, ep_echo, "[echo]"};
//
//     if (!echo_.start()) {
//         co_return;
//     }
//
//     while (true) {
//         auto [error, sock, remote_ep] = co_await echo_.accept();
//         log_info("echo_ accept~  ep {}", remote_ep.to_string());
//         anio::spawn(loop, echo_client(sock));
//     }
//
//     co_return;
// };

anio::task<void> rtmp_listen(uint16_t port) {
    auto* io_ctx = co_await anio::await_executor{};
    // rtmp server
    anio::ipv4_address ipv4_rtmp;
    ipv4_rtmp.from_string("::");
    anio::endpoint ep_rtmp(ipv4_rtmp, port);
    acceptor acceptor_{io_ctx, ep_rtmp, "[rtmp]"};
    if (!acceptor_.start()) {
        co_return;
    }

    while (true) {
        anio::endpoint peer_ep;
        //.with_timeout()
        auto [ec, sock] = co_await acceptor_.accept(peer_ep);
        if (ec.value() == 0 && sock.is_valid()) {
            log_info("[rtmp]new rtmp remote:{} fd {}", peer_ep.to_string(), sock.native());

            anio::spawn(anio::context_pool::context(), media::rtmp_conn::launch(sock));
        } else {
            log_error("rtmp accept error:{}", ec.message());
        }
    }
    log_critical("[rtmp] accept down!!");
    co_return;
};

anio::task<void> http_listen(http::http_routing& routing, uint16_t port) {
    auto* io_ctx = co_await anio::await_executor{};
    // http server
    anio::ipv4_address ipv4;
    ipv4.from_string("::");
    anio::endpoint ep(ipv4, port);
    acceptor acceptor_{io_ctx, ep, "[http]"};
    if (!acceptor_.start()) {
        co_return;
    }

    while (true) {
        anio::endpoint peer_ep;
        auto [ec, sock] = co_await acceptor_.accept(peer_ep);
        if (ec.value() == 0 && sock.is_valid()) {
            log_info("[http]new http remote:{} fd {}", peer_ep.to_string(), sock.native());
            anio::spawn(anio::context_pool::context(), http::http_launch(sock, routing));
        } else {
            auto ms
                = std::chrono::floor<std::chrono::milliseconds>(std::chrono::system_clock::now());
            log_error("http accept():{},time:{}", ec.message(),
                      std::format("{:%Y-%m-%d %H:%M:%S}\n", std::chrono::system_clock::now()));
        }
    }
    log_critical("[http] accept down!!");
    co_return;
};

int main() {
    auto& ctx_pool = anio::context_pool::instance();
    ctx_pool.init(std::min(std::thread::hardware_concurrency(), 2u));
    ctx_pool.context()->post([]() { log_info("media_live,hello~"); });

    media::media_factory::instance();
    // tick
    ctx_pool.for_echo([](anio::io_context* loop) {
        loop->register_tick(std::move(
            [](anio::io_context* loop_) { media::media_factory::instance().update_local(loop_); }));
    });

    // start rtmp
    anio::spawn(anio::context_pool::context(), rtmp_listen(media::rtmp_port));

    http::http_routing routing_{http_not_find};
    routing_.filter(http_filter);
    routing_.handler("/", http_home);
    routing_.handler("/test", http_test);
    routing_.handler("/live_list", http_livelist);  // json
    routing_.handler("/dump_alloc", http_dump_alloc);
    routing_.resource_dir("resource");

    // start http
    anio::spawn(anio::context_pool::context(), http_listen(routing_, 9099));

    std::string val;
    std::string line;
    while (true) {
        line.clear();
        std::getline(std::cin, line);
        if (line == "malloc") {
            val.clear();
            anio::dump_allocinfo(val);
            std::cout << val;
        }
    };

    ctx_pool.stop_join();

    return 0;
}