LibWebSocket – dwukierunkowa asynchroniczna komunikacja JS-C++

/*
 * lws-minimal-ws-server
 *
 * Written in 2010-2019 by Andy Green <andy@warmcat.com>
 *
 * This file is made available under the Creative Commons CC0 1.0
 * Universal Public Domain Dedication.
 *
 * This demonstrates the most minimal http server you can make with lws,
 * with an added websocket chat server using a ringbuffer.
 *
 * To keep it simple, it serves stuff in the subdirectory "./mount-origin" of
 * the directory it was started in.
 * You can change that by changing mount.origin.
 */

#include <libwebsockets.h>
#include <string.h>
#include <signal.h>
#include <thread>
#include <chrono>
#include <iostream>
#include <mutex>

#define LWS_PLUGIN_STATIC

using namespace std;
mutex mtx;

#include "protocol_lws_minimal.c"


static struct lws_protocols protocols[] = {
        { "http", lws_callback_http_dummy, 0, 0 },
        LWS_PLUGIN_PROTOCOL_MINIMAL,
        { NULL, NULL, 0, 0 } /* terminator */
};

static int interrupted;

static const struct lws_http_mount mount = {
        /* .mount_next */               NULL,           /* linked-list "next" */
        /* .mountpoint */               "/",            /* mountpoint URL */
        /* .origin */                   "./mount-origin", /* serve from dir */
        /* .def */                      "index.html",   /* default filename */
        /* .protocol */                 NULL,
        /* .cgienv */                   NULL,
        /* .extra_mimetypes */          NULL,
        /* .interpret */                NULL,
        /* .cgi_timeout */              0,
        /* .cache_max_age */            0,
        /* .auth_mask */                0,
        /* .cache_reusable */           0,
        /* .cache_revalidate */         0,
        /* .cache_intermediaries */     0,
        /* .origin_protocol */          LWSMPRO_FILE,   /* files in a dir */
        /* .mountpoint_len */           1,              /* char count */
        /* .basic_auth_login_file */    NULL,
};

void sigint_handler(int sig)
{
        interrupted = 1;
}

void watek()
{
                const char *napis ="heya"; //komunikat do wysłania
                std::this_thread::sleep_for(std::chrono::seconds(7)); //czekam na włączenie klientów

        cerr << "wstart"; //start wątku
        while (!interrupted) // przerywamy ctrl+c
        {
                struct msg amsg; //wiadomość do wysłania
                amsg.len=4; //długość wiadomości do wysłania
                amsg.payload = malloc(LWS_PRE + 5); //przydzielenie pamięci na wiadomość do wysłania
                memcpy((char *)amsg.payload + LWS_PRE, napis, 4); //skopiowanie wiadomości do wysłania ze stałej tekstowej do zmiennej
        mtx.lock(); //semafor v - blokada przerwania obsługującego przychodzące wiadomości
      int n = (int)lws_ring_get_count_free_elements(wskaznik->ring);
               
if (!n) {
                        /* forcibly make space */
                        cull_lagging_clients(wskaznik);
                        n = (int)lws_ring_get_count_free_elements(wskaznik->ring);
                }
                //if (!n)
                  //      break;

                if (!n || !lws_ring_insert(wskaznik->ring, &amsg, 1)) {
                        __minimal_destroy_message(&amsg);
                        lwsl_user("dropping!\n");
                        mtx.unlock();
                        return;
                }

                //makro: pętla - dla każdego podłączonego klienta wywołaj przerwanie z sygnałem WRITABLE (tj. jest coś do wysłania)
                //ppss to lokalna zmienna typu struct per_per_session_data_minimal**
                lws_start_foreach_llp(struct per_session_data__minimal **,
                                      ppss, wskaznik->pss_list) {
                        lws_callback_on_writable((*ppss)->wsi);
                } lws_end_foreach_llp(ppss, pss_list);
                mtx.unlock(); //semafor ^ odblokowanie przerwania
                cerr <<"SENDED EXTRA MESSAGE TO ALL CLIENTS (heya)!\n";
                std::this_thread::sleep_for(std::chrono::seconds(5)); //5 sekund przerwy i od nowa.
        }
        cerr << "sstop"; //koniec wątku
}


