Crude, hard coded websockets support. No TLS.

ChangeLog.txt

@@ -1,4 +1,8 @@
+1.4 - xxxxxxxx
+==============
+
 Broker:
+- Initial websockets support.
 - Default TLS mode now accepts TLS v1.2, v1.1 and v1.0.
 - Support for ECDHE-ECDSA family ciphers.
 

config.mk

@@ -70,6 +70,9 @@ WITH_PYTHON:=yes
 # Build with SRV lookup support.
 WITH_SRV:=yes
 
+# Build with websockets support on the broker.
+WITH_WEBSOCKETS:=yes
+
 # Use elliptic keys in broker
 WITH_EC:=yes
 
@@ -212,6 +215,11 @@ ifeq ($(WITH_SRV),yes)
  LIB_LIBS:=$(LIB_LIBS) -lcares
 endif
 
+ifeq ($(WITH_WEBSOCKETS),yes)
+ BROKER_CFLAGS:=$(BROKER_CFLAGS) -DWITH_WEBSOCKETS
+ BROKER_LIBS:=$(BROKER_LIBS) -lwebsockets
+endif
+
 ifeq ($(UNAME),SunOS)
  BROKER_LIBS:=$(BROKER_LIBS) -lsocket -lnsl
  LIB_LIBS:=$(LIB_LIBS) -lsocket -lnsl

lib/mosquitto_internal.h

