Support Unix Domain Socket connections (#146)

Author: jpgrayson

.devcontainer/docker-compose.yml

@@ -1,11 +1,12 @@
-# run this with docker-compose -f docker/docker-compose.yml run test
+# run this with: docker-compose -f docker-compose.yml run test
 version: "3.3"
 
 services:
   app:
     image: swift:5.9
     volumes:
       - ..:/workspace
+      - mosquitto-socket:/workspace/mosquitto/socket
     depends_on:
       - mosquitto
     environment:
@@ -18,8 +19,12 @@ services:
     volumes:
       - ../mosquitto/config:/mosquitto/config
       - ../mosquitto/certs:/mosquitto/certs
+      - mosquitto-socket:/mosquitto/socket
     ports:
       - "1883:1883"
       - "8883:8883"
       - "8080:8080"
       - "8081:8081"
+
+volumes:
+  mosquitto-socket:

.github/workflows/ci.yml

@@ -67,6 +67,7 @@ jobs:
         volumes:
           - ${{ github.workspace }}/mosquitto/config:/mosquitto/config
           - ${{ github.workspace }}/mosquitto/certs:/mosquitto/certs
+          - ${{ github.workspace }}/mosquitto/socket:/mosquitto/socket
 
     steps:
     - name: Checkout

README.md

@@ -14,6 +14,7 @@ MQTTNIO is a Swift NIO based implementation of a MQTT client. It supports
 - WebSocket connections
 - Posix sockets
 - Apple's Network framework via [NIOTransportServices](https://github.com/apple/swift-nio-transport-services) (required for iOS).
+- Unix domain sockets
 
 You can find documentation for MQTTNIO
 [here](https://swift-server-community.github.io/mqtt-nio/documentation/mqttnio/). There is also a sample demonstrating the use MQTTNIO in an iOS app found [here](https://github.com/adam-fowler/EmCuTeeTee)

Sources/MQTTNIO/MQTTClient.swift

@@ -141,6 +141,30 @@ public final class MQTTClient {
         self.inflight = .init()
     }
 
+    /// Create MQTT client
+    /// - Parameters:
+    ///   - unixSocketPath: Path to unix socket of MQTT broker
+    ///   - identifier: Client identifier. This must be unique
+    ///   - eventLoopGroupProvider: EventLoopGroup to run on
+    ///   - logger: Logger client should use
+    ///   - configuration: Configuration of client
+    public convenience init(
+        unixSocketPath: String,
+        identifier: String,
+        eventLoopGroupProvider: NIOEventLoopGroupProvider,
+        logger: Logger? = nil,
+        configuration: Configuration = Configuration()
+    ) {
+        self.init(
+            host: unixSocketPath,
+            port: 0,
+            identifier: identifier,
+            eventLoopGroupProvider: eventLoopGroupProvider,
+            logger: logger,
+            configuration: configuration
+        )
+    }
+
     deinit {
         guard isShutdown.load(ordering: .relaxed) else {
             preconditionFailure("Client not shut down before the deinit. Please call client.syncShutdownGracefully() when no longer needed.")

Sources/MQTTNIO/MQTTConnection.swift

@@ -46,7 +46,6 @@ final class MQTTConnection {
         do {
             // get bootstrap based off what eventloop we are running on
             let bootstrap = try getBootstrap(client: client)
-            bootstrap
                 .channelOption(ChannelOptions.socketOption(.so_reuseaddr), value: 1)
                 .channelOption(ChannelOptions.socket(IPPROTO_TCP, TCP_NODELAY), value: 1)
                 .connectTimeout(client.configuration.connectTimeout)
@@ -75,7 +74,16 @@ final class MQTTConnection {
                         return channel.pipeline.addHandlers(handlers)
                     }
                 }
-                .connect(host: client.host, port: client.port)
+
+            let channelFuture: EventLoopFuture<Channel>
+
+            if client.port == 0 {
+                channelFuture = bootstrap.connect(unixDomainSocketPath: client.host)
+            } else {
+                channelFuture = bootstrap.connect(host: client.host, port: client.port)
+            }
+
+            channelFuture
                 .map { channel in
                     if !client.configuration.useWebSockets {
                         channelPromise.succeed(channel)

Sources/MQTTNIO/MQTTNIO.docc/mqttnio-connections.md

@@ -1,6 +1,6 @@
 # Connections
 
-Support for TLS and WebSockets.
+Support for TLS, WebSockets, and Unix Domain Sockets.
 
 ## TLS
 
@@ -31,3 +31,23 @@ MQTT also supports Web Socket connections. Provide a `WebSocketConfiguration` wh
 ## NIO Transport Services
 
 On macOS and iOS you can use the NIO Transport Services library (NIOTS) and Apple's `Network.framework` for communication with the MQTT broker. If you don't provide an `eventLoopGroup` or a `TLSConfigurationType` then this is the default for both platforms. If you do provide either of these then the library will base it's decision on whether to use NIOTS or NIOSSL on what you provide. Provide a `MultiThreadedEventLoopGroup` or `NIOSSL.TLSConfiguration` and the client will use NIOSSL. Provide a `NIOTSEventLoopGroup` or `TSTLSConfiguration` and the client will use NIOTS. If you provide a `MultiThreadedEventLoopGroup` and a `TSTLSConfiguration` then the client will throw an error. If you are running on iOS you should always choose NIOTS.
+
+## Unix Domain Sockets
+
+MQTT NIO can connect to a local MQTT broker via a Unix Domain Socket.
+
+```swift
+let client = MQTTClient(
+    unixSocketPath: "/path/to/broker.socket",
+    identifier: "UDSClient",
+    eventLoopGroupProvider: .createNew
+)
+```
+
+Under the hood, `MQTTClient.port` will be 0 and `MQTTClient.host` will be the specified unix socket path when connecting to a unix socket.
+
+Note that mosquitto supports listening on a unix domain socket. This can be enabled by adding a `listener` option to the mosquitto config.
+
+```
+listener 0 /path/to/broker.socket
+```

Tests/MQTTNIOTests/MQTTNIOTests.swift

@@ -146,6 +146,20 @@ final class MQTTNIOTests: XCTestCase {
     }
     #endif
 
+    func testUnixDomainConnect() throws {
+        let client = MQTTClient(
+            unixSocketPath: MQTTNIOTests.rootPath + "/mosquitto/socket/mosquitto.sock",
+            identifier: "testUnixDomainConnect",
+            eventLoopGroupProvider: .createNew,
+            logger: self.logger,
+            configuration: .init()
+        )
+        defer { XCTAssertNoThrow(try client.syncShutdownGracefully()) }
+        _ = try client.connect().wait()
+        try client.ping().wait()
+        try client.disconnect().wait()
+    }
+
     func testMQTTPublishQoS0() throws {
         let client = self.createClient(identifier: "testMQTTPublishQoS0")
         defer { XCTAssertNoThrow(try client.syncShutdownGracefully()) }

docker-compose.yml

@@ -1,4 +1,4 @@
-# run this with docker-compose -f docker/docker-compose.yml run test
+# run this with: docker-compose -f docker-compose.yml run test
 version: "3.3"
 
 services:
@@ -7,20 +7,25 @@ services:
     working_dir: /mqtt-nio
     volumes:
       - .:/mqtt-nio
+      - mosquitto-socket:/mqtt-nio/mosquitto/socket
     depends_on:
       - mosquitto
     environment:
       - MOSQUITTO_SERVER=mosquitto
       - CI=true
-    command: /bin/bash -xcl "swift test --enable-test-discovery"
+    command: /bin/bash -xcl "swift test"
 
   mosquitto:
     image: eclipse-mosquitto
     volumes:
       - ./mosquitto/config:/mosquitto/config
       - ./mosquitto/certs:/mosquitto/certs
+      - mosquitto-socket:/mosquitto/socket
     ports:
       - "1883:1883"
       - "8883:8883"
       - "8080:8080"
       - "8081:8081"
+
+volumes:
+  mosquitto-socket:

mosquitto/config/mosquitto.conf

@@ -36,3 +36,8 @@ allow_anonymous true
 cafile ./mosquitto/certs/ca.pem
 certfile ./mosquitto/certs/server.pem
 keyfile ./mosquitto/certs/server.key
+
+# Unix Domain Socket
+listener 0 ./mosquitto/socket/mosquitto.sock
+protocol mqtt
+allow_anonymous true

mosquitto/socket/.gitkeep

@@ -0,0 +1,7 @@
+A local mosquitto server using `mosquitto/config/mosquitto.conf` will create a `mosquitto.sock` socket in this directory.
+
+If using the `docker-compose.yml` container environment, a shared container volume will be mounted here.
+
+This allows tests that connect to mosquitto via unix domain socket to assume the socket(s) will be found in this directory and work from multiple environments.
+
+Do not remove this directory.