domo/domo-iot
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Add functional test Exemple

Browse Source
This commit is contained in:
NADAL Jean-Baptiste
2019-12-05 19:52:44 +01:00
parent 20e192462a
commit f0e80d657b
19 changed files with 2562 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
src/CMakeLists.txt
View File

@@ -6,6 +6,8 @@ set (CMAKE_MODULE_PATH "${MODULE_PATH}")
set (CMAKE_CXX_STANDARD 11)
include_directories(${CMAKE_SOURCE_DIR}/../nats.c/src)
include_directories(${CMAKE_SOURCE_DIR}/../libuv/include)
include_directories(${CMAKE_SOURCE_DIR}/build/libwebsockets/include)
include_directories(${CMAKE_SOURCE_DIR}/src)
#set(CMAKE_CXX_FLAGS "-Wall -Wextra -pedantic -Werror=strict-aliasing")
@@ -13,12 +15,7 @@ include_directories(${CMAKE_SOURCE_DIR}/src)
file(
GLOB_RECURSE
source_files
main.cpp
server/domo-server.cpp
broker/nats-broker.cpp
web/web-server.cpp
tinyweb/tinyweb.c
tinyweb/tools.c
main.c
)
add_executable (domo-iot ${source_files})
@@ -26,8 +23,8 @@ add_executable (domo-iot ${source_files})
target_link_libraries (domo-iot
LINK_PUBLIC
nats_static
json-c
uv
websockets
uv_a
rt
)

148
src/plugins/protocol_dumb_increment.c Normal file
View File

