WolfSSL
/
wolfMQTT
mirror of https://github.com/wolfSSL/wolfMQTT.git
1
0
Fork 0
Code Issues Releases Wiki Activity

Add WebSocket broker configurations to CI workflow 

Add two new matrix entries to broker-check.yml for testing the broker
with WebSocket support: plain WebSocket and WebSocket + TLS. Installs
libwebsockets-dev and configures wolfSSL with --enable-opensslcoexist
for system lws compatibility.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
pull/466/head
Eric Blankenhorn 2026-02-17 13:01:03 -06:00
parent a4299c576d
commit dba0ca83ad
5 changed files with 68 additions and 67 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
.github/workflows/broker-check.yml vendored
View File

@ -37,13 +37,23 @@ jobs:
- name: "Broker TLS-only (no insecure)"
cflags: ""
wolfmqtt_opts: "--enable-broker --enable-tls --disable-broker-insecure"
- name: "Broker with WebSocket"
cflags: ""
wolfmqtt_opts: "--enable-broker --enable-websocket"
extra_deps: "libwebsockets-dev"
wolfssl_opts: "--enable-opensslcoexist"
- name: "Broker with WebSocket + TLS"
cflags: ""
wolfmqtt_opts: "--enable-broker --enable-tls --enable-websocket"
extra_deps: "libwebsockets-dev"
wolfssl_opts: "--enable-opensslcoexist --enable-enckeys"
steps:
- name: Install dependencies
run: |
export DEBIAN_FRONTEND=noninteractive
sudo apt-get update
sudo apt-get install -y mosquitto-clients
sudo apt-get install -y mosquitto-clients ${{ matrix.extra_deps }}
- uses: actions/checkout@master
with:
@ -54,7 +64,7 @@ jobs:
run: ./autogen.sh
- name: wolfssl configure
working-directory: ./wolfssl
run: ./configure --enable-enckeys
run: ./configure ${{ matrix.wolfssl_opts || '--enable-enckeys' }}
- name: wolfssl make
working-directory: ./wolfssl
run: make

44
examples/websocket/net_libwebsockets.c
View File

@ -33,6 +33,11 @@
#include <libwebsockets.h>
/* Compatibility for older libwebsockets versions (pre-4.1) */
#ifndef LWS_PROTOCOL_LIST_TERM
#define LWS_PROTOCOL_LIST_TERM { NULL, NULL, 0, 0, 0, NULL, 0 }
#endif
/* Network context for libwebsockets */
typedef struct _LibwebsockContext {
struct lws_context *context;
@ -74,37 +79,14 @@ static int callback_mqtt(struct lws *wsi, enum lws_callback_reasons reason,
XMEMCPY(net->rx_buffer + net->rx_len, in, len);
net->rx_len += len;
} else {
/* Buffer overflow - handle error */
lwsl_err("WebSocket receive buffer overflow"
"- dropping oldest data\n");
/* Simple approach: If new data is larger than buffer,
* just keep newest data */
if (len >= sizeof(net->rx_buffer)) {
/* New data is larger than entire buffer,
* keep only what fits */
/* Cast to byte pointer to allow pointer arithmetic */
const byte* in_bytes = (const byte*)in;
XMEMCPY(net->rx_buffer,
&in_bytes[len - sizeof(net->rx_buffer)],
sizeof(net->rx_buffer));
net->rx_len = sizeof(net->rx_buffer);
} else {
/* Keep as much new data as possible */
size_t keep_bytes = sizeof(net->rx_buffer) - len;
/* Move the portion of old data we want to
* keep to the beginning */
if (keep_bytes > 0 && net->rx_len > 0) {
XMEMMOVE(net->rx_buffer,
net->rx_buffer + (net->rx_len - keep_bytes),
keep_bytes);
}
/* Append all new data */
XMEMCPY(net->rx_buffer + keep_bytes, in, len);
net->rx_len = keep_bytes + len;
}
/* Dropping bytes would desynchronize MQTT packet framing,
* so treat overflow as a fatal protocol error. */
lwsl_err("WebSocket receive buffer overflow "
"(have=%d, need=%d, max=%d)\n",
(int)net->rx_len, (int)len,
(int)sizeof(net->rx_buffer));
net->status = -1;
return -1; /* close connection */
}
}
}

