Program realizuje połączenia dwukierunkowe websocktets: JAVASCRIPT <-> C++ na dwóch osobnych wątkach – jeden wysyła co 5 sekund komunikat z serwera do wszystkich podłączonych klientów (autorska część), drugi odpowiada podłączonym klientom tym samym komunikatem, który od nich otrzymuje (przykład autorstwa twórców biblioteki).

Wątki są zsynchronizowane (mutex);

Jest to przetworzenie przykładu dołączanego do LibWebSockets w katalogu /minimal-examples/ws-server/minimal-ws-server-ring, należy uruchamiać te pliki w katalogu projektu:

kompiluj

1
2
#export LD_LIBRARY_PATH=/usr/local/lib:$LD_LIBRARY_PATH
g++ minimal-ws-server-ring.c  -lpthread -lwebsockets -fpermissive -std=c++11

minimal-ws-server-ring.c

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
/*
 * 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;
}

i drugi plik
protocol_lws_minimal.c

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
/*
 * 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

Komentowanie wyłączone.