/src/gpac/src/utils/rmt_ws.c

Source
#include <gpac/tools.h>
#include <gpac/thread.h>
#include <gpac/list.h>
#include <gpac/bitstream.h>
#include <gpac/network.h>
#include <gpac/download.h>
#include <gpac/base_coding.h>

#if !defined(GPAC_DISABLE_RMTWS) && !defined(GPAC_DISABLE_NETWORK)


static const char websocket_guid[] = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11";


struct RMT_WS {

    // The main server thread
    GF_Thread* thread;

    // Should we stop the thread?
    Bool should_stop;

    RMT_Settings settings;

};



GF_DownloadSession *gf_dm_sess_new_server(GF_DownloadManager *dm, GF_Socket *server, void *ssl_ctx, gf_dm_user_io user_io, void *usr_cbk, Bool async, GF_Err *e);
void  gf_dm_sess_set_header(GF_DownloadSession *sess, const char *name, const char *value);
GF_Err gf_dm_sess_send_reply(GF_DownloadSession *sess, u32 reply_code, const char *response_body, u32 body_len, Bool no_body);
void gf_dm_sess_clear_headers(GF_DownloadSession *sess);
void  gf_dm_sess_set_header(GF_DownloadSession *sess, const char *name, const char *value);
GF_Err dm_sess_write(GF_DownloadSession *session, const u8 *buffer, u32 size);
GF_Err gf_dm_read_data(GF_DownloadSession *sess, char *data, u32 data_size, u32 *out_read);
void httpout_format_date(u64 time, char szDate[200], Bool for_listing);

#ifdef GPAC_HAS_SSL

void *gf_ssl_new(void *ssl_server_ctx, GF_Socket *client_sock, GF_Err *e);
void *gf_ssl_server_context_new(const char *cert, const char *key);
void gf_ssl_server_context_del(void *ssl_server_ctx);
Bool gf_ssl_init_lib();

#endif

enum {
    RMT_WEBSOCKET_CONTINUATION = 0,
    RMT_WEBSOCKET_TEXT = 1,
    RMT_WEBSOCKET_BINARY = 2,
    RMT_WEBSOCKET_CLOSE = 8,
    RMT_WEBSOCKET_PING = 9,
    RMT_WEBSOCKET_PONG = 10,
};

static char RMT_WEBSOCKET_PING_MSG[2] = { 0x89, 0x00 };

struct __rmt_serverctx {

    RMT_WS* rmt;

    GF_Socket* server_sock;

    GF_SockGroup* sg;

    void *ssl_ctx;

    GF_DownloadManager* dm_sess;

    GF_List* active_clients;

};

struct __rmt_clientctx {

    GF_Socket* client_sock;
    RMT_ServerCtx* ctx;
    GF_DownloadSession* http_sess;

    char peer_address[GF_MAX_IP_NAME_LEN + 16];
    char buffer[1024];

    Bool is_ws;
    u64 last_active_time;
    u64 last_ping_time;
    Bool should_close;

    rmt_client_on_data_cbk on_data_cbk;
    void* on_data_cbk_task;

    rmt_client_on_del_cbk on_del_cbk;
    void* on_del_cbk_task;

};

GF_EXPORT
const char* gf_rmt_get_peer_address(RMT_ClientCtx* client) {
    if (client)
        return client->peer_address;
    return NULL;
}

GF_EXPORT
void* gf_rmt_client_get_on_data_task(RMT_ClientCtx* client) {
    if (client)
        return client->on_data_cbk_task;
    return NULL;
}

GF_EXPORT
void gf_rmt_set_on_new_client_cbk(RMT_WS* rmt, void *task, rmt_on_new_client_cbk cbk) {
    RMT_Settings *rmtcfg = gf_rmt_get_settings(rmt);
    if (rmtcfg) {
        rmtcfg->on_new_client_cbk = cbk;
        rmtcfg->on_new_client_cbk_task = task;
    }
}

GF_EXPORT
void* gf_rmt_get_on_new_client_task(RMT_WS* rmt) {
    if (!rmt) return NULL;
    return rmt->settings.on_new_client_cbk_task;
}

GF_EXPORT
void gf_rmt_client_set_on_del_cbk(RMT_ClientCtx* client, void* task, rmt_client_on_del_cbk cbk) {
    GF_LOG(GF_LOG_DEBUG, GF_LOG_RMTWS, ("gf_rmt_client_set_on_del_cbk client %p task %p cbk %p\n", client, task, cbk));
    if (client) {
        client->on_del_cbk = cbk;
        client->on_del_cbk_task = task;
    }
}