20
scripts/broker.test
View File

@ -887,7 +887,7 @@ if [ "$has_websocket" = "yes" ] && [ -x ./$ws_client_bin ]; then
broker_pid=$no_pid
fi
./$broker_bin -p $tcp_port -w $ws_port \
>"${TMP_DIR}/t17_broker.log" 2>&1 &
>"${TMP_DIR}/t22_broker.log" 2>&1 &
broker_pid=$!
check_broker $tcp_port
check_broker $ws_port
@ -899,29 +899,29 @@ if [ "$has_websocket" = "yes" ] && [ -x ./$ws_client_bin ]; then
# Use stdbuf to force line-buffered stdout (otherwise printf output
# is fully buffered when redirected to a file and lost on kill).
timeout 15 stdbuf -oL ./$ws_client_bin -h 127.0.0.1 -p $ws_port \
>"${TMP_DIR}/t17.log" 2>&1 &
T17_WS_PID=$!
TEST_PIDS+=($T17_WS_PID)
>"${TMP_DIR}/t22.log" 2>&1 &
T22_WS_PID=$!
TEST_PIDS+=($T22_WS_PID)
sleep 3
# Verify WS client connected and subscribed before publishing.
# Check both client log and broker log (broker always prints via
# PRINTF which is unbuffered).
ws_connected=no
if grep -q "MQTT Connected" "${TMP_DIR}/t17.log" 2>/dev/null; then
if grep -q "MQTT Connected" "${TMP_DIR}/t22.log" 2>/dev/null; then
ws_connected=yes
elif grep -q "SUBACK sock=-1" "${TMP_DIR}/t17_broker.log" 2>/dev/null; then
elif grep -q "SUBACK sock=-1" "${TMP_DIR}/t22_broker.log" 2>/dev/null; then
ws_connected=yes
fi
if [ "$ws_connected" = "yes" ]; then
# Publish from TCP client to the WS client's subscribed topic
./$pub_bin -T -h 127.0.0.1 -p $tcp_port -n "test/topic" -m "ws_hello" \
>"${TMP_DIR}/t17_pub.log" 2>&1
>"${TMP_DIR}/t22_pub.log" 2>&1
sleep 2
fi
kill $T17_WS_PID 2>/dev/null
wait $T17_WS_PID 2>/dev/null || true
kill $T22_WS_PID 2>/dev/null
wait $T22_WS_PID 2>/dev/null || true
TEST_PIDS=()
if grep -q "ws_hello" "${TMP_DIR}/t17.log" 2>/dev/null; then
if grep -q "ws_hello" "${TMP_DIR}/t22.log" 2>/dev/null; then
echo "PASS: WebSocket connect and subscribe (message received)"
elif [ "$ws_connected" = "yes" ]; then
echo "PASS: WebSocket connect and subscribe (connected ok)"

52
src/mqtt_broker.c
View File

