path: root/aclk/aclk_lws_wss_client.c

#include "aclk_lws_wss_client.h"

#include "libnetdata/libnetdata.h"

static int aclk_lws_wss_callback(struct lws *wsi, enum lws_callback_reasons reason, void *user, void *in, size_t len);

struct aclk_lws_wss_perconnect_data {
	int todo;
};

struct lws_wss_packet_buffer {
	unsigned char* data;
	size_t data_size;
	struct lws_wss_packet_buffer *next;
};

static inline struct lws_wss_packet_buffer *lws_wss_packet_buffer_new(void* data, size_t size)
{
	struct lws_wss_packet_buffer *new = callocz(1, sizeof(struct lws_wss_packet_buffer));
	if(data) {
		new->data = mallocz(LWS_PRE+size);
		memcpy(new->data+LWS_PRE, data, size);
		new->data_size = size;
	}
	return new;
}

static inline void lws_wss_packet_buffer_append(struct lws_wss_packet_buffer **list, struct lws_wss_packet_buffer *item)
{
	struct lws_wss_packet_buffer *tail = *list;
	if(!*list) {
		*list = item;
		return;
	}
	while(tail->next) {
		tail = tail->next;
	}
	tail->next = item;
}

static inline struct lws_wss_packet_buffer *lws_wss_packet_buffer_pop(struct lws_wss_packet_buffer **list)
{
	struct lws_wss_packet_buffer *ret = *list;
	if(ret != NULL)
		*list = ret->next;

	return ret;
}

static inline void lws_wss_packet_buffer_free(struct lws_wss_packet_buffer *item)
{
	freez(item->data);
	freez(item);
}

static inline void _aclk_lws_wss_read_buffer_clear(struct lws_ring *ringbuffer)
{
	size_t elems = lws_ring_get_count_waiting_elements(ringbuffer, NULL);
	if(elems > 0)
		lws_ring_consume(ringbuffer, NULL, NULL, elems);
}

static inline void _aclk_lws_wss_write_buffer_clear(struct lws_wss_packet_buffer **list)
{
	struct lws_wss_packet_buffer *i;
	while((i = lws_wss_packet_buffer_pop(list)) != NULL) {
		lws_wss_packet_buffer_free(i);
	}
	*list = NULL;
}

static inline void aclk_lws_wss_clear_io_buffers(struct aclk_lws_wss_engine_instance *inst)
{
	aclk_lws_mutex_lock(&inst->read_buf_mutex);
	_aclk_lws_wss_read_buffer_clear(inst->read_ringbuffer);
	aclk_lws_mutex_unlock(&inst->read_buf_mutex);
	aclk_lws_mutex_lock(&inst->write_buf_mutex);
	_aclk_lws_wss_write_buffer_clear(&inst->write_buffer_head);
	aclk_lws_mutex_unlock(&inst->write_buf_mutex);
}

static const struct lws_protocols protocols[] = {
	{
		"aclk-wss",
		aclk_lws_wss_callback,
		sizeof(struct aclk_lws_wss_perconnect_data),
		0, 0, 0, 0
	},
	{ NULL, NULL, 0, 0, 0, 0, 0 }
};

static void aclk_lws_wss_log_divert(int level, const char *line) {
	switch(level){
		case LLL_ERR:
			error("Libwebsockets Error: %s", line);
			break;
		case LLL_WARN:
			debug(D_ACLK, "Libwebsockets Warn: %s", line);
			break;
		default:
			error("Libwebsockets try to log with unknown log level (%d), msg: %s", level, line);
	}
}

struct aclk_lws_wss_engine_instance* aclk_lws_wss_client_init (const struct aclk_lws_wss_engine_callbacks *callbacks, const char *target_hostname, int target_port) {
	static int lws_logging_initialized = 0;
	struct lws_context_creation_info info;
	struct aclk_lws_wss_engine_instance *inst;

	if(unlikely(!lws_logging_initialized)) {
		lws_set_log_level(LLL_ERR | LLL_WARN, aclk_lws_wss_log_divert);
		lws_logging_initialized = 1;
	}

	if(!callbacks || !target_hostname)
		return NULL;

	inst = callocz(1, sizeof(struct aclk_lws_wss_engine_instance));

	inst->host = target_hostname;
	inst->port = target_port;

	memset(&info, 0, sizeof(struct lws_context_creation_info));
	info.options = LWS_SERVER_OPTION_DO_SSL_GLOBAL_INIT;
	info.port = CONTEXT_PORT_NO_LISTEN;
	info.protocols = protocols;
	info.user = inst;
	