GF_EXPORT
void* gf_rmt_client_get_on_del_task(RMT_ClientCtx* client) {
    if (client)
        return client->on_del_cbk_task;
    return NULL;
}

GF_EXPORT
RMT_WS* gf_rmt_client_get_rmt(RMT_ClientCtx* client) {
    if (client && client->ctx)
        return client->ctx->rmt;
    return NULL;
}


void rmt_clientctx_del(RMT_ClientCtx* client) {

    if (!client) return;

    GF_LOG(GF_LOG_INFO, GF_LOG_RMTWS, ("[RMT] closing client %s\n", client->peer_address));
    GF_LOG(GF_LOG_DEBUG, GF_LOG_RMTWS, ("rmt_clientctx_del client %p del_cbk %p\n", client, client ? client->on_del_cbk : NULL));

    if (client->on_del_cbk) {
        client->on_del_cbk(client->on_del_cbk_task);
    }

    if (client->client_sock) {
        if (client->ctx && client->ctx->sg)
            gf_sk_group_unregister(client->ctx->sg, client->client_sock);

        if (!client->http_sess)
            gf_sk_del(client->client_sock); //socket deleted by gf_dm_sess_del() if http_sess has been created
        client->client_sock = NULL;
    }

    if (client->http_sess) {
        GF_LOG(GF_LOG_DEBUG, GF_LOG_RMTWS, ("deleting http_sess %p\n", client->http_sess));
        gf_dm_sess_del(client->http_sess);
        client->http_sess = NULL;
    }

    client->ctx = NULL;
    client->on_data_cbk = NULL;
    GF_LOG(GF_LOG_DEBUG, GF_LOG_RMTWS, ("%s:%d rmt_clientctx_del client %p task %p\n", __FILE__, __LINE__, client, client->on_data_cbk_task));

    if (client->on_data_cbk_task) gf_free(client->on_data_cbk_task);
    client->on_data_cbk_task = NULL;

    if (client->on_del_cbk_task) gf_free(client->on_del_cbk_task);
    client->on_del_cbk_task = NULL;

    gf_free(client);
}

void rmt_serverctx_reset(RMT_ServerCtx* ctx) {
  if (ctx->active_clients) {
    while (gf_list_count(ctx->active_clients)) {
      RMT_ClientCtx* client = gf_list_pop_back(ctx->active_clients);
            rmt_clientctx_del(client);
            client = NULL;
    }
    gf_list_del(ctx->active_clients);
    ctx->active_clients = NULL;
  }

    if (ctx->server_sock) {
        if (ctx->sg)
            gf_sk_group_unregister(ctx->sg, ctx->server_sock);

        gf_sk_del(ctx->server_sock);
        ctx->server_sock = NULL;
    }

    if (ctx->sg) { gf_sk_group_del(ctx->sg); ctx->sg = NULL;  }

    if (ctx->dm_sess) { gf_dm_del(ctx->dm_sess); ctx->dm_sess = NULL; }

    if (ctx->ssl_ctx) {
#ifdef GPAC_HAS_SSL
    gf_ssl_server_context_del(ctx->ssl_ctx);
#else
        gf_free(ctx->ssl_ctx);
#endif
        ctx->ssl_ctx = NULL;
    }


}

void rmt_close_client(RMT_ClientCtx* client) {
    if (!client) return;
    RMT_ServerCtx* ctx = client->ctx;
    if (ctx && ctx->active_clients) {
        gf_list_del_item(ctx->active_clients, client);
    }
    rmt_clientctx_del(client);
}

GF_Err rmt_send_reply(GF_DownloadSession* http_sess, int responseCode, char* response_body, char* content_type) {

        u32 body_size = 0;
        char szFmt[100];
        char szDate[200];

      httpout_format_date(gf_net_get_utc(), szDate, GF_FALSE);
      gf_dm_sess_set_header(http_sess, "Date", szDate);
        gf_dm_sess_set_header(http_sess, "Server", gf_gpac_version());

        if (response_body) {
            body_size = (u32) strlen(response_body);
            gf_dm_sess_set_header(http_sess, "Content-Type", content_type ? content_type : "text/html");
            sprintf(szFmt, "%d", body_size);
            gf_dm_sess_set_header(http_sess, "Content-Length", szFmt);

        }

        return gf_dm_sess_send_reply(http_sess, responseCode, response_body, response_body ? (u32) strlen(response_body) : 0, (response_body==NULL));

}