@ -465,6 +465,11 @@ static void BrokerTls_Free(MqttBroker* broker)
#include <libwebsockets.h>
/* Compatibility for older libwebsockets versions (pre-4.1) */
#ifndef LWS_PROTOCOL_LIST_TERM
#define LWS_PROTOCOL_LIST_TERM { NULL, NULL, 0, 0, 0, NULL, 0 }
#endif
/* Forward declaration for the no-op connect callback (defined after WS section) */
static int BrokerNetConnect(void* context, const char* host, word16 port,
int timeout_ms);
@ -622,25 +627,13 @@ static int callback_broker_mqtt(struct lws *wsi,
ws->rx_len += len;
}
else {
WBLOG_ERR(broker, "broker: ws rx buffer overflow (wsi=%p)",
(void*)wsi);
/* Drop oldest data to make room */
if (len >= sizeof(ws->rx_buffer)) {
const byte* in_bytes = (const byte*)in;
XMEMCPY(ws->rx_buffer,
&in_bytes[len - sizeof(ws->rx_buffer)],
sizeof(ws->rx_buffer));
ws->rx_len = sizeof(ws->rx_buffer);
}
else {
size_t keep = sizeof(ws->rx_buffer) - len;
if (keep > 0 && ws->rx_len > 0) {
XMEMMOVE(ws->rx_buffer,
ws->rx_buffer + (ws->rx_len - keep), keep);
}
XMEMCPY(ws->rx_buffer + keep, in, len);
ws->rx_len = keep + len;
}
/* Dropping bytes would desynchronize MQTT packet framing,
* so treat overflow as a fatal protocol error. */
WBLOG_ERR(broker, "broker: ws rx buffer overflow "
"(wsi=%p, have=%d, need=%d, max=%d)",
(void*)wsi, (int)ws->rx_len, (int)len,
(int)sizeof(ws->rx_buffer));
return -1; /* close connection */
}
}
else if (reason == LWS_CALLBACK_SERVER_WRITEABLE) {
@ -653,14 +646,17 @@ static int callback_broker_mqtt(struct lws *wsi,
if (ws->tx_pending != NULL && ws->tx_len > 0) {
int n = lws_write(wsi, ws->tx_pending + LWS_PRE,
ws->tx_len, LWS_WRITE_BINARY);
if (n < (int)ws->tx_len) {
WBLOG_ERR(broker, "broker: ws write failed (wsi=%p, "
"n=%d, len=%d)", (void*)wsi, n, (int)ws->tx_len);
WOLFMQTT_FREE(ws->tx_pending);
ws->tx_pending = NULL;
ws->tx_len = 0;
return -1;
}
WOLFMQTT_FREE(ws->tx_pending);
ws->tx_pending = NULL;
ws->tx_len = 0;
if (n < 0) {
WBLOG_ERR(broker, "broker: ws write failed (wsi=%p)",
(void*)wsi);
return -1;
}
}
}
else if (reason == LWS_CALLBACK_CLOSED) {
@ -804,7 +800,15 @@ static int BrokerWsNetDisconnect(void* context)
}
if (ws->wsi != NULL && ws->status > 0) {
BrokerClient **bc_ptr;
WBLOG_INFO(bc->broker, "broker: ws disconnect (wsi=%p)", (void*)ws->wsi);
/* Clear lws per-session user data so that any later callbacks
* (e.g. LWS_CALLBACK_CLOSED) see NULL and skip processing.
* Without this, the callback would dereference the freed bc. */
bc_ptr = (BrokerClient**)lws_wsi_user(ws->wsi);
if (bc_ptr != NULL) {
*bc_ptr = NULL;
}
lws_close_reason(ws->wsi, LWS_CLOSE_STATUS_NORMAL, NULL, 0);
ws->wsi = NULL;
}

5
wolfmqtt/mqtt_broker.h
View File

@ -167,6 +167,11 @@ typedef struct MqttBrokerNet {
/* WebSocket per-client context */
/* -------------------------------------------------------------------------- */
#ifdef ENABLE_MQTT_WEBSOCKET
#ifdef WOLFMQTT_STATIC_MEMORY
#error "WebSocket support (ENABLE_MQTT_WEBSOCKET) is incompatible with " \
"static memory mode (WOLFMQTT_STATIC_MEMORY). libwebsockets " \
"requires dynamic allocation internally."
#endif
#ifndef BROKER_WS_RX_BUF_SZ
#define BROKER_WS_RX_BUF_SZ BROKER_RX_BUF_SZ
#endif