int main(int argc, const char **argv)
{

        std::thread drugiwatek(watek);

        struct lws_context_creation_info info;
        struct lws_context *context;
        const char *p;
        int n = 0, logs = LLL_USER | LLL_ERR | LLL_WARN | LLL_NOTICE
                        /* for LLL_ verbosity above NOTICE to be built into lws,
                         * lws must have been configured and built with
                         * -DCMAKE_BUILD_TYPE=DEBUG instead of =RELEASE */
                        /* | LLL_INFO */ /* | LLL_PARSER */ /* | LLL_HEADER */
                        /* | LLL_EXT */ /* | LLL_CLIENT */ /* | LLL_LATENCY */
                        /* | LLL_DEBUG */;

        signal(SIGINT, sigint_handler);

        if ((p = lws_cmdline_option(argc, argv, "-d")))
                logs = atoi(p);

        lws_set_log_level(logs, NULL);
        lwsl_user("LWS minimal ws server (lws_ring) | visit http://localhost:7681\n");

        memset(&info, 0, sizeof info); /* otherwise uninitialized garbage */
        info.port = 7681;
        info.mounts = &mount;
        info.protocols = protocols;
        info.options =
                LWS_SERVER_OPTION_HTTP_HEADERS_SECURITY_BEST_PRACTICES_ENFORCE;

        context = lws_create_context(&info);
        if (!context) {
                lwsl_err("lws init failed\n");
                return 1;
        }

        while (n >= 0 && !interrupted)
                {
                        n = lws_service(context, 0);
                         std::this_thread::sleep_for(std::chrono::milliseconds(30));

                }
        lws_context_destroy(context);
        drugiwatek.join();
        return 0;
}

/*
 * ws protocol handler plugin for "lws-minimal"
 *
 * Written in 2010-2019 by Andy Green <andy@warmcat.com>
 *
 * This file is made available under the Creative Commons CC0 1.0
 * Universal Public Domain Dedication.
 *
 * This version uses an lws_ring ringbuffer to cache up to 8 messages at a time,
 * so it's not so easy to lose messages.
 *
 * This also demonstrates how to "cull", ie, kill, connections that can't
 * keep up for some reason.
 */

#if !defined (LWS_PLUGIN_STATIC)
#define LWS_DLL
#define LWS_INTERNAL
#include <libwebsockets.h>
#endif

#include <string.h>

/* one of these created for each message */

struct msg {
        void *payload; /* is malloc'd */
        size_t len;
};

/* one of these is created for each client connecting to us */

struct per_session_data__minimal {
        struct per_session_data__minimal *pss_list;
        struct lws *wsi;
        uint32_t tail;

        unsigned int culled:1;
};

/* one of these is created for each vhost our protocol is used with */

struct per_vhost_data__minimal {
        struct lws_context *context;
        struct lws_vhost *vhost;
        const struct lws_protocols *protocol;

        struct per_session_data__minimal *pss_list; /* linked-list of live pss*/

        struct lws_ring *ring; /* ringbuffer holding unsent messages */
};