static void rmt_on_http_session_data(void *usr_cbk, GF_NETIO_Parameter *parameter) {

    RMT_ClientCtx* client_ctx = (RMT_ClientCtx*) usr_cbk;
    if (!client_ctx || !client_ctx->ctx) {
        GF_LOG(GF_LOG_ERROR, GF_LOG_RMTWS, ("[RMT] sess io on null session\n"));
        return;
    }
    RMT_Settings* rmt_settings = gf_rmt_get_settings(client_ctx->ctx->rmt);

    GF_LOG(GF_LOG_DEBUG, GF_LOG_RMTWS, ("rmt_on_http_session_data on peer %s param %p msg_type %d \nerror %s \ndata %p \nsize %d \nname %p \nvalue %p \nreply %d\n",
                    client_ctx->peer_address,
                    parameter,
                    parameter->msg_type,
                    gf_error_to_string(parameter->error),
                    parameter->data,
                    parameter->size,
                    parameter->name,
                    parameter->value,
                    parameter->reply
            ));

    client_ctx->last_active_time = gf_sys_clock_high_res();

    const char* durl = gf_dm_sess_get_resource_name(parameter->sess);
    const char* ua = gf_dm_sess_get_header(parameter->sess, "User-Agent");
    GF_LOG(GF_LOG_DEBUG, GF_LOG_RMTWS, ("requested url: %s with ua %s\n", durl, ua));
    GF_Err e;

    if (parameter->msg_type != GF_NETIO_PARSE_REPLY) {
        if (parameter->size) {
            strncat(client_ctx->buffer, parameter->data, parameter->size);
            GF_LOG(GF_LOG_DEBUG, GF_LOG_RMTWS, ("session data is now: %s\n", client_ctx->buffer));
        }
    }

    if (parameter->msg_type == GF_NETIO_PARSE_REPLY) {

        GF_LOG(GF_LOG_DEBUG, GF_LOG_RMTWS, ("final buffer before response: %s\n", client_ctx->buffer));

        GF_DownloadSession* http_sess = parameter->sess ;

        if (!client_ctx->is_ws) {
            const char* ws_header_version = gf_dm_sess_get_header(http_sess, "Sec-WebSocket-Version");
            const char* ws_header_key = gf_dm_sess_get_header(http_sess, "Sec-WebSocket-Key");

            if (!ws_header_version || strcmp(ws_header_version, "13") || !ws_header_key) {

                gf_dm_sess_clear_headers(http_sess);
                gf_dm_sess_set_header(http_sess, "Connection", "close");

                rmt_send_reply(http_sess, 404, "only ws connections are accepted", NULL);
                client_ctx->should_close = GF_TRUE;
                return;

            }
            else {

                char* resp_key = gf_strdup(ws_header_key);
                gf_dynstrcat(&resp_key, websocket_guid, NULL);

                u32 resp_key_len = (u32) strlen(resp_key);
                if (resp_key_len < 1)
                    return;


                u8 hash[GF_SHA1_DIGEST_SIZE];
                gf_sha1_csum( resp_key, resp_key_len, hash );

                u32 end_b64 = gf_base64_encode(hash, GF_SHA1_DIGEST_SIZE, resp_key, resp_key_len);
                resp_key[resp_key_len-1] = 0;
                if (end_b64 < resp_key_len) {
                    resp_key[end_b64] = 0;
                } else {
                    return;
                }

              gf_dm_sess_clear_headers(http_sess);

              gf_dm_sess_set_header(http_sess, "Connection", "Upgrade");
                gf_dm_sess_set_header(http_sess, "Upgrade", "websocket");
                gf_dm_sess_set_header(http_sess, "Sec-WebSocket-Accept", resp_key);

                e = rmt_send_reply(http_sess, 101, NULL, NULL);

                if (e==GF_OK) {
                    client_ctx->is_ws = GF_TRUE;

                    if (rmt_settings->on_new_client_cbk && rmt_settings->on_new_client_cbk_task) {
                        GF_LOG(GF_LOG_DEBUG, GF_LOG_RMTWS, ("%s:%d calling on new client cb %p with client %p\n", __FILE__,__LINE__, rmt_settings->on_new_client_cbk, client_ctx));
                        rmt_settings->on_new_client_cbk(rmt_settings->on_new_client_cbk_task, client_ctx);
                    }
                }
                gf_free(resp_key);
            }
        } else {
            GF_LOG(GF_LOG_WARNING, GF_LOG_RMTWS, ("this should be a ws message right?\n"));
        }



        memset(&client_ctx->buffer, 0, 1024);
    }


}