	inst->lws_context = lws_create_context(&info);
	if(!inst->lws_context)
		goto failure_cleanup_2;

	inst->callbacks = *callbacks;

	aclk_lws_mutex_init(&inst->write_buf_mutex);
	aclk_lws_mutex_init(&inst->read_buf_mutex);

	inst->read_ringbuffer = lws_ring_create(1, ACLK_LWS_WSS_RECV_BUFF_SIZE_BYTES, NULL);
	if(!inst->read_ringbuffer)
		goto failure_cleanup;

	return inst;

failure_cleanup:
	lws_context_destroy(inst->lws_context);
failure_cleanup_2:
	freez(inst);
	return NULL;
}

void aclk_lws_wss_client_destroy(struct aclk_lws_wss_engine_instance* inst) {
	lws_context_destroy(inst->lws_context);
	inst->lws_context = NULL;
	inst->lws_wsi = NULL;

	aclk_lws_wss_clear_io_buffers(inst);

#ifdef ACLK_LWS_MOSQUITTO_IO_CALLS_MULTITHREADED
	pthread_mutex_destroy(&inst->write_buf_mutex);
	pthread_mutex_destroy(&inst->read_buf_mutex);
#endif
}

void _aclk_wss_connect(struct aclk_lws_wss_engine_instance *inst){
	struct lws_client_connect_info i;

	memset(&i, 0, sizeof(i));
	i.context = inst->lws_context;
	i.port = inst->port;
	i.address = inst->host;
	i.path = "/mqtt";
	i.host = inst->host;
	i.protocol = "mqtt";
#ifdef ACLK_SSL_ALLOW_SELF_SIGNED
	i.ssl_connection = LCCSCF_USE_SSL | LCCSCF_ALLOW_SELFSIGNED | LCCSCF_SKIP_SERVER_CERT_HOSTNAME_CHECK;
#else
	i.ssl_connection = LCCSCF_USE_SSL;
#endif
	lws_client_connect_via_info(&i);
}

static inline int received_data_to_ringbuff(struct lws_ring *buffer, void* data, size_t len) {
	if( lws_ring_insert(buffer, data, len) != len ) {
		error("ACLK_LWS_WSS_CLIENT: receive buffer full. Closing connection to prevent flooding.");
		return 0;
	}
	return 1;
}

static int
aclk_lws_wss_callback(struct lws *wsi, enum lws_callback_reasons reason,
			void *user, void *in, size_t len)
{
	UNUSED(user);
	struct aclk_lws_wss_engine_instance *inst = lws_context_user(lws_get_context(wsi));
	struct lws_wss_packet_buffer *data;
	int retval = 0;

	if( !inst ) {
		error("Callback received without any aclk_lws_wss_engine_instance!");
		return -1;
	}

	if( inst->upstream_reconnect_request ) {
		error("Closing lws connectino due to libmosquitto error.");
		char *upstream_connection_error = "MQTT protocol error. Closing underlying wss connection.";
		lws_close_reason(wsi, LWS_CLOSE_STATUS_PROTOCOL_ERR, (unsigned char*)upstream_connection_error, strlen(upstream_connection_error));
		retval = -1;
		inst->upstream_reconnect_request = 0;
	}

	switch (reason) {
	case LWS_CALLBACK_CLIENT_WRITEABLE:
		aclk_lws_mutex_lock(&inst->write_buf_mutex);
		data = lws_wss_packet_buffer_pop(&inst->write_buffer_head);
		if(likely(data)) {
			lws_write(wsi, data->data + LWS_PRE, data->data_size, LWS_WRITE_BINARY);
			lws_wss_packet_buffer_free(data);
			if(inst->write_buffer_head)
				lws_callback_on_writable(inst->lws_wsi);
		}
		aclk_lws_mutex_unlock(&inst->write_buf_mutex);
		break;
	case LWS_CALLBACK_CLIENT_RECEIVE:
		aclk_lws_mutex_lock(&inst->read_buf_mutex);
		if(!received_data_to_ringbuff(inst->read_ringbuffer, in, len))
			retval = 1;
		aclk_lws_mutex_unlock(&inst->read_buf_mutex);

		if(likely(inst->callbacks.data_rcvd_callback))
			// to future myself -> do not call this while read lock is active as it will eventually
			// want to acquire same lock later in aclk_lws_wss_client_read() function
			inst->callbacks.data_rcvd_callback