static void
cull_lagging_clients(struct per_vhost_data__minimal *vhd)
{
        uint32_t oldest_tail = lws_ring_get_oldest_tail(vhd->ring);
        struct per_session_data__minimal *old_pss = NULL;
        int most = 0, before = lws_ring_get_count_waiting_elements(vhd->ring,
                                        &oldest_tail), m;

        /*
         * At least one guy with the oldest tail has lagged too far, filling
         * the ringbuffer with stuff waiting for them, while new stuff is
         * coming in, and they must close, freeing up ringbuffer entries.
         */

        lws_start_foreach_llp_safe(struct per_session_data__minimal **,
                              ppss, vhd->pss_list, pss_list) {

                if ((*ppss)->tail == oldest_tail) {
                        old_pss = *ppss;

                        lwsl_user("Killing lagging client %p\n", (*ppss)->wsi);

                        lws_set_timeout((*ppss)->wsi, PENDING_TIMEOUT_LAGGING,
                                        /*
                                         * we may kill the wsi we came in on,
                                         * so the actual close is deferred
                                         */
                                        LWS_TO_KILL_ASYNC);

                        /*
                         * We might try to write something before we get a
                         * chance to close.  But this pss is now detached
                         * from the ring buffer.  Mark this pss as culled so we
                         * don't try to do anything more with it.
                         */

                        (*ppss)->culled = 1;

                        /*
                         * Because we can't kill it synchronously, but we
                         * know it's closing momentarily and don't want its
                         * participation any more, remove its pss from the
                         * vhd pss list early.  (This is safe to repeat
                         * uselessly later in the close flow).
                         *
                         * Notice this changes *ppss!
                         */

                        lws_ll_fwd_remove(struct per_session_data__minimal,
                                          pss_list, (*ppss), vhd->pss_list);

                        /* use the changed *ppss so we won't skip anything */

                        continue;

                } else {
                        /*
                         * so this guy is a survivor of the cull.  Let's track
                         * what is the largest number of pending ring elements
                         * for any survivor.
                         */
                        m = lws_ring_get_count_waiting_elements(vhd->ring,
                                                        &((*ppss)->tail));
                        if (m > most)
                                most = m;
                }

        } lws_end_foreach_llp_safe(ppss);

        /* it would mean we lost track of oldest... but Coverity insists */
        if (!old_pss)
                return;

        /*
         * Let's recover (ie, free up) all the ring slots between the
         * original oldest's last one and the "worst" survivor.
         */

        lws_ring_consume_and_update_oldest_tail(vhd->ring,
                struct per_session_data__minimal, &old_pss->tail, before - most,
                vhd->pss_list, tail, pss_list);

        lwsl_user("%s: shrunk ring from %d to %d\n", __func__, before, most);
}

/* destroys the message when everyone has had a copy of it */

static void
__minimal_destroy_message(void *_msg)
{
        struct msg *msg = _msg;

        free(msg->payload);
        msg->payload = NULL;
        msg->len = 0;
}


struct per_vhost_data__minimal *wskaznik;