GF_Err rmt_create_server(RMT_ServerCtx* ctx) {

    GF_Err e = GF_OK;

    rmt_serverctx_reset(ctx);

    RMT_Settings* rmt_settings = gf_rmt_get_settings(ctx->rmt);

    if (rmt_settings->cert && rmt_settings->pkey) {
#ifdef GPAC_HAS_SSL
    if (!gf_file_exists(rmt_settings->cert)) {
      GF_LOG(GF_LOG_ERROR, GF_LOG_RMTWS, ("[RMT] Certificate file %s not found\n", rmt_settings->cert));
      return GF_IO_ERR;
    }
    if (!gf_file_exists(rmt_settings->pkey)) {
      GF_LOG(GF_LOG_ERROR, GF_LOG_RMTWS, ("[RMT] Private key file %s not found\n", rmt_settings->pkey));
      return GF_IO_ERR;
    }
    if (gf_ssl_init_lib()) {
      GF_LOG(GF_LOG_ERROR, GF_LOG_RMTWS, ("[RMT] Failed to initialize OpenSSL library\n"));
      return GF_IO_ERR;
    }

        Bool prev_noh2 = gf_opts_get_bool("core", "no-h2");
        if (!prev_noh2)
            gf_opts_set_key("core", "no-h2", "1");

        ctx->ssl_ctx = gf_ssl_server_context_new(rmt_settings->cert, rmt_settings->pkey);

        if (!prev_noh2)
            gf_opts_set_key("core", "no-h2", "0");

    if (!ctx->ssl_ctx) return GF_IO_ERR;
#else
    GF_LOG(GF_LOG_ERROR, GF_LOG_RMTWS, ("[RMT] TLS key/certificate set but GPAC compiled without TLS support\n"));
    return GF_NOT_SUPPORTED;

#endif
    }

    ctx->sg = gf_sk_group_new();

    ctx->server_sock = gf_sk_new(GF_SOCK_TYPE_TCP);
    e = gf_sk_bind( ctx->server_sock,
                    rmt_settings->limit_connections_to_localhost ? "127.0.0.1" : "0.0.0.0",
                    rmt_settings->port,
                    NULL, 0, GF_SOCK_REUSE_PORT );

    if (!e) e = gf_sk_listen(ctx->server_sock, 0);
    if (e) {
        GF_LOG(GF_LOG_ERROR, GF_LOG_RMTWS, ("[RMT] failed to start server on port %d: %s\n", rmt_settings->port, gf_error_to_string(e) ));
        return e;
    }

    gf_sk_group_register(ctx->sg, ctx->server_sock);

    gf_sk_server_mode(ctx->server_sock, GF_TRUE);
    GF_LOG(GF_LOG_INFO, GF_LOG_RMTWS, ("[RMT] Server running on port %d\n", rmt_settings->port));

    Bool prev_noh2 = gf_opts_get_bool("core", "no-h2");
    if (!prev_noh2)
        gf_opts_set_key("core", "no-h2", "1");
    ctx->dm_sess = gf_dm_new(NULL);
    if (!prev_noh2)
        gf_opts_set_key("core", "no-h2", "0");
    ctx->active_clients = gf_list_new();

    return e;

}

GF_EXPORT
void gf_rmt_client_set_on_data_cbk(RMT_ClientCtx* client, void* task, rmt_client_on_data_cbk cbk) {
    if (client) {
        client->on_data_cbk = cbk;
        client->on_data_cbk_task = task;
    }
}

