plugins/protocol_lws_mirror.c

   1 /*
   2  * libwebsockets-test-server - libwebsockets test implementation
   3  *
   4  * Copyright (C) 2010-2016 Andy Green <andy@warmcat.com>
   5  *
   6  * This file is made available under the Creative Commons CC0 1.0
   7  * Universal Public Domain Dedication.
   8  *
   9  * The person who associated a work with this deed has dedicated
  10  * the work to the public domain by waiving all of his or her rights
  11  * to the work worldwide under copyright law, including all related
  12  * and neighboring rights, to the extent allowed by law. You can copy,
  13  * modify, distribute and perform the work, even for commercial purposes,
  14  * all without asking permission.
  15  *
  16  * The test apps are intended to be adapted for use in your code, which
  17  * may be proprietary.  So unlike the library itself, they are licensed
  18  * Public Domain.
  19  */
  20
  21 #if !defined (LWS_PLUGIN_STATIC)
  22 #define LWS_DLL
  23 #define LWS_INTERNAL
  24 #include "../lib/libwebsockets.h"
  25 #endif
  26
  27 #include <string.h>
  28 #include <stdlib.h>
  29
  30 /* lws-mirror_protocol */
  31
  32 #if defined(LWS_WITH_ESP8266)
  33 #define MAX_MESSAGE_QUEUE 64
  34 #else
  35 #define MAX_MESSAGE_QUEUE 512
  36 #endif
  37 struct per_session_data__lws_mirror {
  38         struct lws *wsi;
  39         int ringbuffer_tail;
  40 };
  41
  42 struct a_message {
  43         void *payload;
  44         size_t len;
  45 };
  46
  47 struct per_vhost_data__lws_mirror {
  48         struct a_message ringbuffer[MAX_MESSAGE_QUEUE];
  49         int ringbuffer_head;
  50 };
  51
  52 static int
  53 callback_lws_mirror(struct lws *wsi, enum lws_callback_reasons reason,
  54                     void *user, void *in, size_t len)
  55 {
  56         struct per_session_data__lws_mirror *pss =
  57                         (struct per_session_data__lws_mirror *)user;
  58         struct per_vhost_data__lws_mirror *v =
  59                         (struct per_vhost_data__lws_mirror *)
  60                         lws_protocol_vh_priv_get(lws_get_vhost(wsi),
  61                                         lws_get_protocol(wsi));
  62         int n, m;
  63
  64         switch (reason) {
  65
  66         case LWS_CALLBACK_ESTABLISHED:
  67                 lwsl_info("%s: LWS_CALLBACK_ESTABLISHED\n", __func__);
  68                 pss->ringbuffer_tail = v->ringbuffer_head;
  69                 pss->wsi = wsi;
  70                 break;
  71
  72         case LWS_CALLBACK_PROTOCOL_INIT: /* per vhost */
  73                 lws_protocol_vh_priv_zalloc(lws_get_vhost(wsi),
  74                                 lws_get_protocol(wsi),
  75                                 sizeof(struct per_vhost_data__lws_mirror));
  76                 break;
  77
  78         case LWS_CALLBACK_PROTOCOL_DESTROY: /* per vhost */
  79                 if (!v)
  80                         break;
  81                 lwsl_info("%s: mirror protocol cleaning up %p\n", __func__, v);
  82                 for (n = 0; n < ARRAY_SIZE(v->ringbuffer); n++)
  83                         if (v->ringbuffer[n].payload) {
  84                                 free(v->ringbuffer[n].payload);
  85                                 v->ringbuffer[n].payload = NULL;
  86                         }
  87                 break;
  88
  89         case LWS_CALLBACK_SERVER_WRITEABLE:
  90                 while (pss->ringbuffer_tail != v->ringbuffer_head) {
  91                         m = v->ringbuffer[pss->ringbuffer_tail].len;
  92                         n = lws_write(wsi, (unsigned char *)
  93                                    v->ringbuffer[pss->ringbuffer_tail].payload +
  94                                    LWS_PRE, m, LWS_WRITE_TEXT);
  95                         if (n < 0) {
  96                                 lwsl_err("ERROR %d writing to mirror socket\n", n);
  97                                 return -1;
  98                         }
  99                         if (n < m)
 100                                 lwsl_err("mirror partial write %d vs %d\n", n, m);
 101
 102                         if (pss->ringbuffer_tail == (MAX_MESSAGE_QUEUE - 1))
 103                                 pss->ringbuffer_tail = 0;
 104                         else
 105                                 pss->ringbuffer_tail++;
 106
 107                         if (((v->ringbuffer_head - pss->ringbuffer_tail) &
 108                             (MAX_MESSAGE_QUEUE - 1)) == (MAX_MESSAGE_QUEUE - 15))
 109                                 lws_rx_flow_allow_all_protocol(lws_get_context(wsi),
 110                                                lws_get_protocol(wsi));
 111
 112                         if (lws_send_pipe_choked(wsi)) {
 113                                 lws_callback_on_writable(wsi);
 114                                 break;
 115                         }
 116                 }
 117                 break;
 118
 119         case LWS_CALLBACK_RECEIVE:
 120                 if (((v->ringbuffer_head - pss->ringbuffer_tail) &
 121                     (MAX_MESSAGE_QUEUE - 1)) == (MAX_MESSAGE_QUEUE - 1)) {
 122                         lwsl_err("dropping!\n");
 123                         goto choke;
 124                 }
 125
 126                 if (v->ringbuffer[v->ringbuffer_head].payload)
 127                         free(v->ringbuffer[v->ringbuffer_head].payload);
 128
 129                 v->ringbuffer[v->ringbuffer_head].payload = malloc(LWS_PRE + len);
 130                 v->ringbuffer[v->ringbuffer_head].len = len;
 131                 memcpy((char *)v->ringbuffer[v->ringbuffer_head].payload +
 132                        LWS_PRE, in, len);
 133                 if (v->ringbuffer_head == (MAX_MESSAGE_QUEUE - 1))
 134                         v->ringbuffer_head = 0;
 135                 else
 136                         v->ringbuffer_head++;
 137
 138                 if (((v->ringbuffer_head - pss->ringbuffer_tail) &
 139                     (MAX_MESSAGE_QUEUE - 1)) != (MAX_MESSAGE_QUEUE - 2))
 140                         goto done;
 141
 142 choke:
 143                 lwsl_debug("LWS_CALLBACK_RECEIVE: throttling %p\n", wsi);
 144                 lws_rx_flow_control(wsi, 0);
 145
 146 done:
 147                 lws_callback_on_writable_all_protocol(lws_get_context(wsi),
 148                                                       lws_get_protocol(wsi));
 149                 break;
 150
 151         default:
 152                 break;
 153         }
 154
 155         return 0;
 156 }
 157
 158 #define LWS_PLUGIN_PROTOCOL_MIRROR { \
 159                 "lws-mirror-protocol", \
 160                 callback_lws_mirror, \
 161                 sizeof(struct per_session_data__lws_mirror), \
 162                 128, /* rx buf size must be >= permessage-deflate rx size */ \
 163         }
 164
 165 #if !defined (LWS_PLUGIN_STATIC)
 166
 167 static const struct lws_protocols protocols[] = {
 168         LWS_PLUGIN_PROTOCOL_MIRROR
 169 };
 170
 171 LWS_EXTERN LWS_VISIBLE int
 172 init_protocol_lws_mirror(struct lws_context *context,
 173                              struct lws_plugin_capability *c)
 174 {
 175         if (c->api_magic != LWS_PLUGIN_API_MAGIC) {
 176                 lwsl_err("Plugin API %d, library API %d", LWS_PLUGIN_API_MAGIC,
 177                          c->api_magic);
 178                 return 1;
 179         }
 180
 181         c->protocols = protocols;
 182         c->count_protocols = ARRAY_SIZE(protocols);
 183         c->extensions = NULL;
 184         c->count_extensions = 0;
 185
 186         return 0;
 187 }
 188
 189 LWS_EXTERN LWS_VISIBLE int
 190 destroy_protocol_lws_mirror(struct lws_context *context)
 191 {
 192         return 0;
 193 }
 194 #endif