static int
callback_minimal(struct lws *wsi, enum lws_callback_reasons reason,
                        void *user, void *in, size_t len)
{
        mtx.lock();
        struct per_session_data__minimal *pss =
                        (struct per_session_data__minimal *)user;
        struct per_vhost_data__minimal *vhd =
                        (struct per_vhost_data__minimal *)
                        lws_protocol_vh_priv_get(lws_get_vhost(wsi),
                                        lws_get_protocol(wsi));
        wskaznik = vhd;//FX

        const struct msg *pmsg;
        struct msg amsg;
        int n, m;

        switch (reason) {
        case LWS_CALLBACK_PROTOCOL_INIT:
                vhd = lws_protocol_vh_priv_zalloc(lws_get_vhost(wsi),
                                lws_get_protocol(wsi),
                                sizeof(struct per_vhost_data__minimal));
                vhd->context = lws_get_context(wsi);
                vhd->protocol = lws_get_protocol(wsi);
                vhd->vhost = lws_get_vhost(wsi);

                vhd->ring = lws_ring_create(sizeof(struct msg), 8,
                                            __minimal_destroy_message);
                if (!vhd->ring)
                {
                        mtx.unlock();
                        return 1;
                }
                break;

        case LWS_CALLBACK_PROTOCOL_DESTROY:
                lws_ring_destroy(vhd->ring);
                break;

        case LWS_CALLBACK_ESTABLISHED:
                /* add ourselves to the list of live pss held in the vhd */
                lwsl_user("LWS_CALLBACK_ESTABLISHED: wsi %p\n", wsi);
                lws_ll_fwd_insert(pss, pss_list, vhd->pss_list);
                pss->tail = lws_ring_get_oldest_tail(vhd->ring);
                pss->wsi = wsi;
                break;

        case LWS_CALLBACK_CLOSED:
                lwsl_user("LWS_CALLBACK_CLOSED: wsi %p\n", wsi);
                /* remove our closing pss from the list of live pss */
                lws_ll_fwd_remove(struct per_session_data__minimal, pss_list,
                                  pss, vhd->pss_list);
                break;

        case LWS_CALLBACK_SERVER_WRITEABLE:
                if (pss->culled)
                        break;
                pmsg = lws_ring_get_element(vhd->ring, &pss->tail);
                if (!pmsg)
                        break;

                /* notice we allowed for LWS_PRE in the payload already */
                m = lws_write(wsi, ((unsigned char *)pmsg->payload) +
                              LWS_PRE, pmsg->len, LWS_WRITE_TEXT);
                if (m < (int)pmsg->len) {
                        lwsl_err("ERROR %d writing to ws socket\n", m);
                        mtx.unlock();
                        return -1;
                }

                lws_ring_consume_and_update_oldest_tail(
                        vhd->ring,      /* lws_ring object */
                        struct per_session_data__minimal, /* type of objects with tails */
                        &pss->tail,     /* tail of guy doing the consuming */
                        1,              /* number of payload objects being consumed */
                        vhd->pss_list,  /* head of list of objects with tails */
                        tail,           /* member name of tail in objects with tails */
                        pss_list        /* member name of next object in objects with tails */
                );

                /* more to do for us? */
                if (lws_ring_get_element(vhd->ring, &pss->tail))
                        /* come back as soon as we can write more */
                        lws_callback_on_writable(pss->wsi);
                break;

        case LWS_CALLBACK_RECEIVE:
                n = (int)lws_ring_get_count_free_elements(vhd->ring);
                if (!n) {
                        /* forcibly make space */
                        cull_lagging_clients(vhd);
                        n = (int)lws_ring_get_count_free_elements(vhd->ring);
                }
                if (!n)
                        break;

                lwsl_user("LWS_CALLBACK_RECEIVE: free space %d\n", n);

                amsg.len = len;
                /* notice we over-allocate by LWS_PRE... */
                amsg.payload = malloc(LWS_PRE + len);


                if (!amsg.payload) {
                        lwsl_user("OOM: dropping\n");
                        break;
                }

                /* ...and we copy the payload in at +LWS_PRE */
                memcpy((char *)amsg.payload + LWS_PRE, in, len);
                if (!lws_ring_insert(vhd->ring, &amsg, 1)) {
                        __minimal_destroy_message(&amsg);
                        lwsl_user("dropping!\n");
                        break;
                }

                /*
                 * let everybody know we want to write something on them
                 * as soon as they are ready
                 */
                lws_start_foreach_llp(struct per_session_data__minimal **,
                                      ppss, vhd->pss_list) {
                        lws_callback_on_writable((*ppss)->wsi);
                } lws_end_foreach_llp(ppss, pss_list);
                break;

        default:
                break;
        }
        mtx.unlock();
        return 0;
}

#define LWS_PLUGIN_PROTOCOL_MINIMAL \
        { \
                "lws-minimal", \
                callback_minimal, \
                sizeof(struct per_session_data__minimal), \
                0, \
                0, NULL, 0 \
        }

#if !defined (LWS_PLUGIN_STATIC)

/* boilerplate needed if we are built as a dynamic plugin */

static const struct lws_protocols protocols[] = {
        LWS_PLUGIN_PROTOCOL_MINIMAL
};

LWS_EXTERN LWS_VISIBLE int
init_protocol_minimal(struct lws_context *context,
                      struct lws_plugin_capability *c)
{
        if (c->api_magic != LWS_PLUGIN_API_MAGIC) {
                lwsl_err("Plugin API %d, library API %d", LWS_PLUGIN_API_MAGIC,
                         c->api_magic);
                return 1;
        }

        c->protocols = protocols;
        c->count_protocols = LWS_ARRAY_SIZE(protocols);
        c->extensions = NULL;
        c->count_extensions = 0;

        return 0;
}

LWS_EXTERN LWS_VISIBLE int
destroy_protocol_minimal(struct lws_context *context)
{
        return 0;
}
#endif