GF_Err rmt_server_handle_new_client(RMT_ServerCtx* ctx) {
    GF_Err e = GF_OK;

    void *ssl_c = NULL;

    RMT_ClientCtx* new_client;
    GF_SAFEALLOC(new_client, RMT_ClientCtx);

    new_client->ctx = ctx;

    e = gf_sk_accept(ctx->server_sock, &new_client->client_sock);
    gf_sk_group_register(ctx->sg, new_client->client_sock);

    u32 port;
    gf_sk_get_remote_address_port(new_client->client_sock, new_client->peer_address, &port);
    sprintf(new_client->peer_address + strlen(new_client->peer_address), ":%d", port); //TDOO: size check
    GF_LOG(GF_LOG_INFO, GF_LOG_RMTWS, ("[RMT] connected to remote peer %s\n",  new_client->peer_address));


#ifdef GPAC_HAS_SSL
  if (ctx->ssl_ctx) {
    ssl_c = gf_ssl_new(ctx->ssl_ctx, new_client->client_sock, &e);
    if (e) {
      GF_LOG(GF_LOG_ERROR, GF_LOG_RMTWS, ("[RMT] Failed to create TLS session from %s: %s\n", new_client->peer_address, gf_error_to_string(e) ));
      rmt_clientctx_del(new_client);
      return e;
    }
  }
#endif

    new_client->http_sess = gf_dm_sess_new_server(ctx->dm_sess, new_client->client_sock, ssl_c, rmt_on_http_session_data, new_client, GF_TRUE, &e);

    new_client->is_ws = GF_FALSE;
    new_client->last_active_time = gf_sys_clock_high_res();
    new_client->last_ping_time = gf_sys_clock_high_res();

    gf_list_add(ctx->active_clients, new_client);
    GF_LOG(GF_LOG_DEBUG, GF_LOG_RMTWS, ("adding active client socket %p %s\n",  new_client, new_client->peer_address));

    return e;
}

GF_Err rmt_client_send_payload(RMT_ClientCtx* client, const u8* payload, u64 size, Bool is_binary) {

    GF_Err e = GF_OK;

    GF_BitStream* respbs = gf_bs_new(NULL, size+10, GF_BITSTREAM_WRITE_DYN);
    gf_bs_write_int(respbs, 1, 1); //FIN=1
    gf_bs_write_int(respbs, 0, 3); //RSV=0
    gf_bs_write_int(respbs, is_binary ? RMT_WEBSOCKET_BINARY : RMT_WEBSOCKET_TEXT, 4); //opcode=text
    gf_bs_write_int(respbs, 0, 1); //masked=0

    if (size < 126)
        gf_bs_write_int(respbs, (s32)size, 7);
    else if (size < 65536) {
        gf_bs_write_int(respbs, 126, 7);
        gf_bs_write_long_int(respbs, size, 16);
    } else {
        gf_bs_write_long_int(respbs, 127, 7);
        gf_bs_write_long_int(respbs, size, 64);
    }

    gf_bs_write_data(respbs, payload, (u32)size);

    u8* respbuf=NULL;
    u32 respsize=0;
    gf_bs_get_content(respbs, &respbuf, &respsize);
    GF_LOG(GF_LOG_DEBUG, GF_LOG_RMTWS, ("ready to send respbuf of size %d: %.*s\n", respsize, respsize, respbuf));
    e = dm_sess_write(client->http_sess, respbuf, respsize);

    gf_bs_del(respbs);
    gf_free(respbuf);

    return e;

}

GF_Err rmt_client_handle_ws_payload(RMT_ClientCtx* client, u8* payload, u64 size, Bool is_binary) {

    if (client->on_data_cbk && client->on_data_cbk_task) {
        client->on_data_cbk(client->on_data_cbk_task, payload, size, is_binary);
    }

    return GF_OK;

    // if (size)
    //     payload[0] = 'A';

    // return rmt_client_send_payload(client, payload, size, is_binary);

}