@@ -0,0 +1,148 @@
/*
* ws protocol handler plugin for "dumb increment"
*
* 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.
*
* The person who associated a work with this deed has dedicated
* the work to the public domain by waiving all of his or her rights
* to the work worldwide under copyright law, including all related
* and neighboring rights, to the extent allowed by law. You can copy,
* modify, distribute and perform the work, even for commercial purposes,
* all without asking permission.
*
* These test plugins are intended to be adapted for use in your code, which
* may be proprietary. So unlike the library itself, they are licensed
* Public Domain.
*/
#if !defined (LWS_PLUGIN_STATIC)
#define LWS_DLL
#define LWS_INTERNAL
#include <libwebsockets.h>
#endif
#include <string.h>
#define DUMB_PERIOD_US 50000
struct pss__dumb_increment {
int number;
};
struct vhd__dumb_increment {
const unsigned int *options;
};
static int
callback_dumb_increment(struct lws *wsi, enum lws_callback_reasons reason,
void *user, void *in, size_t len)
{
struct pss__dumb_increment *pss = (struct pss__dumb_increment *)user;
struct vhd__dumb_increment *vhd =
(struct vhd__dumb_increment *)
lws_protocol_vh_priv_get(lws_get_vhost(wsi),
lws_get_protocol(wsi));
uint8_t buf[LWS_PRE + 20], *p = &buf[LWS_PRE];
const struct lws_protocol_vhost_options *opt;
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 vhd__dumb_increment));
if (!vhd)
return -1;
if ((opt = lws_pvo_search(
(const struct lws_protocol_vhost_options *)in,
"options")))
vhd->options = (unsigned int *)opt->value;
break;
case LWS_CALLBACK_ESTABLISHED:
pss->number = 0;
if (!vhd->options || !((*vhd->options) & 1))
lws_set_timer_usecs(wsi, DUMB_PERIOD_US);
break;
case LWS_CALLBACK_SERVER_WRITEABLE:
n = lws_snprintf((char *)p, sizeof(buf) - LWS_PRE, "%d",
pss->number++);
m = lws_write(wsi, p, n, LWS_WRITE_TEXT);
if (m < n) {
lwsl_err("ERROR %d writing to di socket\n", n);
return -1;
}
break;
case LWS_CALLBACK_RECEIVE:
if (len < 6)
break;
if (strncmp((const char *)in, "reset\n", 6) == 0)
pss->number = 0;
if (strncmp((const char *)in, "closeme\n", 8) == 0) {
lwsl_notice("dumb_inc: closing as requested\n");
lws_close_reason(wsi, LWS_CLOSE_STATUS_GOINGAWAY,
(unsigned char *)"seeya", 5);
return -1;
}
break;
case LWS_CALLBACK_TIMER:
if (!vhd->options || !((*vhd->options) & 1)) {
lws_callback_on_writable_all_protocol_vhost(
lws_get_vhost(wsi), lws_get_protocol(wsi));
lws_set_timer_usecs(wsi, DUMB_PERIOD_US);
}
break;
default:
break;
}
return 0;
}
#define LWS_PLUGIN_PROTOCOL_DUMB_INCREMENT \
{ \
"dumb-increment-protocol", \
callback_dumb_increment, \
sizeof(struct pss__dumb_increment), \
10, /* rx buf size must be >= permessage-deflate rx size */ \
0, NULL, 0 \
}
#if !defined (LWS_PLUGIN_STATIC)
static const struct lws_protocols protocols[] = {
LWS_PLUGIN_PROTOCOL_DUMB_INCREMENT
};
LWS_EXTERN LWS_VISIBLE int
init_protocol_dumb_increment(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_dumb_increment(struct lws_context *context)
{
return 0;
}
#endif

497
src/plugins/protocol_lws_mirror.c Normal file
View File

@@ -0,0 +1,497 @@
/*
* libwebsockets-test-server - libwebsockets test implementation
*
* 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.
*
* The person who associated a work with this deed has dedicated
* the work to the public domain by waiving all of his or her rights
* to the work worldwide under copyright law, including all related
* and neighboring rights, to the extent allowed by law. You can copy,
* modify, distribute and perform the work, even for commercial purposes,
* all without asking permission.
*
* The test apps are intended to be adapted for use in your code, which
* may be proprietary. So unlike the library itself, they are licensed
* Public Domain.
*
* Notice that the lws_pthread... locking apis are all zero-footprint
* NOPs in the case LWS_MAX_SMP == 1, which is the default. When lws
* is built for multiple service threads though, they resolve to their
* pthreads equivalents.
*/
#if !defined (LWS_PLUGIN_STATIC)
#define LWS_DLL
#define LWS_INTERNAL
#include <libwebsockets.h>
#endif
#include <string.h>
#include <stdlib.h>
#define QUEUELEN 32
/* queue free space below this, rx flow is disabled */
#define RXFLOW_MIN (4)
/* queue free space above this, rx flow is enabled */
#define RXFLOW_MAX ((2 * QUEUELEN) / 3)
#define MAX_MIRROR_INSTANCES 3
struct mirror_instance;
struct per_session_data__lws_mirror {
struct lws *wsi;
struct mirror_instance *mi;
struct per_session_data__lws_mirror *same_mi_pss_list;
uint32_t tail;
};
/* this is the element in the ring */
struct a_message {
void *payload;
size_t len;
};
struct mirror_instance {
struct mirror_instance *next;
lws_pthread_mutex(lock) /* protects all mirror instance data */
struct per_session_data__lws_mirror *same_mi_pss_list;
/**< must hold the the per_vhost_data__lws_mirror.lock as well
* to change mi list membership */
struct lws_ring *ring;
int messages_allocated;
char name[30];
char rx_enabled;
};
struct per_vhost_data__lws_mirror {
lws_pthread_mutex(lock) /* protects mi_list membership changes */
struct mirror_instance *mi_list;
};
/* enable or disable rx from all connections to this mirror instance */
static void
__mirror_rxflow_instance(struct mirror_instance *mi, int enable)
{
lws_start_foreach_ll(struct per_session_data__lws_mirror *,
pss, mi->same_mi_pss_list) {
lws_rx_flow_control(pss->wsi, enable);
} lws_end_foreach_ll(pss, same_mi_pss_list);
mi->rx_enabled = enable;
}
/*
* Find out which connection to this mirror instance has the longest number
* of still unread elements in the ringbuffer and update the lws_ring "oldest
* tail" with it. Elements behind the "oldest tail" are freed and recycled for
* new head content. Elements after the "oldest tail" are still waiting to be
* read by somebody.
*
* If the oldest tail moved on from before, check if it created enough space
* in the queue to re-enable RX flow control for the mirror instance.
*
* Mark connections that are at the oldest tail as being on a 3s timeout to
* transmit something, otherwise the connection will be closed. Without this,
* a choked or nonresponsive connection can block the FIFO from freeing up any
* new space for new data.
*
* You can skip calling this if on your connection, before processing, the tail
* was not equal to the current worst, ie, if the tail you will work on is !=
* lws_ring_get_oldest_tail(ring) then no need to call this when the tail
* has changed; it wasn't the oldest so it won't change the oldest.
*
* Returns 0 if oldest unchanged or 1 if oldest changed from this call.
*/
static int
__mirror_update_worst_tail(struct mirror_instance *mi)
{
uint32_t wai, worst = 0, worst_tail = 0, oldest;
struct per_session_data__lws_mirror *worst_pss = NULL;
oldest = lws_ring_get_oldest_tail(mi->ring);
lws_start_foreach_ll(struct per_session_data__lws_mirror *,
pss, mi->same_mi_pss_list) {
wai = (uint32_t)lws_ring_get_count_waiting_elements(mi->ring,
&pss->tail);
if (wai >= worst) {
worst = wai;
worst_tail = pss->tail;
worst_pss = pss;
}
} lws_end_foreach_ll(pss, same_mi_pss_list);
if (!worst_pss)
return 0;
lws_ring_update_oldest_tail(mi->ring, worst_tail);
if (oldest == lws_ring_get_oldest_tail(mi->ring))
return 0;
/*
* The oldest tail did move on. Check if we should re-enable rx flow
* for the mirror instance since we made some space now.
*/
if (!mi->rx_enabled && /* rx is disabled */
lws_ring_get_count_free_elements(mi->ring) >= RXFLOW_MAX)
/* there is enough space, let's re-enable rx for our instance */
__mirror_rxflow_instance(mi, 1);
/* if nothing in queue, no timeout needed */
if (!worst)
return 1;
/*
* The guy(s) with the oldest tail block the ringbuffer from recycling
* the FIFO entries he has not read yet. Don't allow those guys to
* block the FIFO operation for very long.
*/
lws_start_foreach_ll(struct per_session_data__lws_mirror *,
pss, mi->same_mi_pss_list) {
if (pss->tail == worst_tail)
/*
* Our policy is if you are the slowest connection,
* you had better transmit something to help with that
* within 3s, or we will hang up on you to stop you
* blocking the FIFO for everyone else.
*/
lws_set_timeout(pss->wsi,
PENDING_TIMEOUT_USER_REASON_BASE, 3);
} lws_end_foreach_ll(pss, same_mi_pss_list);
return 1;
}
static void
__mirror_callback_all_in_mi_on_writable(struct mirror_instance *mi)
{
/* ask for WRITABLE callback for every wsi on this mi */
lws_start_foreach_ll(struct per_session_data__lws_mirror *,
pss, mi->same_mi_pss_list) {
lws_callback_on_writable(pss->wsi);
} lws_end_foreach_ll(pss, same_mi_pss_list);
}
static void
__mirror_destroy_message(void *_msg)
{
struct a_message *msg = _msg;
free(msg->payload);
msg->payload = NULL;
msg->len = 0;
}
static int
callback_lws_mirror(struct lws *wsi, enum lws_callback_reasons reason,
void *user, void *in, size_t len)
{
struct per_session_data__lws_mirror *pss =
(struct per_session_data__lws_mirror *)user;
struct per_vhost_data__lws_mirror *v =
(struct per_vhost_data__lws_mirror *)
lws_protocol_vh_priv_get(lws_get_vhost(wsi),
lws_get_protocol(wsi));
char name[300], update_worst, sent_something, *pn = name;
struct mirror_instance *mi = NULL;
const struct a_message *msg;
struct a_message amsg;
uint32_t oldest_tail;
int n, count_mi = 0;
switch (reason) {
case LWS_CALLBACK_ESTABLISHED:
lwsl_info("%s: LWS_CALLBACK_ESTABLISHED\n", __func__);
/*
* mirror instance name... defaults to "", but if URL includes
* "?mirror=xxx", will be "xxx"
*/
name[0] = '\0';
if (!lws_get_urlarg_by_name(wsi, "mirror", name,
sizeof(name) - 1))
lwsl_debug("get urlarg failed\n");
if (strchr(name, '='))
pn = strchr(name, '=') + 1;
//lwsl_notice("%s: mirror name '%s'\n", __func__, pn);
/* is there already a mirror instance of this name? */
lws_pthread_mutex_lock(&v->lock); /* vhost lock { */
lws_start_foreach_ll(struct mirror_instance *, mi1,
v->mi_list) {
count_mi++;
if (!strcmp(pn, mi1->name)) {
/* yes... we will join it */
mi = mi1;
break;
}
} lws_end_foreach_ll(mi1, next);
if (!mi) {
/* no existing mirror instance for name */
if (count_mi == MAX_MIRROR_INSTANCES) {
lws_pthread_mutex_unlock(&v->lock); /* } vh lock */
return -1;
}
/* create one with this name, and join it */
mi = malloc(sizeof(*mi));
if (!mi)
goto bail1;
memset(mi, 0, sizeof(*mi));
mi->ring = lws_ring_create(sizeof(struct a_message),
QUEUELEN,
__mirror_destroy_message);
if (!mi->ring) {
free(mi);
goto bail1;
}
mi->next = v->mi_list;
v->mi_list = mi;
lws_snprintf(mi->name, sizeof(mi->name) - 1, "%s", pn);
mi->rx_enabled = 1;
lws_pthread_mutex_init(&mi->lock);
lwsl_notice("Created new mi %p '%s'\n", mi, pn);
}
/* add our pss to list of guys bound to this mi */
lws_ll_fwd_insert(pss, same_mi_pss_list, mi->same_mi_pss_list);
/* init the pss */
pss->mi = mi;
pss->tail = lws_ring_get_oldest_tail(mi->ring);
pss->wsi = wsi;
lws_pthread_mutex_unlock(&v->lock); /* } vhost lock */
break;
bail1:
lws_pthread_mutex_unlock(&v->lock); /* } vhost lock */
return 1;
case LWS_CALLBACK_CLOSED:
/* detach our pss from the mirror instance */
mi = pss->mi;
if (!mi)
break;
lws_pthread_mutex_lock(&v->lock); /* vhost lock { */
/* remove our closing pss from its mirror instance list */
lws_ll_fwd_remove(struct per_session_data__lws_mirror,
same_mi_pss_list, pss, mi->same_mi_pss_list);
pss->mi = NULL;
if (mi->same_mi_pss_list) {
/*
* Still other pss using the mirror instance. The pss
* going away may have had the oldest tail, reconfirm
* using the remaining pss what is the current oldest
* tail. If the oldest tail moves on, this call also
* will re-enable rx flow control when appropriate.
*/
lws_pthread_mutex_lock(&mi->lock); /* mi lock { */
__mirror_update_worst_tail(mi);
lws_pthread_mutex_unlock(&mi->lock); /* } mi lock */
lws_pthread_mutex_unlock(&v->lock); /* } vhost lock */
break;
}
/* No more pss using the mirror instance... delete mi */
lws_start_foreach_llp(struct mirror_instance **,
pmi, v->mi_list) {
if (*pmi == mi) {
*pmi = (*pmi)->next;
lws_ring_destroy(mi->ring);
lws_pthread_mutex_destroy(&mi->lock);
free(mi);
break;
}
} lws_end_foreach_llp(pmi, next);
lws_pthread_mutex_unlock(&v->lock); /* } vhost lock */
break;
case LWS_CALLBACK_CONFIRM_EXTENSION_OKAY:
return 1; /* disallow compression */
case LWS_CALLBACK_PROTOCOL_INIT: /* per vhost */
lws_protocol_vh_priv_zalloc(lws_get_vhost(wsi),
lws_get_protocol(wsi),
sizeof(struct per_vhost_data__lws_mirror));
v = (struct per_vhost_data__lws_mirror *)
lws_protocol_vh_priv_get(lws_get_vhost(wsi),
lws_get_protocol(wsi));
lws_pthread_mutex_init(&v->lock);
break;
case LWS_CALLBACK_PROTOCOL_DESTROY:
lws_pthread_mutex_destroy(&v->lock);
break;
case LWS_CALLBACK_SERVER_WRITEABLE:
lws_pthread_mutex_lock(&pss->mi->lock); /* instance lock { */
oldest_tail = lws_ring_get_oldest_tail(pss->mi->ring);
update_worst = oldest_tail == pss->tail;
sent_something = 0;
do {
msg = lws_ring_get_element(pss->mi->ring, &pss->tail);
if (!msg)
break;
if (!msg->payload) {
lwsl_err("%s: NULL payload: worst = %d,"
" pss->tail = %d\n", __func__,
oldest_tail, pss->tail);
if (lws_ring_consume(pss->mi->ring, &pss->tail,
NULL, 1))
continue;
break;
}
n = lws_write(wsi, (unsigned char *)msg->payload +
LWS_PRE, msg->len, LWS_WRITE_TEXT);
if (n < 0) {
lwsl_info("%s: WRITEABLE: %d\n", __func__, n);
goto bail2;
}
sent_something = 1;
lws_ring_consume(pss->mi->ring, &pss->tail, NULL, 1);
} while (!lws_send_pipe_choked(wsi));
/* if any left for us to send, ask for writeable again */
if (lws_ring_get_count_waiting_elements(pss->mi->ring,
&pss->tail))
lws_callback_on_writable(wsi);
if (!sent_something || !update_worst)
goto done1;
/*
* We are no longer holding the oldest tail (since we sent
* something. So free us of the timeout related to hogging the
* oldest tail.
*/
lws_set_timeout(pss->wsi, NO_PENDING_TIMEOUT, 0);
/*
* If we were originally at the oldest fifo position of
* all the tails, now we used some up we may have
* changed the oldest fifo position and made some space.
*/
__mirror_update_worst_tail(pss->mi);
done1:
lws_pthread_mutex_unlock(&pss->mi->lock); /* } instance lock */
break;
bail2:
lws_pthread_mutex_unlock(&pss->mi->lock); /* } instance lock */
return -1;
case LWS_CALLBACK_RECEIVE:
lws_pthread_mutex_lock(&pss->mi->lock); /* mi lock { */
n = (int)lws_ring_get_count_free_elements(pss->mi->ring);
if (!n) {
lwsl_notice("dropping!\n");
if (pss->mi->rx_enabled)
__mirror_rxflow_instance(pss->mi, 0);
goto req_writable;
}
amsg.payload = malloc(LWS_PRE + len);
amsg.len = len;
if (!amsg.payload) {
lwsl_notice("OOM: dropping\n");
goto done2;
}
memcpy((char *)amsg.payload + LWS_PRE, in, len);
if (!lws_ring_insert(pss->mi->ring, &amsg, 1)) {
__mirror_destroy_message(&amsg);
lwsl_notice("dropping!\n");
if (pss->mi->rx_enabled)
__mirror_rxflow_instance(pss->mi, 0);
goto req_writable;
}
if (pss->mi->rx_enabled &&
lws_ring_get_count_free_elements(pss->mi->ring) <
RXFLOW_MIN)
__mirror_rxflow_instance(pss->mi, 0);
req_writable:
__mirror_callback_all_in_mi_on_writable(pss->mi);
done2:
lws_pthread_mutex_unlock(&pss->mi->lock); /* } mi lock */
break;
case LWS_CALLBACK_EVENT_WAIT_CANCELLED:
lwsl_info("LWS_CALLBACK_EVENT_WAIT_CANCELLED\n");
break;
default:
break;
}
return 0;
}
#define LWS_PLUGIN_PROTOCOL_MIRROR { \
"lws-mirror-protocol", \
callback_lws_mirror, \
sizeof(struct per_session_data__lws_mirror), \
4096, /* rx buf size must be >= permessage-deflate rx size */ \
0, NULL, 0 \
}
#if !defined (LWS_PLUGIN_STATIC)
static const struct lws_protocols protocols[] = {
LWS_PLUGIN_PROTOCOL_MIRROR
};
LWS_EXTERN LWS_VISIBLE int
init_protocol_lws_mirror(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_lws_mirror(struct lws_context *context)
{
return 0;
}
#endif

271
src/plugins/protocol_lws_status.c Normal file
View File

@@ -0,0 +1,271 @@
/*
* libwebsockets-test-server - libwebsockets test implementation
*
* 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.
*
* The person who associated a work with this deed has dedicated
* the work to the public domain by waiving all of his or her rights
* to the work worldwide under copyright law, including all related
* and neighboring rights, to the extent allowed by law. You can copy,
* modify, distribute and perform the work, even for commercial purposes,
* all without asking permission.
*
* The test apps are intended to be adapted for use in your code, which
* may be proprietary. So unlike the library itself, they are licensed
* Public Domain.
*/
#if !defined (LWS_PLUGIN_STATIC)
#define LWS_DLL
#define LWS_INTERNAL
#include <libwebsockets.h>
#endif
#include <time.h>
#include <string.h>
#ifdef WIN32
#include <io.h>
#include <gettimeofday.h>
#endif
typedef enum {
WALK_NONE,
WALK_INITIAL,
WALK_LIST,
WALK_FINAL
} e_walk;
struct per_session_data__lws_status {
struct per_session_data__lws_status *next;
struct lws *wsi;
time_t time_est;
char user_agent[256];
e_walk walk;
struct per_session_data__lws_status *walk_next;
unsigned char subsequent:1;
unsigned char changed_partway:1;
unsigned char wss_over_h2:1;
};
struct per_vhost_data__lws_status {
struct per_session_data__lws_status *live_pss_list;
struct lws_context *context;
struct lws_vhost *vhost;
const struct lws_protocols *protocol;
int count_live_pss;
};
static void
trigger_resend(struct per_vhost_data__lws_status *vhd)
{
lws_start_foreach_ll(struct per_session_data__lws_status *, pss,
vhd->live_pss_list) {
if (pss->walk == WALK_NONE) {
pss->subsequent = 0;
pss->walk_next = vhd->live_pss_list;
pss->walk = WALK_INITIAL;
} else
pss->changed_partway = 1;
} lws_end_foreach_ll(pss, next);
lws_callback_on_writable_all_protocol(vhd->context, vhd->protocol);
}
/* lws-status protocol */
int
callback_lws_status(struct lws *wsi, enum lws_callback_reasons reason,
void *user, void *in, size_t len)
{
struct per_session_data__lws_status *pss =
(struct per_session_data__lws_status *)user;
struct per_vhost_data__lws_status *vhd =
(struct per_vhost_data__lws_status *)
lws_protocol_vh_priv_get(lws_get_vhost(wsi),
lws_get_protocol(wsi));
char buf[LWS_PRE + 384], ip[24], *start = buf + LWS_PRE - 1, *p = start,
*end = buf + sizeof(buf) - 1;
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__lws_status));
vhd->context = lws_get_context(wsi);
vhd->protocol = lws_get_protocol(wsi);
vhd->vhost = lws_get_vhost(wsi);
break;
case LWS_CALLBACK_ESTABLISHED:
/*
* This shows how to stage sending a single ws message in
* multiple fragments. In this case, it lets us trade off
* memory needed to make the data vs time to send it.
*/
vhd->count_live_pss++;
pss->next = vhd->live_pss_list;
vhd->live_pss_list = pss;
pss->wss_over_h2 = !!len;
time(&pss->time_est);
pss->wsi = wsi;
if (lws_hdr_copy(wsi, pss->user_agent, sizeof(pss->user_agent),
WSI_TOKEN_HTTP_USER_AGENT) < 0) /* too big */
strcpy(pss->user_agent, "unknown");
trigger_resend(vhd);
break;
case LWS_CALLBACK_SERVER_WRITEABLE:
switch (pss->walk) {
case WALK_INITIAL:
n = LWS_WRITE_TEXT | LWS_WRITE_NO_FIN;
p += lws_snprintf(p, end - p,
"{ \"version\":\"%s\","
" \"wss_over_h2\":\"%d\","
" \"hostname\":\"%s\","
" \"wsi\":\"%d\", \"conns\":[",
lws_get_library_version(),
pss->wss_over_h2,
lws_canonical_hostname(vhd->context),
vhd->count_live_pss);
pss->walk = WALK_LIST;
pss->walk_next = vhd->live_pss_list;
break;
case WALK_LIST:
n = LWS_WRITE_CONTINUATION | LWS_WRITE_NO_FIN;
if (!pss->walk_next)
goto walk_final;
if (pss->subsequent)
*p++ = ',';
pss->subsequent = 1;
m = 0;
lws_start_foreach_ll(struct per_session_data__lws_status *,
pss2, vhd->live_pss_list) {
if (pss2 == pss->walk_next) {
m = 1;
break;
}
} lws_end_foreach_ll(pss2, next);
if (!m) {
/* our next guy went away */
pss->walk = WALK_FINAL;
pss->changed_partway = 1;
break;
}
strcpy(ip, "unknown");
lws_get_peer_simple(pss->walk_next->wsi, ip, sizeof(ip));
p += lws_snprintf(p, end - p,
"{\"peer\":\"%s\",\"time\":\"%ld\","
"\"ua\":\"%s\"}",
ip, (unsigned long)pss->walk_next->time_est,
pss->walk_next->user_agent);
pss->walk_next = pss->walk_next->next;
if (!pss->walk_next)
pss->walk = WALK_FINAL;
break;
case WALK_FINAL:
walk_final:
n = LWS_WRITE_CONTINUATION;
p += lws_snprintf(p, 4, "]}");
if (pss->changed_partway) {
pss->changed_partway = 0;
pss->subsequent = 0;
pss->walk_next = vhd->live_pss_list;
pss->walk = WALK_INITIAL;
} else
pss->walk = WALK_NONE;
break;
default:
return 0;
}
m = lws_write(wsi, (unsigned char *)start, p - start, n);
if (m < 0) {
lwsl_err("ERROR %d writing to di socket\n", m);
return -1;
}
if (pss->walk != WALK_NONE)
lws_callback_on_writable(wsi);
break;
case LWS_CALLBACK_RECEIVE:
lwsl_notice("pmd test: RX len %d\n", (int)len);
break;
case LWS_CALLBACK_CLOSED:
// lwsl_debug("****** LWS_CALLBACK_CLOSED\n");
lws_start_foreach_llp(struct per_session_data__lws_status **,
ppss, vhd->live_pss_list) {
if (*ppss == pss) {
*ppss = pss->next;
break;
}
} lws_end_foreach_llp(ppss, next);
trigger_resend(vhd);
break;
default:
break;
}
return 0;
}
#define LWS_PLUGIN_PROTOCOL_LWS_STATUS \
{ \
"lws-status", \
callback_lws_status, \
sizeof(struct per_session_data__lws_status), \
512, /* rx buf size must be >= permessage-deflate rx size */ \
0, NULL, 0 \
}
#if !defined (LWS_PLUGIN_STATIC)
static const struct lws_protocols protocols[] = {
LWS_PLUGIN_PROTOCOL_LWS_STATUS
};
LWS_EXTERN LWS_VISIBLE int
init_protocol_lws_status(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_lws_status(struct lws_context *context)
{
return 0;
}
#endif

314
src/plugins/protocol_post_demo.c Normal file
View File

@@ -0,0 +1,314 @@
/*
* ws protocol handler plugin for "POST demo"
*
* 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.
*
* The person who associated a work with this deed has dedicated
* the work to the public domain by waiving all of his or her rights
* to the work worldwide under copyright law, including all related
* and neighboring rights, to the extent allowed by law. You can copy,
* modify, distribute and perform the work, even for commercial purposes,
* all without asking permission.
*
* These test plugins are intended to be adapted for use in your code, which
* may be proprietary. So unlike the library itself, they are licensed
* Public Domain.
*/
#if !defined (LWS_PLUGIN_STATIC)
#define LWS_DLL
#define LWS_INTERNAL
#include <libwebsockets.h>
#endif
#include <stdlib.h>
#include <string.h>
#include <fcntl.h>
#include <sys/types.h>
#include <sys/stat.h>
#ifdef WIN32
#include <io.h>
#endif
#include <stdio.h>
struct per_session_data__post_demo {
struct lws_spa *spa;
char result[LWS_PRE + LWS_RECOMMENDED_MIN_HEADER_SPACE];
char filename[64];
long file_length;
#if !defined(LWS_WITH_ESP32)
lws_filefd_type fd;
#endif
uint8_t completed:1;
uint8_t sent_headers:1;
uint8_t sent_body:1;
};
static const char * const param_names[] = {
"text",
"send",
"file",
"upload",
};
enum enum_param_names {
EPN_TEXT,
EPN_SEND,
EPN_FILE,
EPN_UPLOAD,
};
static int
file_upload_cb(void *data, const char *name, const char *filename,
char *buf, int len, enum lws_spa_fileupload_states state)
{
struct per_session_data__post_demo *pss =
(struct per_session_data__post_demo *)data;
#if !defined(LWS_WITH_ESP32)
int n;
(void)n;
#endif
switch (state) {
case LWS_UFS_OPEN:
lws_strncpy(pss->filename, filename, sizeof(pss->filename));
/* we get the original filename in @filename arg, but for
* simple demo use a fixed name so we don't have to deal with
* attacks */
#if !defined(LWS_WITH_ESP32)
pss->fd = (lws_filefd_type)(long long)lws_open("/tmp/post-file",
O_CREAT | O_TRUNC | O_RDWR, 0600);
#endif
break;
case LWS_UFS_FINAL_CONTENT:
case LWS_UFS_CONTENT:
if (len) {
pss->file_length += len;
/* if the file length is too big, drop it */
if (pss->file_length > 100000)
return 1;
#if !defined(LWS_WITH_ESP32)
n = write((int)(long long)pss->fd, buf, len);
lwsl_info("%s: write %d says %d\n", __func__, len, n);
#else
lwsl_notice("%s: Received chunk size %d\n", __func__, len);
#endif
}
if (state == LWS_UFS_CONTENT)
break;
#if !defined(LWS_WITH_ESP32)
close((int)(long long)pss->fd);
pss->fd = LWS_INVALID_FILE;
#endif
break;
case LWS_UFS_CLOSE:
break;
}
return 0;
}
/*
* returns length in bytes
*/
static int
format_result(struct per_session_data__post_demo *pss)
{
unsigned char *p, *start, *end;
int n;
p = (unsigned char *)pss->result + LWS_PRE;
start = p;
end = p + sizeof(pss->result) - LWS_PRE - 1;
p += lws_snprintf((char *)p, end -p,
"<!DOCTYPE html><html lang=\"en\"><head>"
"<meta charset=utf-8 http-equiv=\"Content-Language\" "
"content=\"en\"/>"
"<title>LWS Server Status</title>"
"</head><body><h1>Form results (after urldecoding)</h1>"
"<table><tr><td>Name</td><td>Length</td><td>Value</td></tr>");
for (n = 0; n < (int)LWS_ARRAY_SIZE(param_names); n++) {
if (!lws_spa_get_string(pss->spa, n))
p += lws_snprintf((char *)p, end - p,
"<tr><td><b>%s</b></td><td>0"
"</td><td>NULL</td></tr>",
param_names[n]);
else
p += lws_snprintf((char *)p, end - p,
"<tr><td><b>%s</b></td><td>%d"
"</td><td>%s</td></tr>",
param_names[n],
lws_spa_get_length(pss->spa, n),
lws_spa_get_string(pss->spa, n));
}
p += lws_snprintf((char *)p, end - p,
"</table><br><b>filename:</b> %s, "
"<b>length</b> %ld",
pss->filename, pss->file_length);
p += lws_snprintf((char *)p, end - p, "</body></html>");
return (int)lws_ptr_diff(p, start);
}
static int
callback_post_demo(struct lws *wsi, enum lws_callback_reasons reason,
void *user, void *in, size_t len)
{
struct per_session_data__post_demo *pss =
(struct per_session_data__post_demo *)user;
unsigned char *p, *start, *end;
int n;
switch (reason) {
case LWS_CALLBACK_HTTP_BODY:
/* create the POST argument parser if not already existing */
if (!pss->spa) {
pss->spa = lws_spa_create(wsi, param_names,
LWS_ARRAY_SIZE(param_names), 1024,
file_upload_cb, pss);
if (!pss->spa)
return -1;
pss->filename[0] = '\0';
pss->file_length = 0;
}
/* let it parse the POST data */
if (lws_spa_process(pss->spa, in, (int)len))
return -1;
break;
case LWS_CALLBACK_HTTP_BODY_COMPLETION:
lwsl_debug("LWS_CALLBACK_HTTP_BODY_COMPLETION: %p\n", wsi);
/* call to inform no more payload data coming */
lws_spa_finalize(pss->spa);
pss->completed = 1;
lws_callback_on_writable(wsi);
break;
case LWS_CALLBACK_HTTP_WRITEABLE:
if (!pss->completed)
break;
p = (unsigned char *)pss->result + LWS_PRE;
start = p;
end = p + sizeof(pss->result) - LWS_PRE - 1;
if (!pss->sent_headers) {
n = format_result(pss);
if (lws_add_http_header_status(wsi, HTTP_STATUS_OK,
&p, end))
goto bail;
if (lws_add_http_header_by_token(wsi,
WSI_TOKEN_HTTP_CONTENT_TYPE,
(unsigned char *)"text/html", 9,
&p, end))
goto bail;
if (lws_add_http_header_content_length(wsi, n, &p, end))
goto bail;
if (lws_finalize_http_header(wsi, &p, end))
goto bail;
/* first send the headers ... */
n = lws_write(wsi, start, lws_ptr_diff(p, start),
LWS_WRITE_HTTP_HEADERS);
if (n < 0)
goto bail;
pss->sent_headers = 1;
lws_callback_on_writable(wsi);
break;
}
if (!pss->sent_body) {
n = format_result(pss);
n = lws_write(wsi, (unsigned char *)start, n,
LWS_WRITE_HTTP_FINAL);
pss->sent_body = 1;
if (n < 0)
return 1;
goto try_to_reuse;
}
break;
case LWS_CALLBACK_HTTP_DROP_PROTOCOL:
/* called when our wsi user_space is going to be destroyed */
if (pss->spa) {
lws_spa_destroy(pss->spa);
pss->spa = NULL;
}
break;
default:
break;
}
return 0;
bail:
return 1;
try_to_reuse:
if (lws_http_transaction_completed(wsi))
return -1;
return 0;
}
#define LWS_PLUGIN_PROTOCOL_POST_DEMO \
{ \
"protocol-post-demo", \
callback_post_demo, \
sizeof(struct per_session_data__post_demo), \
1024, \
0, NULL, 0 \
}
#if !defined (LWS_PLUGIN_STATIC)
static const struct lws_protocols protocols[] = {
LWS_PLUGIN_PROTOCOL_POST_DEMO
};
LWS_EXTERN LWS_VISIBLE int
init_protocol_post_demo(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_post_demo(struct lws_context *context)
{
return 0;
}
#endif