Monitor.cpp

#include "Monitor.h"

#include <CxxPtr/GlibPtr.h>
#include <CxxPtr/libwebsocketsPtr.h>

#include "Signalling/WsServer.h"

#include "WebRTSP/Client/WsClient.h"

#include "RtStreaming/GstRtStreaming/GstClient.h"
#include "RtStreaming/GstRtStreaming/GstStreamingSource.h"

#include "Log.h"
#include "RecordSession.h"
#include "Session.h"
#include "UrlPlayer.h"
#include "OnvifPlayer.h"


static const auto Log = MonitorLog;
typedef std::map<std::string, std::unique_ptr<GstStreamingSource>> MountPoints;

enum {
    MIN_RECONNECT_TIMEOUT = 3, // seconds
    MAX_RECONNECT_TIMEOUT = 10, // seconds
};

static std::unique_ptr<WebRTCPeer>
CreatePeer(
    const Config* config,
    const std::string& uri)
{
    return nullptr;
}

static std::unique_ptr<WebRTCPeer>
CreateRecordPeer(
    const Config* config,
    const std::string& uri)
{
    return std::make_unique<GstClient>(
        config->videoOutput.showStats,
        config->videoOutput.sync);
}

static std::unique_ptr<rtsp::ServerSession> CreateServerSession(
    const Config* config,
    const rtsp::Session::SendRequest& sendRequest,
    const rtsp::Session::SendResponse& sendResponse)
{
    std::unique_ptr<RecordSession> session =
        std::make_unique<RecordSession>(
            config,
            std::bind(CreatePeer, config, std::placeholders::_1),
            std::bind(CreateRecordPeer, config, std::placeholders::_1),
            sendRequest, sendResponse);

    return session;
}

static void OnRecorderConnected(const std::string& uri)
{
    Log()->info("Recorder connected to \"{}\" streamer", uri);
}

static void OnRecorderDisconnected(const std::string& uri)
{
    Log()->info("Recorder disconnected from \"{}\" streamer", uri);
}


static std::unique_ptr<WebRTCPeer>
CreateClientPeer(const Config* config)
{
    return std::make_unique<GstClient>(
        config->videoOutput.showStats,
        config->videoOutput.sync);
}

static std::unique_ptr<rtsp::Session> CreateClientSession (
    const Config* config,
    const rtsp::Session::SendRequest& sendRequest,
    const rtsp::Session::SendResponse& sendResponse) noexcept
{
    return std::make_unique<Session>(
        config,
        std::bind(CreateClientPeer, config),
        sendRequest, sendResponse);
}

GSourcePtr reconnectTimeoutSourcePtr;

static void ClientDisconnected(client::WsClient& client)
{
    if(reconnectTimeoutSourcePtr) {
        Log()->warn("Trying to create new reconnect timout source while previous one is still active");
        return;
    }

    const unsigned reconnectTimeout =
        g_random_int_range(MIN_RECONNECT_TIMEOUT, MAX_RECONNECT_TIMEOUT + 1);
    Log()->info("Scheduling reconnect within {} seconds...", reconnectTimeout);
    reconnectTimeoutSourcePtr.reset(g_timeout_source_new_seconds(reconnectTimeout));
    GSource* timeoutSource = reconnectTimeoutSourcePtr.get();
    g_source_set_callback(timeoutSource,
        [] (gpointer userData) -> gboolean {
            reconnectTimeoutSourcePtr.reset();
            static_cast<client::WsClient*>(userData)->connect();
            return false;
        }, &client, nullptr);
    g_source_attach(timeoutSource, g_main_context_get_thread_default());
}

static void onUrlPlayerEos(UrlPlayer& player, const std::string& url)
{
    if(reconnectTimeoutSourcePtr) {
        Log()->warn("Trying to create new reconnect timout source while previous one is still active");
        return;
    }

    struct Data {
        UrlPlayer *const  player;
        const std::string url;
    };

    const unsigned reconnectTimeout =
        g_random_int_range(MIN_RECONNECT_TIMEOUT, MAX_RECONNECT_TIMEOUT + 1);
    Log()->info("Scheduling playback restart within {} seconds...", reconnectTimeout);
    reconnectTimeoutSourcePtr.reset(g_timeout_source_new_seconds(reconnectTimeout));
    GSource* timeoutSource = reconnectTimeoutSourcePtr.get();
    g_source_set_callback(
        timeoutSource,
        [] (gpointer userData) -> gboolean {
            reconnectTimeoutSourcePtr.reset();
            Data* data = static_cast<Data*>(userData);
            data->player->play(data->url);
            return false;
        },
        new Data { &player, url },
        [] (gpointer userData) { delete static_cast<Data*>(userData); });
    g_source_attach(timeoutSource, g_main_context_get_thread_default());
}