GF_Err rmt_client_handle_ws_frame(RMT_ClientCtx* client, GF_BitStream* bs) {

    GF_Err e = GF_OK;

    if (gf_bs_available(bs) < 2) return GF_IO_ERR;

    u32 FIN = gf_bs_read_int(bs, 1); //TODO: handle fragmented?
    /*u32 RSV =*/ gf_bs_read_int(bs, 3);
    u32 opcode = gf_bs_read_int(bs, 4);
    u32 masked = gf_bs_read_int(bs, 1);
    u64 payload_size = gf_bs_read_int(bs, 7);
    if (payload_size == 126) {
        if (gf_bs_available(bs) < 2) return GF_IO_ERR;
        payload_size = gf_bs_read_int(bs, 16);
    }
    else if (payload_size == 127) {
        if (gf_bs_available(bs) < 8) return GF_IO_ERR;
        payload_size = gf_bs_read_long_int(bs, 64);
    }
    if (gf_bs_available(bs) < 4) return GF_IO_ERR;
    char masking_key[4] = {0};
    if (masked) {
        gf_bs_read_data(bs, (u8*)&masking_key, 4);
    }

    u8* extra_payload = NULL;
    u32 extra_read = 0;

    if (payload_size + gf_bs_get_position(bs) > gf_bs_get_size(bs)) {
        u64 extra_size = payload_size + gf_bs_get_position(bs) - gf_bs_get_size(bs) ;
        GF_LOG(GF_LOG_DEBUG, GF_LOG_RMTWS, ("buffer too small for payload_size %llu bs_pos %u bs_size %u => extra_size %llu\n", payload_size, gf_bs_get_position(bs), gf_bs_get_size(bs), extra_size));
        extra_payload = gf_malloc( sizeof(u8) * extra_size );

        e = GF_OK;
        while (!e && extra_read < extra_size) {
            u32 new_read = 0;
            e = gf_dm_read_data(client->http_sess, extra_payload + extra_read, (u32)(extra_size-extra_read), &new_read);
            GF_LOG(GF_LOG_DEBUG, GF_LOG_RMTWS, ("extra gf_dm_read_data => %d e=%s\n",extra_read, gf_error_to_string(e)));
            extra_read += new_read;
        }
        GF_LOG(GF_LOG_DEBUG, GF_LOG_RMTWS, ("extra gf_dm_read_data => %d e=%s\n",extra_read, gf_error_to_string(e)));
    }

    u8* unmasked_payload = gf_malloc( payload_size * sizeof(u8) + 1); // add 1 to add null to get c string
    int i=0;
    for (i=0; i<payload_size && gf_bs_available(bs); i++) {
        unmasked_payload[i] = (u8) ( gf_bs_read_u8(bs) ^ masking_key[i%4] );
    }

    if (extra_payload) {
        for (u32 j=0; j<extra_read; j++) {
            unmasked_payload[i] = (u8) ( extra_payload[j] ^ masking_key[i%4] );
            i++;
        }
    }

    unmasked_payload[i] = 0;

    GF_LOG(GF_LOG_DEBUG, GF_LOG_RMTWS, ("parsed ws message: FIN %d \n opcode %d \n masked %d \n payload_size %d \n masking_key %.*s \n unmasked_payload \n",
        FIN,
        opcode,
        masked,
        payload_size,
        4, masking_key
    ));

    switch(opcode) {

        case RMT_WEBSOCKET_CLOSE:
            client->should_close = GF_TRUE;
            break;

        case RMT_WEBSOCKET_TEXT:
        case RMT_WEBSOCKET_BINARY:

            e = rmt_client_handle_ws_payload(client, unmasked_payload, payload_size, (opcode==RMT_WEBSOCKET_BINARY));

            break;


        case RMT_WEBSOCKET_PING:;

            GF_BitStream* respbs = gf_bs_new(NULL, 0, GF_BITSTREAM_WRITE_DYN);
            gf_bs_write_int(respbs, 1, 1); //FIN=1
            gf_bs_write_int(respbs, 0, 3); //RSV=0
            gf_bs_write_int(respbs, RMT_WEBSOCKET_PONG, 4); //opcode=pong
            gf_bs_write_int(respbs, 0, 1); //masked=0
            gf_bs_write_int(respbs, (s32)payload_size, 7);
            gf_bs_write_data(respbs, unmasked_payload, (u32)payload_size);

            u8* respbuf=NULL;
            u32 respsize=0;
            gf_bs_get_content(respbs, &respbuf, &respsize);
            GF_LOG(GF_LOG_DEBUG, GF_LOG_RMTWS, ("ready to send PONG respbuf of size %d: %.*s\n", respsize, respsize, respbuf));
            e = dm_sess_write(client->http_sess, respbuf, respsize);

            gf_bs_del(respbs);
            gf_free(respbuf);

            break;


        case RMT_WEBSOCKET_CONTINUATION:    // not supported yet
        case RMT_WEBSOCKET_PONG:            // last active timer already reset
        default:
            break;
    }


    gf_free(unmasked_payload);
    if (extra_payload)
        gf_free(extra_payload);

    return e;
}

GF_Err rmt_client_handle_event(RMT_ClientCtx* client) {
    GF_Err e = GF_OK;

    GF_LOG(GF_LOG_DEBUG, GF_LOG_RMTWS, ("select ready for socket %p %s\n",  client, client->peer_address));

    if (!client->is_ws) {
        e = gf_dm_sess_process(client->http_sess);
        GF_LOG(GF_LOG_DEBUG, GF_LOG_RMTWS, ("[RMT] gf_dm_sess_process: %s\n",  gf_error_to_string(e)));
    }
    else {
        u8 buffer[1024]; //TODO: what size here?? do something like extra_payload??
        u32 read = 0;
        e = gf_dm_read_data(client->http_sess, buffer, 1024, &read);
        GF_LOG(GF_LOG_DEBUG, GF_LOG_RMTWS, ("gf_dm_read_data => %d sHTTP %.*s\n", read, read, buffer));

        client->last_active_time = gf_sys_clock_high_res();

        GF_BitStream* bs = gf_bs_new(buffer, read, GF_BITSTREAM_READ);

        while (e==GF_OK && gf_bs_available(bs)) {

            e = rmt_client_handle_ws_frame(client, bs);

            GF_LOG(GF_LOG_DEBUG, GF_LOG_RMTWS, ("rmt_client_handle_ws_frame() returned %s bs has avail %d \n", gf_errno_str(e), gf_bs_available(bs)));

        }


        gf_bs_del(bs);


    }


    return e;
}

