Add MOSQ_EVT_CONNECT event.

This allows plugins to know when a client has successfully authenticated
to the broker.

Add connection-state example plugin to demonstrate MOSQ_EVT_CONNECT.

ChangeLog.txt

@@ -1,6 +1,11 @@
 2.1.0 - 2021-xx-xx
 ==================
 
+Broker:
+- Add MOSQ_EVT_CONNECT, to allow plugins to know when a client has
+  successfully authenticated to the broker.
+- Add connection-state example plugin to demonstrate MOSQ_EVT_CONNECT.
+
 Client library:
 - Add MOSQ_OPT_DISABLE_SOCKETPAIR to allow the disabling of the socketpair
   feature that allows the network thread to be woken from select() by another

include/mosquitto_broker.h

@@ -65,6 +65,7 @@ enum mosquitto_plugin_event {
 	MOSQ_EVT_PSK_KEY = 8,
 	MOSQ_EVT_TICK = 9,
 	MOSQ_EVT_DISCONNECT = 10,
+	MOSQ_EVT_CONNECT = 11,
 };
 
 /* Data for the MOSQ_EVT_RELOAD event */
@@ -162,6 +163,13 @@ struct mosquitto_evt_tick {
 	void *future2[4];
 };
 
+/* Data for the MOSQ_EVT_CONNECT event */
+struct mosquitto_evt_connect {
+	void *future;
+	struct mosquitto *client;
+	void *future2[4];
+};
+
 /* Data for the MOSQ_EVT_DISCONNECT event */
 struct mosquitto_evt_disconnect {
 	void *future;
@@ -185,15 +193,40 @@ typedef struct mosquitto_plugin_id_t mosquitto_plugin_id_t;
  *  identifier - the plugin identifier, as provided by <mosquitto_plugin_init>.
  *  event - the event to register a callback for. Can be one of:
  *          * MOSQ_EVT_RELOAD
+ *              Called when the broker is sent a signal indicating it should
+ *              reload its configuration.
  *          * MOSQ_EVT_ACL_CHECK
+ *              Called when a publish/subscribe/unsubscribe command is received
+ *              and the broker wants to check when the client is allowed to carry
+ *              out this command.
  *          * MOSQ_EVT_BASIC_AUTH
+ *              Called when a client connects to the broker, to allow the
+ *              username/password/clientid to be authenticated.
  *          * MOSQ_EVT_EXT_AUTH_START
+ *              Called when an MQTT v5 client connects, if it is using extended
+ *              authentication.
  *          * MOSQ_EVT_EXT_AUTH_CONTINUE
+ *              Called when an MQTT v5 client connects, if it is using extended
+ *              authentication.
  *          * MOSQ_EVT_CONTROL
+ *              Called on receipt of a $CONTROL message that the plugin has
+ *              registered for.
  *          * MOSQ_EVT_MESSAGE
+ *              Called for each PUBLISH message after it has been received and
+ *              authorised, but before it is sent to subscribing clients. The
+ *              contents of the message can be modified.
  *          * MOSQ_EVT_PSK_KEY
+ *              Called when a client connects with TLS-PSK and the broker needs
+ *              the PSK information.
  *          * MOSQ_EVT_TICK
+ *              Called periodically in the event loop. At the moment this
+ *              occurs at a regular frequency, but this should not be relied
+ *              upon.
  *          * MOSQ_EVT_DISCONNECT
+ *              Called when a client disconnects from the broker.
+ *          * MOSQ_EVT_CONNECT
+ *              Called when a client has successfully connected to the broker,
+ *              i.e. has been authenticated.
  *  cb_func - the callback function
  *  event_data - event specific data
  *
@@ -229,6 +262,7 @@ mosq_EXPORT int mosquitto_callback_register(
  *          * MOSQ_EVT_PSK_KEY
  *          * MOSQ_EVT_TICK
  *          * MOSQ_EVT_DISCONNECT
+ *          * MOSQ_EVT_CONNECT
  *  cb_func - the callback function
  *  event_data - event specific data
  *

plugins/Makefile

@@ -1,5 +1,6 @@
 DIRS= \
 		auth-by-ip \
+		connection-state \
 		dynamic-security \
 		message-timestamp \
 		payload-modification

plugins/README.md

@@ -7,6 +7,12 @@ This is a fully functioning plugin that implements authentication and access
 control, with configuration via a $CONTROL topic. See the readme in
 dynamic-security for more information.
 
+## Connection state
+This is an **example** plugin to demonstrate the use of the connect and
+disconnect events. It publishes messages to
+$SYS/broker/connection/client/<client id>/state for every client that connects
+to the broker, to indicate the connection state of that client.
+
 ## Message timestamp
 This is an **example** plugin to demonstrate how it is possible to attach MQTT v5 properties to messages after they have been received, and before they are sent on to subscribers.
 

plugins/connection-state/CMakeLists.txt

@@ -0,0 +1,11 @@
+include_directories(${mosquitto_SOURCE_DIR} ${mosquitto_SOURCE_DIR}/include
+			${STDBOOL_H_PATH} ${STDINT_H_PATH})
+
+add_library(mosquitto_connection_state SHARED mosquitto_connection_state.c)
+set_target_properties(mosquitto_connection_state PROPERTIES
+	POSITION_INDEPENDENT_CODE 1
+)
+set_target_properties(mosquitto_connection_state PROPERTIES PREFIX "")
+
+# Don't install, these are example plugins only.
+#install(TARGETS mosquitto_connection_state RUNTIME DESTINATION "${CMAKE_INSTALL_BINDIR}" LIBRARY DESTINATION "${CMAKE_INSTALL_LIBDIR}")

plugins/connection-state/Makefile

@@ -0,0 +1,27 @@
+include ../../config.mk
+
+.PHONY : all binary check clean reallyclean test install uninstall
+
+PLUGIN_NAME=mosquitto_connection_state
+
+all : binary
+
+binary : ${PLUGIN_NAME}.so
+
+${PLUGIN_NAME}.so : ${PLUGIN_NAME}.c
+	$(CROSS_COMPILE)$(CC) $(PLUGIN_CPPFLAGS) $(PLUGIN_CFLAGS) $(PLUGIN_LDFLAGS) -fPIC -shared $< -o $@
+
+reallyclean : clean
+clean:
+	-rm -f *.o ${PLUGIN_NAME}.so *.gcda *.gcno
+
+check: test
+test:
+
+install: ${PLUGIN_NAME}.so
+	# Don't install, these are examples only.
+	#$(INSTALL) -d "${DESTDIR}$(libdir)"
+	#$(INSTALL) ${STRIP_OPTS} ${PLUGIN_NAME}.so "${DESTDIR}${libdir}/${PLUGIN_NAME}.so"
+
+uninstall :
+	-rm -f "${DESTDIR}${libdir}/${PLUGIN_NAME}.so"

plugins/connection-state/mosquitto_connection_state.c

@@ -0,0 +1,117 @@
+/*
+Copyright (c) 2021 Roger Light <roger@atchoo.org>
+
+All rights reserved. This program and the accompanying materials
+are made available under the terms of the Eclipse Public License 2.0
+and Eclipse Distribution License v1.0 which accompany this distribution.
+ 
+The Eclipse Public License is available at
+   https://www.eclipse.org/legal/epl-2.0/
+and the Eclipse Distribution License is available at
+  http://www.eclipse.org/org/documents/edl-v10.php.
+ 
+SPDX-License-Identifier: EPL-2.0 OR EDL-1.0
+
+Contributors:
+   Roger Light - initial implementation and documentation.
+*/
+
+/*
+ * This is an example plugin showing how you could publish online/offline
+ * state for all clients.
+ *
+ * Compile with:
+ *   gcc -I<path to mosquitto-repo/include> -fPIC -shared mosquitto_connection_state.c -o mosquitto_connection_state.so
+ *
+ * Use in config with:
+ *
+ *   plugin /path/to/mosquitto_connection_state.so
+ *
+ * Note that this only works on Mosquitto 2.0 or later.
+ */
+
+
+#include <stdbool.h>
+#include <stdio.h>
+#include <string.h>
+
+#include "mosquitto_broker.h"
+#include "mosquitto_plugin.h"
+#include "mosquitto.h"
+#include "mqtt_protocol.h"
+
+static mosquitto_plugin_id_t *mosq_pid = NULL;
+
+static int connect_callback(int event, void *event_data, void *userdata)
+{
+	struct mosquitto_evt_connect *ed = event_data;
+	const char *client_id;
+	char topic[1024];
+	int len;
+
+	client_id = mosquitto_client_id(ed->client);
+	len = snprintf(topic, sizeof(topic), "$SYS/broker/connection/client/%s/state", client_id);
+	if(len < sizeof(topic)){
+		mosquitto_broker_publish_copy(NULL, topic, 1, "1", 0, true, NULL);
+	}else{
+		/* client id too large */
+	}
+
+	return MOSQ_ERR_SUCCESS;
+}
+
+static int disconnect_callback(int event, void *event_data, void *userdata)
+{
+	struct mosquitto_evt_disconnect *ed = event_data;
+	const char *client_id;
+	char topic[1024];
+	int len;
+	mosquitto_property *proplist = NULL;
+	int rc;
+
+	client_id = mosquitto_client_id(ed->client);
+	len = snprintf(topic, sizeof(topic), "$SYS/broker/connection/client/%s/state", client_id);
+	if(len < sizeof(topic)){
+		/* Expire our "disconnected" message after a day. */
+		mosquitto_property_add_int32(&proplist, MQTT_PROP_MESSAGE_EXPIRY_INTERVAL, 86400);
+		rc = mosquitto_broker_publish_copy(NULL, topic, 1, "0", 0, true, proplist);
+		if(rc){
+			mosquitto_property_free_all(&proplist);
+		}
+	}else{
+		/* client id too large */
+	}
+
+	return MOSQ_ERR_SUCCESS;
+}
+
+
+int mosquitto_plugin_version(int supported_version_count, const int *supported_versions)
+{
+	int i;
+
+	for(i=0; i<supported_version_count; i++){
+		if(supported_versions[i] == 5){
+			return 5;
+		}
+	}
+	return -1;
+}
+
+int mosquitto_plugin_init(mosquitto_plugin_id_t *identifier, void **user_data, struct mosquitto_opt *opts, int opt_count)
+{
+	int rc;
+
+	mosq_pid = identifier;
+
+	rc = mosquitto_callback_register(mosq_pid, MOSQ_EVT_CONNECT, connect_callback, NULL, NULL);
+	if(rc) return rc;
+	rc = mosquitto_callback_register(mosq_pid, MOSQ_EVT_DISCONNECT, disconnect_callback, NULL, NULL);
+	return rc;
+}
+
+int mosquitto_plugin_cleanup(void *user_data, struct mosquitto_opt *opts, int opt_count)
+{
+	mosquitto_callback_unregister(mosq_pid, MOSQ_EVT_CONNECT, connect_callback, NULL);
+	return mosquitto_callback_unregister(mosq_pid, MOSQ_EVT_DISCONNECT, disconnect_callback, NULL);
+}

src/handle_connect.c

@@ -283,6 +283,10 @@ int connect__on_authorised(struct mosquitto *context, void *auth_data_out, uint1
 	rc = db__message_write_queued_out(context);
 	if(rc) return rc;
 	rc = db__message_write_inflight_out_all(context);
+
+	if(rc == MOSQ_ERR_SUCCESS){
+		plugin__handle_connect(context);
+	}
 	return rc;
 error:
 	free(auth_data_out);

src/mosquitto_broker_internal.h

@@ -150,6 +150,7 @@ struct plugin__callbacks{
 	struct mosquitto__callback *tick;
 	struct mosquitto__callback *acl_check;
 	struct mosquitto__callback *basic_auth;
+	struct mosquitto__callback *connect;
 	struct mosquitto__callback *control;
 	struct mosquitto__callback *disconnect;
 	struct mosquitto__callback *ext_auth_continue;
@@ -762,6 +763,7 @@ void listeners__add_websockets(struct lws_context *ws_context, mosq_sock_t fd);
  * Plugin related functions
  * ============================================================ */
 int plugin__load_v5(struct mosquitto__listener *listener, struct mosquitto__auth_plugin *plugin, struct mosquitto_opt *auth_options, int auth_option_count, void *lib);
+void plugin__handle_connect(struct mosquitto *context);
 void plugin__handle_disconnect(struct mosquitto *context, int reason);
 int plugin__handle_message(struct mosquitto *context, struct mosquitto_msg_store *stored);
 void LIB_ERROR(void);

src/plugin.c

@@ -103,6 +103,29 @@ int plugin__load_v5(struct mosquitto__listener *listener, struct mosquitto__auth
 }
 
 
+void plugin__handle_connect(struct mosquitto *context)
+{
+	struct mosquitto_evt_connect event_data;
+	struct mosquitto__callback *cb_base;
+	struct mosquitto__security_options *opts;
+
+	if(db.config->per_listener_settings){
+		if(context->listener == NULL){
+			return;
+		}
+		opts = &context->listener->security_options;
+	}else{
+		opts = &db.config->security_options;
+	}
+	memset(&event_data, 0, sizeof(event_data));
+
+	event_data.client = context;
+	DL_FOREACH(opts->plugin_callbacks.connect, cb_base){
+		cb_base->cb(MOSQ_EVT_CONNECT, &event_data, cb_base->userdata);
+	}
+}
+
+
 void plugin__handle_disconnect(struct mosquitto *context, int reason)
 {
 	struct mosquitto_evt_disconnect event_data;
@@ -116,8 +139,8 @@ void plugin__handle_disconnect(struct mosquitto *context, int reason)
 		opts = &context->listener->security_options;
 	}else{
 		opts = &db.config->security_options;
-		memset(&event_data, 0, sizeof(event_data));
 	}
+	memset(&event_data, 0, sizeof(event_data));
 
 	event_data.client = context;
 	event_data.reason = reason;
@@ -244,6 +267,9 @@ int mosquitto_callback_register(
 		case MOSQ_EVT_DISCONNECT:
 			cb_base = &security_options->plugin_callbacks.disconnect;
 			break;
+		case MOSQ_EVT_CONNECT:
+			cb_base = &security_options->plugin_callbacks.connect;
+			break;
 		default:
 			return MOSQ_ERR_NOT_SUPPORTED;
 			break;