@@ -192,6 +192,10 @@ struct mosquitto {
  int db_index;
  struct _mosquitto_packet *out_packet_last;
  bool is_dropping;
+# ifdef WITH_WEBSOCKETS
+ struct libwebsocket_context *ws_context;
+ struct libwebsocket *wsi;
+# endif
 #else
  void *userdata;
  bool in_callback;

lib/net_mosq.c

@@ -65,6 +65,9 @@ and the Eclipse Distribution License is available at
  extern unsigned long g_pub_msgs_received;
  extern unsigned long g_pub_msgs_sent;
 # endif
+# ifdef WITH_WEBSOCKETS
+# include <libwebsockets.h>
+# endif
 #else
 # include <read_handle.h>
 #endif
@@ -155,7 +158,16 @@ int _mosquitto_packet_queue(struct mosquitto *mosq, struct _mosquitto_packet *pa
  mosq->out_packet_last = packet;
  pthread_mutex_unlock(&mosq->out_packet_mutex);
 #ifdef WITH_BROKER
+# ifdef WITH_WEBSOCKETS
+ if(mosq->wsi){
+ libwebsocket_callback_on_writable(mosq->ws_context, mosq->wsi);
+ return 0;
+ }else{
+ return _mosquitto_packet_write(mosq);
+ }
+# else
  return _mosquitto_packet_write(mosq);
+# endif
 #else
 
  /* Write a single byte to sockpairW (connected to sockpairR) to break out

lib/send_mosq.c

@@ -97,7 +97,12 @@ int _mosquitto_send_publish(struct mosquitto *mosq, uint16_t mid, const char *to
  assert(mosq);
  assert(topic);
 
+#if defined(WITH_BROKER) && defined(WITH_WEBSOCKETS)
+ if(mosq->sock == INVALID_SOCKET && !mosq->wsi) return MOSQ_ERR_NO_CONN;
+#else
  if(mosq->sock == INVALID_SOCKET) return MOSQ_ERR_NO_CONN;
+#endif
+
 #ifdef WITH_BROKER
  if(mosq->listener && mosq->listener->mount_point){
  len = strlen(mosq->listener->mount_point);

lib/util_mosq.c

@@ -34,6 +34,10 @@ and the Eclipse Distribution License is available at
 #include <mosquitto_broker.h>
 #endif
 
+#ifdef WITH_WEBSOCKETS
+#include <libwebsockets.h>
+#endif
+
 int _mosquitto_packet_alloc(struct _mosquitto_packet *packet)
 {
  uint8_t remaining_bytes[5], byte;
@@ -57,7 +61,11 @@ int _mosquitto_packet_alloc(struct _mosquitto_packet *packet)
  }while(remaining_length > 0 && packet->remaining_count < 5);
  if(packet->remaining_count == 5) return MOSQ_ERR_PAYLOAD_SIZE;
  packet->packet_length = packet->remaining_length + 1 + packet->remaining_count;
+#ifdef WITH_WEBSOCKETS
+ packet->payload = _mosquitto_malloc(sizeof(uint8_t)*packet->packet_length + LWS_SEND_BUFFER_PRE_PADDING + LWS_SEND_BUFFER_POST_PADDING);
+#else
  packet->payload = _mosquitto_malloc(sizeof(uint8_t)*packet->packet_length);
+#endif
  if(!packet->payload) return MOSQ_ERR_NOMEM;
 
  packet->payload[0] = packet->command;

src/CMakeLists.txt

@@ -26,6 +26,7 @@ set (MOSQ_SRCS
  ../lib/time_mosq.c
  ../lib/tls_mosq.c
  ../lib/util_mosq.c ../lib/util_mosq.h
+ websockets.c
  ../lib/will_mosq.c ../lib/will_mosq.h)
 
 option(INC_BRIDGE_SUPPORT
@@ -65,6 +66,12 @@ if (${WITH_SYS_TREE} STREQUAL ON)
  add_definitions("-DWITH_SYS_TREE")
 endif (${WITH_SYS_TREE} STREQUAL ON)
 
+option(WITH_WEBSOCKETS
+ "Include websockets support?" ON)
+if (${WITH_WEBSOCKETS} STREQUAL ON)
+ add_definitions("-DWITH_WEBSOCKETS")
+endif (${WITH_WEBSOCKETS} STREQUAL ON)
+
 if (WIN32 OR CYGWIN)
  set (MOSQ_SRCS ${MOSQ_SRCS} service.c)
 endif (WIN32 OR CYGWIN)
@@ -87,6 +94,10 @@ if (WIN32)
  set (MOSQ_LIBS ${MOSQ_LIBS} ws2_32)
 endif (WIN32)
 
+if (${WITH_WEBSOCKETS} STREQUAL ON)
+ set (MOSQ_LIBS ${MOSQ_LIBS} websockets)
+endif (${WITH_WEBSOCKETS} STREQUAL ON)
+
 target_link_libraries(mosquitto ${MOSQ_LIBS})
 
 install(TARGETS mosquitto RUNTIME DESTINATION ${SBINDIR} LIBRARY DESTINATION ${LIBDIR})

src/Makefile

@@ -8,7 +8,7 @@ else
 all : mosquitto
 endif
 
-mosquitto : mosquitto.o bridge.o conf.o context.o database.o logging.o loop.o memory_mosq.o persist.o net.o net_mosq.o read_handle.o read_handle_client.o read_handle_server.o read_handle_shared.o security.o security_default.o send_client_mosq.o send_mosq.o send_server.o service.o subs.o sys_tree.o time_mosq.o tls_mosq.o util_mosq.o will_mosq.o
+mosquitto : mosquitto.o bridge.o conf.o context.o database.o logging.o loop.o memory_mosq.o persist.o net.o net_mosq.o read_handle.o read_handle_client.o read_handle_server.o read_handle_shared.o security.o security_default.o send_client_mosq.o send_mosq.o send_server.o service.o subs.o sys_tree.o time_mosq.o tls_mosq.o util_mosq.o websockets.o will_mosq.o
  ${CC} $^ -o $@ ${LDFLAGS} $(BROKER_LIBS)
 
 mosquitto.o : mosquitto.c mosquitto_broker.h
@@ -89,6 +89,9 @@ tls_mosq.o : ../lib/tls_mosq.c
 util_mosq.o : ../lib/util_mosq.c ../lib/util_mosq.h
  ${CC} $(BROKER_CFLAGS) -c $< -o $@
 
+websockets.o : websockets.c mosquitto_broker.h
+ ${CC} $(BROKER_CFLAGS) -c $< -o $@
+
 will_mosq.o : ../lib/will_mosq.c ../lib/will_mosq.h
  ${CC} $(BROKER_CFLAGS) -c $< -o $@
 

src/database.c

@@ -256,7 +256,11 @@ int mqtt3_db_message_insert(struct mosquitto_db *db, struct mosquitto *context,
  }
  }
  }
+#ifdef WITH_WEBSOCKETS
+ if(context->sock == INVALID_SOCKET && !context->wsi){
+#else
  if(context->sock == INVALID_SOCKET){
+#endif
  /* Client is not connected only queue messages with QoS>0. */
  if(qos == 0 && !db->config->queue_qos0_messages){
  if(!context->bridge){
@@ -269,7 +273,11 @@ int mqtt3_db_message_insert(struct mosquitto_db *db, struct mosquitto *context,
  }
  }
 
+#ifdef WITH_WEBSOCKETS
+ if(context->sock != INVALID_SOCKET || context->wsi){
+#else
  if(context->sock != INVALID_SOCKET){
+#endif
  if(qos == 0 || max_inflight == 0 || context->msg_count12 < max_inflight){
  if(dir == mosq_md_out){
  switch(qos){
@@ -383,7 +391,15 @@ int mqtt3_db_message_insert(struct mosquitto_db *db, struct mosquitto *context,
  }
 #endif
 
+#ifdef WITH_WEBSOCKETS
+ if(context->wsi){
+ return mqtt3_db_message_write(context);
+ }else{
+ return rc;
+ }
+#else
  return rc;
+#endif
 }
 
 int mqtt3_db_message_update(struct mosquitto *context, uint16_t mid, enum mosquitto_msg_direction dir, enum mosquitto_msg_state state)
@@ -743,8 +759,13 @@ int mqtt3_db_message_write(struct mosquitto *context)
  const void *payload;
  int msg_count = 0;
 
+#ifdef WITH_WEBSOCKETS
+ if(!context || (context->sock == -1 && !context->wsi)
+ || (context->state == mosq_cs_connected && !context->id)){
+#else
  if(!context || context->sock == -1
  || (context->state == mosq_cs_connected && !context->id)){
+#endif
  return MOSQ_ERR_INVAL;
  }
 

src/loop.c

@@ -32,6 +32,10 @@ and the Eclipse Distribution License is available at
 #include <stdio.h>
 #include <string.h>
 
+#ifdef WITH_WEBSOCKETS
+# include <libwebsockets.h>
+#endif
+
 #include <mosquitto_broker.h>
 #include <memory_mosq.h>
 #include <time_mosq.h>
@@ -69,12 +73,19 @@ int mosquitto_main_loop(struct mosquitto_db *db, int *listensock, int listensock
  int bridge_sock;
  int rc;
 #endif
+#ifdef WITH_WEBSOCKETS
+ struct libwebsocket_context *ws_context;
+#endif
 
 #ifndef WIN32
  sigemptyset(&sigblock);
  sigaddset(&sigblock, SIGINT);
 #endif
 
+#ifdef WITH_WEBSOCKETS
+ ws_context = mosq_websockets_init(8080);
+#endif
+
  while(run){
 #ifdef WITH_SYS_TREE
  if(db->config->sys_interval > 0){
@@ -287,7 +298,13 @@ int mosquitto_main_loop(struct mosquitto_db *db, int *listensock, int listensock
  mqtt3_sub_tree_print(&db->subs, 0);
  flag_tree_print = false;
  }
+#ifdef WITH_WEBSOCKETS
+ libwebsocket_service(ws_context, 0);
+#endif
  }
+#ifdef WITH_WEBSOCKETS
+ libwebsocket_context_destroy(ws_context);
+#endif
 
  if(pollfds) _mosquitto_free(pollfds);
  return MOSQ_ERR_SUCCESS;

src/mosquitto_broker.h

@@ -420,4 +420,11 @@ void service_uninstall(void);
 void service_run(void);
 #endif
 
+/* ============================================================
+ * Websockets related functions
+ * ============================================================ */
+#ifdef WITH_WEBSOCKETS
+struct libwebsocket_context *mosq_websockets_init(int port);
+#endif
+
 #endif

src/read_handle_server.c

@@ -290,7 +290,7 @@ int mqtt3_handle_connect(struct mosquitto_db *db, struct mosquitto *context)
  }
 
 #ifdef WITH_TLS
- if(context->listener->use_identity_as_username){
+ if(context->listener && context->listener->use_identity_as_username){
  if(!context->ssl){
  _mosquitto_send_connack(context, CONNACK_REFUSED_BAD_USERNAME_PASSWORD);
  mqtt3_context_disconnect(db, context);