GF_Err rmt_client_send_ping(RMT_ClientCtx* client) {

    GF_Err e = dm_sess_write(client->http_sess, RMT_WEBSOCKET_PING_MSG, 2);
    return e;

}

GF_EXPORT
GF_Err gf_rmt_client_send_to_ws(RMT_ClientCtx* client, const char* msg, u64 size, Bool is_binary) {

    return rmt_client_send_payload(client, (const u8*) msg, size, is_binary);

}

Bool rmt_client_should_close(RMT_ClientCtx* client) {

    // check valid object
    if (!client || !client->client_sock || !client->ctx) {
        GF_LOG(GF_LOG_WARNING, GF_LOG_RMTWS, ("[RMT] weird dead session in rmt\n"));
        return GF_TRUE;
    }
    RMT_Settings* rmt_settings = gf_rmt_get_settings(client->ctx->rmt);

    if (client->should_close) {
        GF_LOG(GF_LOG_DEBUG, GF_LOG_RMTWS, ("Disconnected socket %p %s because request done\n",  client, client->peer_address));
        return GF_TRUE;
    }

    // check disconnection
    GF_Err e = gf_sk_probe(client->client_sock);
    if (e==GF_IP_CONNECTION_CLOSED) {
        GF_LOG(GF_LOG_DEBUG, GF_LOG_RMTWS, ("Disconnected socket %p %s\n",  client, client->peer_address));
        return GF_TRUE;
    }

    // check timeout
    if (rmt_settings->timeout_secs) {
        u32 diff_sec = (u32) (gf_sys_clock_high_res() - client->last_active_time)/1000000;
        if ( diff_sec > rmt_settings->timeout_secs ) {
            GF_LOG(GF_LOG_DEBUG, GF_LOG_RMTWS, ("Disconnected socket %p %s for timeout \n",  client, client->peer_address));
            return GF_TRUE;
        }
    }

    return GF_FALSE;

}

GF_Err rmt_server_wait_for_event(RMT_ServerCtx* ctx) {

    RMT_Settings* rmt_settings = gf_rmt_get_settings(ctx->rmt);

    GF_Err e = gf_sk_group_select(ctx->sg, 10, GF_SK_SELECT_BOTH);

    if (e==GF_OK) {

        // event on server socket = new client
        if (gf_sk_group_sock_is_set(ctx->sg, ctx->server_sock, GF_SK_SELECT_READ)) {

            e = rmt_server_handle_new_client(ctx);
            if (e)
                return e;

        }

        // event on one the active client
        else {

            // check active sessions
        u32 count = gf_list_count(ctx->active_clients);
        for (u32 i=0; i<count; i++) {
          RMT_ClientCtx* client = gf_list_get(ctx->active_clients, i);

                if (rmt_client_should_close(client)) {

                    gf_list_rem(ctx->active_clients, i);
                    i--;
                    count--;
                    rmt_clientctx_del(client);
                    client = NULL;
                    continue;
                }

                if (gf_sk_group_sock_is_set(client->ctx->sg, client->client_sock, GF_SK_SELECT_WRITE)) {

                    // check if we should send ping
                    if (rmt_settings->ping_secs) {

                        u32 diff_sec = (u32) (gf_sys_clock_high_res() - client->last_ping_time)/1000000;

                        if ( diff_sec > rmt_settings->ping_secs ) {

                            GF_LOG(GF_LOG_DEBUG, GF_LOG_RMTWS, ("Sending PING on client %p %s\n",  client, client->peer_address));
                            e = rmt_client_send_ping(client);
                            if (!e) {
                                client->last_ping_time = gf_sys_clock_high_res();
                            }
                            continue;
                        }
                    }

                }

                // handle received event
                if (gf_sk_group_sock_is_set(client->ctx->sg, client->client_sock, GF_SK_SELECT_READ)) {

                    e = rmt_client_handle_event(client);

                }
            }

        }

    }

    gf_sleep(rmt_settings->msSleepBetweenServerUpdates);
    return e;

}