static void onOnvifPlayerEos(OnvifPlayer& player)
{
    if(reconnectTimeoutSourcePtr) {
        Log()->warn("Trying to create new reconnect timout source while previous one is still active");
        return;
    }

    const unsigned reconnectTimeout =
        g_random_int_range(MIN_RECONNECT_TIMEOUT, MAX_RECONNECT_TIMEOUT + 1);
    Log()->info("Scheduling playback restart within {} seconds...", reconnectTimeout);
    reconnectTimeoutSourcePtr.reset(g_timeout_source_new_seconds(reconnectTimeout));
    GSource* timeoutSource = reconnectTimeoutSourcePtr.get();
    g_source_set_callback(timeoutSource,
        [] (gpointer userData) -> gboolean {
            reconnectTimeoutSourcePtr.reset();
            static_cast<OnvifPlayer*>(userData)->play();
            return false;
        }, &player, nullptr);
    g_source_attach(timeoutSource, g_main_context_get_thread_default());
}

int MonitorMain(const Config& config)
{
    if(!config.source)
        return -1;

    GMainContextPtr contextPtr(g_main_context_new());
    GMainContext* context = contextPtr.get();
    g_main_context_push_thread_default(context);

    GMainLoopPtr loopPtr(g_main_loop_new(context, FALSE));
    GMainLoop* loop = loopPtr.get();

    if(config.source->type == StreamSource::Type::WebRTSP) {
        if(config.source->localServer) {
            lws_context_creation_info lwsInfo {};
            lwsInfo.gid = -1;
            lwsInfo.uid = -1;
            lwsInfo.options = LWS_SERVER_OPTION_EXPLICIT_VHOSTS;
        #if defined(LWS_WITH_GLIB)
            lwsInfo.options |= LWS_SERVER_OPTION_GLIB;
            lwsInfo.foreign_loops = reinterpret_cast<void**>(&loop);
        #endif

            LwsContextPtr lwsContextPtr(lws_create_context(&lwsInfo));
            lws_context* lwsContext = lwsContextPtr.get();

            signalling::WsServer server(
                *config.source->localServer,
                loop,
                std::bind(
                    CreateServerSession,
                    &config,
                    std::placeholders::_1,
                    std::placeholders::_2));

            if(server.init(lwsContext)) {
                g_main_loop_run(loop);
                return 0;
            }
        } else if(config.source->client) {
            client::WsClient client(
                config.source->client.value(),
                loop,
                std::bind(
                    CreateClientSession,
                    &config,
                    std::placeholders::_1,
                    std::placeholders::_2),
                ClientDisconnected);

            if(client.init()) {
                client.connect();
                g_main_loop_run(loop);
                return 0;
            }
        }
    } else if(config.source->type == StreamSource::Type::Url) {
        UrlPlayer player(
            config.videoOutput.showStats,
            config.videoOutput.sync,
            std::bind(
                onUrlPlayerEos,
                std::placeholders::_1,
                config.source->uri));
        player.play(config.source->uri);

        g_main_loop_run(loop);
        return 0;
    } else if(config.source->type == StreamSource::Type::Onvif) {
        gchar *scheme;
        gchar *user;
        gchar *password;
        gchar *authParams;
        gchar *host;
        gint port;
        gchar *path;
        gchar *query;
        gchar *fragment;

        GCharPtr urlPtr;
        GCharPtr userPtr(user);
        GCharPtr passwordPtr(password);
        if(g_uri_split_with_user(
            config.source->uri.c_str(),
            G_URI_FLAGS_HAS_PASSWORD,
            &scheme,
            &user,
            &password,
            &authParams,
            &host,
            &port,
            &path,
            &query,
            &fragment,
            nullptr))
        {
            GCharPtr schemePtr(scheme);
            userPtr.reset(user);
            passwordPtr.reset(password);
            GCharPtr authParamsPtr(authParams);
            GCharPtr hostPtr(host);
            GCharPtr pathPtr(path);
            GCharPtr queryPtr(query);
            GCharPtr fragmentPtr(fragment);
            GCharPtr urlPtr(
                g_uri_join(
                    G_URI_FLAGS_NONE,
                    scheme,
                    nullptr, // userinfo
                    host,
                    port,
                    path,
                    query,
                    fragment));

            OnvifPlayer player(
                urlPtr.get(),
                userPtr ? std::optional<std::string>(userPtr.get()) : std::optional<std::string>(),
                passwordPtr ? std::optional<std::string>(passwordPtr.get()) : std::optional<std::string>(),
                config.source->trackMotion,
                config.source->motionPreviewDuration,
                config.videoOutput.showStats,
                config.videoOutput.sync,
                onOnvifPlayerEos);
            player.play();

            g_main_loop_run(loop);
            return 0;
        }
    }

    return -1;
}