static u32 rmt_ws_thread_main(void* par) {

    RMT_WS* rmt = (RMT_WS*)par;

    RMT_ServerCtx* ctx;
    GF_SAFEALLOC(ctx, RMT_ServerCtx);
    ctx->rmt = rmt;

    GF_Err e;

    while (!rmt->should_stop) {

        // create socket if not exist
        if (!ctx->server_sock) {

            e = rmt_create_server(ctx);
            if (e) {
                if (ctx) {
                    rmt_serverctx_reset(ctx);
                    gf_free(ctx);
                    ctx = NULL;
                }
                return e;
            }

            continue;
        }

        e = rmt_server_wait_for_event(ctx);

    }

    GF_LOG(GF_LOG_DEBUG, GF_LOG_RMTWS, ("[RMT] thread main request exit! should cleanup\n"));
    rmt_serverctx_reset(ctx);
    gf_free(ctx);
    ctx = NULL;
    return GF_OK;

}



RMT_Settings* gf_rmt_get_settings(RMT_WS* rmt)
{
    if (!rmt) return NULL;

    return &rmt->settings;
}


void rmt_ws_del(RMT_WS* rmt) {

    if (rmt && rmt->thread) {


        rmt->should_stop = GF_TRUE;
        GF_LOG(GF_LOG_DEBUG, GF_LOG_RMTWS, ("rmt_ws_del() calling stop...\n"));

        gf_th_stop(rmt->thread);

        GF_LOG(GF_LOG_DEBUG, GF_LOG_RMTWS, ("rmt_ws_del() stop returned thread status is now %d\n", gf_th_status(rmt->thread)));

        gf_th_del(rmt->thread);

        rmt->thread = NULL;

    }

    gf_free(rmt);
    rmt = NULL;

}

RMT_WS* rmt_ws_new() {
    RMT_WS* rmt = NULL;

    GF_SAFEALLOC(rmt, RMT_WS);
    if (!rmt) {
        GF_LOG(GF_LOG_ERROR, GF_LOG_RMTWS, ("[RMT_WS] unable init rmt thread\n"));
        return NULL;
    }

    rmt->settings.port = 6363;
    rmt->settings.timeout_secs = 20;
    rmt->settings.ping_secs = 7;

    rmt->settings.limit_connections_to_localhost = GF_FALSE;
    rmt->settings.msSleepBetweenServerUpdates = 50;

    rmt->settings.on_new_client_cbk = NULL;
    rmt->settings.on_new_client_cbk_task = NULL;

    rmt->settings.cert = NULL;
    rmt->settings.pkey = NULL;

    return rmt;
}

void rmt_ws_run(RMT_WS* rmt) {

    rmt->thread = gf_th_new("rmt_ws_main_th");
    rmt->should_stop = GF_FALSE;

    GF_Err e = gf_th_run(rmt->thread, rmt_ws_thread_main, rmt);
    if (e != GF_OK) {
        GF_LOG(GF_LOG_ERROR, GF_LOG_RMTWS, ("[RMT_WS] unable to start websocket thread: %s\n", gf_error_to_string(e)));
        rmt_ws_del(rmt);
        rmt = NULL;
    }

}

#else //GPAC_DISABLE_RMTWS

void gf_rmt_set_on_new_client_cbk(RMT_WS* rmt, void *task, rmt_on_new_client_cbk cbk) {

}
void* gf_rmt_get_on_new_client_task(RMT_WS* rmt) {
    return NULL;
}

const char* gf_rmt_get_peer_address(RMT_ClientCtx* client) {
    return NULL;
}
GF_Err gf_rmt_client_send_to_ws(RMT_ClientCtx* client, const char* msg, u64 size, Bool is_binary) {
    return GF_NOT_SUPPORTED;
}

void gf_rmt_client_set_on_data_cbk(RMT_ClientCtx* client, void* task, rmt_client_on_data_cbk cbk) {

}
void* gf_rmt_client_get_on_data_task(RMT_ClientCtx* client) {
    return NULL;
}

void gf_rmt_client_set_on_del_cbk(RMT_ClientCtx* client, void* task, rmt_client_on_del_cbk cbk) {

}
void* gf_rmt_client_get_on_del_task(RMT_ClientCtx* client) {
    return NULL;
}

RMT_WS* gf_rmt_client_get_rmt(RMT_ClientCtx* client) {
    return NULL;
}

#endif

Coverage Report

Created: 2026-05-30 07:38