Comqtt

https://github.com/wind-c/comqtt
Admin
Comqtt is a lightweight, high-performance MQTT broker server written in Go, supporting MQTT v3.0,...
Tokens:15,273
Snippets:136
Trust Score:8
License:MIT
Update:6 months ago
Show doc for...
Tokens:
Raw
### Comqtt Server Initialization Output

Source: https://github.com/wind-c/comqtt/blob/main/cluster/README.md

Example output logs indicating a Comqtt server has successfully initialized and started.

```shell
8:16PM INF comqtt server initializing...
8:16PM INF added hook hook=redis-db
8:16PM INF connecting to redis service address=127.0.0.1:6379 db=0 hook=redis-db
8:16PM INF connected to redis service hook=redis-db
8:16PM INF added hook hook=allow-all-auth
8:16PM INF added hook hook=agent-event
8:16PM INF found raft leader leader=c01
8:16PM INF setup raft addr=127.0.0.1:8946 node=c01
8:16PM INF local member addr=127.0.0.1 port=7946
8:16PM INF cluster node created
8:16PM INF attached listener address=:1883 id=tcp protocol=tcp
8:16PM INF attached listener address=:1882 id=ws protocol=ws
8:16PM INF attached listener address=:8080 id=stats protocol=http
8:16PM INF comqtt server started
8:16PM INF notify join addr=127.0.0.1 node=c02
8:16PM INF raft joined addr=127.0.0.1:8947 node=c02
8:17PM INF notify join addr=127.0.0.1 node=c03
8:17PM INF raft joined addr=127.0.0.1:8948 node=c03
```

--------------------------------

### Second Comqtt Node Initialization Output

Source: https://github.com/wind-c/comqtt/blob/main/cluster/README.md

Example output logs for the second Comqtt node joining the cluster and starting.

```shell
8:16PM INF comqtt server initializing...
8:16PM INF added hook hook=redis-db
8:16PM INF connecting to redis service address=127.0.0.1:6379 db=0 hook=redis-db
8:16PM INF connected to redis service hook=redis-db
8:16PM INF added hook hook=allow-all-auth
8:16PM INF added hook hook=agent-event
8:16PM INF setup raft addr=127.0.0.1:8947 node=c02
8:16PM INF local member addr=127.0.0.1 port=7947
8:16PM INF cluster node created
8:16PM INF attached listener address=:1885 id=tcp protocol=tcp
8:16PM INF attached listener address=:1886 id=ws protocol=ws
8:16PM INF attached listener address=:1881 id=stats protocol=http
8:16PM INF notify join addr=127.0.0.1 node=c01
8:16PM INF raft joining addr=127.0.0.1:8946 node=c01
8:16PM INF comqtt server started
8:17PM INF notify join addr=127.0.0.1 node=c03
8:17PM INF raft joining addr=127.0.0.1:8948 node=c03
```

--------------------------------

### Build and Start Comqtt Broker

Source: https://github.com/wind-c/comqtt/blob/main/README.md

Commands to compile the broker from source and execute it with optional configuration.

```bash
cd cmd
go build -o comqtt ./single/main.go
```

```bash
./comqtt
or
./comqtt --conf=./config/single.yml
```

--------------------------------

### Start Third Comqtt Node

Source: https://github.com/wind-c/comqtt/blob/main/cluster/README.md

Starts the third Comqtt node, joining an existing cluster. Configuration can be done via command-line flags or a configuration file.

```shell
./comqtt --node-name=c03 --gossip-port=7948 --raft-port=8948 --members=localhost:7946 --tcp=:1887 --ws=:1888 --http=:1882
```

```shell
./comqtt --conf=./config/node3.yml
```

--------------------------------

### Start Second Comqtt Node

Source: https://github.com/wind-c/comqtt/blob/main/cluster/README.md

Starts a second Comqtt node, joining an existing cluster. Specify the existing node using the --members flag or a configuration file.

```shell
./comqtt --node-name=c02 --gossip-port=7947 --raft-port=8947 --members=localhost:7946 --tcp=:1885 --ws=:1886 --http=:1881
```

```shell
./comqtt --conf=./config/node2.yml
```

--------------------------------

### Start First Comqtt Node

Source: https://github.com/wind-c/comqtt/blob/main/cluster/README.md

Initiates the first node in a Comqtt cluster. Can be configured via command-line flags or a configuration file.

```shell
./comqtt --node-name=c01 --gossip-port=7946 --raft-port=8946 --raft-bootstrap=true
```

```shell
./comqtt --conf=./config/node1.yml
```

--------------------------------

### GET /api/v1/mqtt/config

Source: https://context7.com/wind-c/comqtt/llms.txt

Retrieves the current server configuration and capabilities.

```APIDOC
## GET /api/v1/mqtt/config

### Description
Returns the server configuration, including capabilities and buffer sizes.

### Method
GET

### Endpoint
/api/v1/mqtt/config

### Response
#### Success Response (200)
- **capabilities** (object) - Server capabilities.
- **client_write_buffer_size** (integer) - Configured write buffer size.

#### Response Example
{"capabilities":{"maximum_qos":2},"client_write_buffer_size":2048}
```

--------------------------------

### GET /api/v1/mqtt/stat/overall

Source: https://github.com/wind-c/comqtt/blob/main/README.md

Retrieves overall MQTT server information.

```APIDOC
## GET /api/v1/mqtt/stat/overall

### Description
Get MQTT server info.

### Method
GET

### Endpoint
/api/v1/mqtt/stat/overall
```

--------------------------------

### Add AllowAll Hook to MQTT Server

Source: https://github.com/wind-c/comqtt/blob/main/README.md

Example of adding the `auth.AllowHook` to a ComMQTT server to permit all connections, subscriptions, and publications. Use with caution, primarily for development and testing.

```go
server := mqtt.New(nil)
_ = server.AddHook(new(auth.AllowHook), nil)
```

--------------------------------

### GET /api/v1/mqtt/clients/{id}

Source: https://github.com/wind-c/comqtt/blob/main/README.md

Retrieves information for a specific client.

```APIDOC
## GET /api/v1/mqtt/clients/{id}

### Description
Get a client info.

### Method
GET

### Endpoint
/api/v1/mqtt/clients/{id}

### Parameters
#### Path Parameters
- **id** (string) - Required - The client ID
```

--------------------------------

### Generate Bcrypt Password Hash in Python

Source: https://github.com/wind-c/comqtt/blob/main/README.md

Python snippet to generate a bcrypt hashed password for MQTT clients. Ensure the 'bcrypt' library is installed.

```python
import bcrypt
salt = bcrypt.gensalt(rounds=10)
hashed = bcrypt.hashpw(b"VeryVerySecretPa55w0rd", salt)
print(f"Password hash for MQTT client: {hashed}")
```

--------------------------------

### GET /api/v1/mqtt/stat/online

Source: https://context7.com/wind-c/comqtt/llms.txt

Retrieves the current count of connected MQTT clients.

```APIDOC
## GET /api/v1/mqtt/stat/online

### Description
Returns the number of currently connected clients.

### Method
GET

### Endpoint
/api/v1/mqtt/stat/online

### Response
#### Success Response (200)
- **count** (integer) - The number of connected clients.

#### Response Example
5
```

--------------------------------

### GET /api/v1/mqtt/clients/{client_id}

Source: https://context7.com/wind-c/comqtt/llms.txt

Retrieves detailed information about a specific connected client.

```APIDOC
## GET /api/v1/mqtt/clients/{client_id}

### Description
Fetches information for a specific client by ID.

### Method
GET

### Endpoint
/api/v1/mqtt/clients/{client_id}

### Parameters
#### Path Parameters
- **client_id** (string) - Required - The unique identifier of the client.

### Response
#### Success Response (200)
- **id** (string) - Client ID.
- **remote** (string) - Remote address.
- **username** (string) - Username used for connection.

#### Response Example
{"id":"client-123","remote":"192.168.1.100:54321","username":"sensor"}
```

--------------------------------

### Configure Server Options

Source: https://github.com/wind-c/comqtt/blob/main/README.md

Initialize the MQTT server with custom capabilities and buffer settings.

```go
server := mqtt.New(&mqtt.Options{
  Capabilities: mqtt.Capabilities{
    MaximumSessionExpiryInterval: 3600,
    Compatibilities: mqtt.Compatibilities{
      ObscureNotAuthorized: true,
    },
  },
  ClientNetWriteBufferSize: 1024,
  ClientNetReadBufferSize: 1024,
  SysTopicResendInterval: 10,
})
```

--------------------------------

### Import Comqtt as a Package

Source: https://github.com/wind-c/comqtt/blob/main/README.md

Initialize a basic MQTT server instance with a TCP listener and authentication hook.

```go
import (
  "log"

  "github.com/wind-c/comqtt/v2/mqtt"
  "github.com/wind-c/comqtt/v2/mqtt/hooks/auth"
  "github.com/wind-c/comqtt/v2/mqtt/listeners"
)

func main() {
  // Create the new MQTT Server.
  server := mqtt.New(nil)

  // Allow all connections.
  _ = server.AddHook(new(auth.AllowHook), nil)

  // Create a TCP listener on a standard port.
  tcp := listeners.NewTCP("t1", ":1883", nil)
  err := server.AddListener(tcp)
  if err != nil {
    log.Fatal(err)
  }

  err = server.Serve()
  if err != nil {
    log.Fatal(err)
  }
}
```

--------------------------------

### Create a Basic MQTT Server

Source: https://context7.com/wind-c/comqtt/llms.txt

Initializes a new MQTT server instance with custom capabilities and adds an authentication hook and TCP listener. The AllowHook is intended for development purposes only.

```go
package main

import (
    "log"
    "os"
    "os/signal"
    "syscall"

    "github.com/wind-c/comqtt/v2/mqtt"
    "github.com/wind-c/comqtt/v2/mqtt/hooks/auth"
    "github.com/wind-c/comqtt/v2/mqtt/listeners"
)

func main() {
    // Handle shutdown signals
    sigs := make(chan os.Signal, 1)
    done := make(chan bool, 1)
    signal.Notify(sigs, syscall.SIGINT, syscall.SIGTERM)
    go func() {
        <-sigs
        done <- true
    }()

    // Create server with custom options
    server := mqtt.New(&mqtt.Options{
        Capabilities: &mqtt.Capabilities{
            MaximumSessionExpiryInterval: 3600,
            MaximumMessageExpiryInterval: 86400,
            ReceiveMaximum:               1024,
            MaximumQos:                   2,
            RetainAvailable:              1,
            TopicAliasMaximum:            65535,
            MaximumClientWritesPending:   8192,
        },
        ClientNetWriteBufferSize: 2048,
        ClientNetReadBufferSize:  2048,
        SysTopicResendInterval:   10,
        InlineClient:             true, // Enable direct publish from application
    })

    // Add authentication hook (AllowHook permits all connections - use only for development)
    if err := server.AddHook(new(auth.AllowHook), nil); err != nil {
        log.Fatal(err)
    }

    // Add TCP listener on port 1883
    tcp := listeners.NewTCP("tcp1", ":1883", nil)
    if err := server.AddListener(tcp); err != nil {
        log.Fatal(err)
    }

    // Start serving in background
    go func() {
        if err := server.Serve(); err != nil {
            log.Fatal(err)
        }
    }()

    <-done
    server.Log.Warn("shutting down...")
    server.Close()
}
```

--------------------------------

### Build Comqtt from Source

Source: https://github.com/wind-c/comqtt/blob/main/cluster/README.md

Compile the Comqtt binary from the command directory.

```shell
cd cmd
go build -o comqtt cluster/main.go
```

--------------------------------

### View Comqtt Help and Configuration Options

Source: https://github.com/wind-c/comqtt/blob/main/cluster/README.md

Display the available command-line flags and configuration parameters for the Comqtt broker.

```shell
./comqtt -h
Usage of ./comqtt:
  -conf string
        read the program parameters from the config file
        
  -storage-way uint
        storage way options:0 memory, 1 bolt, 2 badger, 3 redis (default 3)
  -auth-ds uint
        authentication datasource options:0 free, 1 redis, 2 mysql, 3 postgresql, 4 http
  -auth-path string
        config file path should correspond to the auth-datasource
  -auth-way uint
        authentication way options:0 anonymous, 1 username and password, 2 clientid

  -node-name string
        node name must be unique in the cluster
  -members string
        seeds member list of cluster,such as 192.168.0.103:7946,192.168.0.104:7946
  -bind-ip string
        the ip used for discovery and communication between nodes. It is usually set to the intranet ip addr. (default "127.0.0.1")
  -gossip-port int
        this port is used to discover nodes in a cluster
  -raft-port int
        this port is used for raft peer communication
  -raft-bootstrap bool
        should be true for the first node of the cluster. It can elect a leader without any other nodes being present. (default false)
  -grpc-enable bool
	    grpc is used for raft transport and reliable communication between nodes. (default false)
  -grpc-port int
        grpc communication port between nodes
        
  -http string
        network address for web info dashboard listener (default ":8080")
  -tcp string
        network address for mqtt tcp listener (default ":1883")
  -ws string
        network address for mqtt websocket listener (default ":1882")

  -redis string
        redis address for cluster mode (default "127.0.0.1:6379")
  -redis-db int
        redis db for cluster mode
  -redis-pass string
        redis password for cluster mode
  
  -log-enable
        log enabled or not (default true)
  -level int
        log level options:0Debug,1Info, 2Warn, 3Error, 4Fatal, 5Panic, 6NoLevel, 7Off (default 1)
  -env int
        app running environment:0 development or 1 production
  -error-file string
        error log filename (default "./logs/co-err.log")
  -info-file string
        info log filename (default "./logs/co-info.log")
```

--------------------------------

### Deploy Comqtt with Docker

Source: https://context7.com/wind-c/comqtt/llms.txt

Provides a Dockerfile for building the Go application and shell commands for running the container.

```dockerfile
# Dockerfile
FROM golang:1.22-alpine AS builder
WORKDIR /app
COPY . .
RUN go build -o comqtt ./cmd/single/main.go

FROM alpine:latest
WORKDIR /app
COPY --from=builder /app/comqtt .
COPY --from=builder /app/cmd/config ./config
EXPOSE 1883 1882 8080
CMD ["./comqtt", "--conf=./config/single.yml"]
```

```bash
# Build and run
docker build -t comqtt:latest .
docker run -p 1883:1883 -p 1882:1882 -p 8080:8080 comqtt:latest

# With docker-compose
docker-compose up -d
```

--------------------------------

### Running Unit Tests

Source: https://github.com/wind-c/comqtt/blob/main/README.md

Execute the project's unit test suite using the Go toolchain.

```bash
go run --cover ./...
```

--------------------------------

### Manage and Disconnect Clients in Go

Source: https://context7.com/wind-c/comqtt/llms.txt

Demonstrates how to inspect connected clients, retrieve their subscriptions, and disconnect specific clients using the server registry.

```go
package main

import (
    "fmt"
    "log"
    "time"

    "github.com/wind-c/comqtt/v2/mqtt"
    "github.com/wind-c/comqtt/v2/mqtt/hooks/auth"
    "github.com/wind-c/comqtt/v2/mqtt/listeners"
    "github.com/wind-c/comqtt/v2/mqtt/packets"
)

func main() {
    server := mqtt.New(nil)
    server.AddHook(new(auth.AllowHook), nil)

    tcp := listeners.NewTCP("t1", ":1883", nil)
    server.AddListener(tcp)

    go server.Serve()

    // Periodically inspect clients
    go func() {
        ticker := time.NewTicker(30 * time.Second)
        for range ticker.C {
            // Get all connected clients
            clients := server.Clients.GetAll()
            fmt.Printf("Connected clients: %d\n", len(clients))

            for id, cl := range clients {
                fmt.Printf("  Client: %s, Remote: %s, Username: %s\n",
                    id, cl.Net.Remote, string(cl.Properties.Username))

                // Get client subscriptions
                subs := cl.State.Subscriptions.GetAll()
                for filter, sub := range subs {
                    fmt.Printf("    Subscription: %s (QoS: %d)\n", filter, sub.Qos)
                }
            }
        }
    }()

    // Example: disconnect a specific client
    go func() {
        time.Sleep(time.Minute)
        if cl, ok := server.Clients.Get("problem-client"); ok {
            // Disconnect with reason code
            server.DisconnectClient(cl, packets.ErrNotAuthorized)
            fmt.Println("Disconnected problem-client")
        }
    }()

    // Example: add to blacklist (prevents reconnection)
    go func() {
        time.Sleep(2 * time.Minute)
        server.Blacklist = append(server.Blacklist, "banned-client")
        if cl, ok := server.Clients.Get("banned-client"); ok {
            server.DisconnectClient(cl, packets.ErrBanned)
        }
    }()

    select {}
}
```

--------------------------------

### Run Comqtt with Docker

Source: https://github.com/wind-c/comqtt/blob/main/README.md

Build and run the Comqtt broker container with standard ports exposed.

```sh
docker build -t comqtt:latest .
docker run -p 1883:1883 -p 1882:1882 -p 8080:8080 comqtt:latest
```

--------------------------------

### Running Performance Benchmarks

Source: https://github.com/wind-c/comqtt/blob/main/README.md

Commands to run performance benchmarks using the mqtt-stresser tool with varying client and message counts.

```bash
mqtt-stresser -broker tcp://localhost:1883 -num-clients=2 -num-messages=10000
```

```bash
mqtt-stresser -broker tcp://localhost:1883 -num-clients=10 -num-messages=10000
```

```bash
mqtt-stresser -broker tcp://localhost:1883 -num-clients=100 -num-messages=10000
```

--------------------------------

### Configure BadgerDB Storage Hook in CoMQTT

Source: https://context7.com/wind-c/comqtt/llms.txt

Initializes a CoMQTT server with the BadgerDB storage hook, specifying a local directory for data persistence.

```go
package main

import (
    "log"

    "github.com/wind-c/comqtt/v2/mqtt"
    "github.com/wind-c/comqtt/v2/mqtt/hooks/auth"
    "github.com/wind-c/comqtt/v2/mqtt/hooks/storage/badger"
    "github.com/wind-c/comqtt/v2/mqtt/listeners"
)

func main() {
    server := mqtt.New(nil)
    server.AddHook(new(auth.AllowHook), nil)

    // Add BadgerDB storage hook
    err := server.AddHook(new(badger.Hook), &badger.Options{
        Path: "./mqtt-data", // Directory for BadgerDB files
    })
    if err != nil {
        log.Fatal(err)
    }

    tcp := listeners.NewTCP("t1", ":1883", nil)
    server.AddListener(tcp)
    server.Serve()
}
```

--------------------------------

### Load Auth Rules from Configuration File

Source: https://context7.com/wind-c/comqtt/llms.txt

Loads authentication and ACL rules from a YAML or JSON file. The system automatically detects the file format based on the provided data.

```go
package main

import (
    "log"
    "os"

    "github.com/wind-c/comqtt/v2/mqtt"
    "github.com/wind-c/comqtt/v2/mqtt/hooks/auth"
    "github.com/wind-c/comqtt/v2/mqtt/listeners"
)

func main() {
    server := mqtt.New(nil)

    // Load auth rules from YAML file
    // auth.yaml contents:
    // auth:
    //   - username: "admin"
    //     password: "secret"
    //     allow: true
    //   - remote: "192.168.1.*:*"
    //     allow: true
    // acl:
    //   - username: "admin"
    //   - username: "guest"
    //     filters:
    //       "public/#": 3  # ReadWrite
    //       "private/#": 0 # Deny
    data, err := os.ReadFile("auth.yaml")
    if err != nil {
        log.Fatal(err)
    }

    err = server.AddHook(new(auth.Hook), &auth.Options{
        Data: data, // Automatically detects JSON or YAML format
    })
    if err != nil {
        log.Fatal(err)
    }

    tcp := listeners.NewTCP("t1", ":1883", nil)
    server.AddListener(tcp)
    server.Serve()
}
```

--------------------------------

### Configure Comqtt via YAML

Source: https://context7.com/wind-c/comqtt/llms.txt

Define server settings, storage backends, authentication methods, and MQTT protocol capabilities in a YAML configuration file.

```yaml
# single.yml - Complete server configuration
storage-way: 3        # 0: memory, 1: bolt, 2: badger, 3: redis
storage-path: ./mqtt.db
bridge-way: 0         # 0: disable, 1: kafka
bridge-path: ./bridge-kafka.yml
pprof-enable: false

auth:
  way: 1              # 0: anonymous, 1: username/password, 2: clientid
  datasource: 1       # 0: free, 1: redis, 2: mysql, 3: postgresql, 4: http
  conf-path: ./auth-redis.yml

mqtt:
  tcp: :1883
  ws: :1882
  http: :8080
  tls:
    ca-cert: ./certs/ca.crt
    server-cert: ./certs/server.crt
    server-key: ./certs/server.key
  options:
    client-write-buffer-size: 2048
    client-read-buffer-size: 2048
    sys-topic-resend-interval: 10
    inline-client: true
    capabilities:
      maximum-message-expiry-interval: 86400
      maximum-session-expiry-interval: 4294967295
      maximum-client-writes-pending: 65535
      maximum-packet-size: 0
      receive-maximum: 1024
      topic-alias-maximum: 65535
      maximum-qos: 2
      retain-available: 1
      wildcard-sub-available: 1
      sub-id-available: 1
      shared-sub-available: 1
      minimum-protocol-version: 3
      compatibilities:
        obscure-not-authorized: false
        passive-client-disconnect: false

redis:
  options:
    addr: localhost:6379
    password: ""
    db: 0
  prefix: comqtt

log:
  enable: true
  format: 1           # 0: text, 1: json
  output: 2           # 0: console, 1: file, 2: both
  filename: ./logs/comqtt.log
  maxsize: 100        # MB
  max-age: 30         # days
  max-backups: 10
  compress: true
  level: 0            # -4: debug, 0: info, 4: warn, 8: error
```

--------------------------------

### Publish a message directly in Go

Source: https://github.com/wind-c/comqtt/blob/main/README.md

Use the server.Publish method to send a message to a specific topic. The QoS byte limits the maximum QoS available to subscribers.

```go
err := server.Publish("direct/publish", []byte("packet scheduled message"), false, 0)
```

--------------------------------

### Configure BadgerDB Storage Hook

Source: https://github.com/wind-c/comqtt/blob/main/README.md

Adds a file-based BadgerDB persistence hook to the server.

```go
err := server.AddHook(new(badger.Hook), &badger.Options{
  Path: badgerPath,
})
if err != nil {
  log.Fatal(err)
}
```

--------------------------------

### Subscribe with Handler Functions in Go

Source: https://context7.com/wind-c/comqtt/llms.txt

Subscribe to MQTT topics and process incoming messages using inline handler functions. Ensure the server is initialized with `InlineClient: true` for this functionality.

```go
package main

import (
    "fmt"
    "log"
    "time"

    "github.com/wind-c/comqtt/v2/mqtt"
    "github.com/wind-c/comqtt/v2/mqtt/hooks/auth"
    "github.com/wind-c/comqtt/v2/mqtt/listeners"
    "github.com/wind-c/comqtt/v2/mqtt/packets"
)

func main() {
    server := mqtt.New(&mqtt.Options{
        InlineClient: true,
    })

    server.AddHook(new(auth.AllowHook), nil)
    tcp := listeners.NewTCP("t1", ":1883", nil)
    server.AddListener(tcp)

    go server.Serve()
    time.Sleep(time.Second)

    // Subscribe with handler function
    // Parameters: filter, subscriptionId, handler
    err := server.Subscribe("sensors/#", 1, func(cl *mqtt.Client, sub packets.Subscription, pk packets.Packet) {
        fmt.Printf("Received on %s: %s\n", pk.TopicName, string(pk.Payload))

        // Process the message
        if pk.TopicName == "sensors/temperature" {
            processTemperature(string(pk.Payload))
        }
    })
    if err != nil {
        log.Fatal(err)
    }

    // Subscribe to multiple topics with different handlers
    err = server.Subscribe("commands/+/action", 2, func(cl *mqtt.Client, sub packets.Subscription, pk packets.Packet) {
        fmt.Printf("Command received: %s -> %s\n", pk.TopicName, string(pk.Payload))
    })
    if err != nil {
        log.Fatal(err)
    }

    // Unsubscribe when done
    // server.Unsubscribe("sensors/#", 1)

    select {}
}

func processTemperature(value string) {
    fmt.Printf("Processing temperature: %s\n", value)
}
```

--------------------------------

### Run MQTT Stress Test

Source: https://github.com/wind-c/comqtt/blob/main/cluster/README.md

Command to run the mqtt-stresser tool to benchmark Comqtt performance. It specifies the broker address, number of clients, and messages to send.

```shell
mqtt-stresser -broker tcp://localhost:1883 -num-clients=500 -num-messages=100
```

--------------------------------

### Publish Messages Directly from Application

Source: https://context7.com/wind-c/comqtt/llms.txt

Enables the inline client to publish messages directly from the server instance. Requires setting InlineClient to true in the server options.

```go
package main

import (
    "log"
    "time"

    "github.com/wind-c/comqtt/v2/mqtt"
    "github.com/wind-c/comqtt/v2/mqtt/hooks/auth"
    "github.com/wind-c/comqtt/v2/mqtt/listeners"
)

func main() {
    // Enable inline client in options
    server := mqtt.New(&mqtt.Options{
        InlineClient: true,
    })

    server.AddHook(new(auth.AllowHook), nil)
    tcp := listeners.NewTCP("t1", ":1883", nil)
    server.AddListener(tcp)

    go server.Serve()

    // Wait for server to start
    time.Sleep(time.Second)

    // Publish messages from application
    go func() {
        ticker := time.NewTicker(5 * time.Second)
        for range ticker.C {
            // Simple publish: topic, payload, retain, qos
            err := server.Publish("sensors/temperature", []byte("23.5"), false, 0)
            if err != nil {
                log.Printf("publish error: %v", err)
            }

            // Publish with retain flag
            err = server.Publish("status/server", []byte("online"), true, 1)
            if err != nil {
                log.Printf("publish error: %v", err)
            }
        }
    }()

    select {} // Block forever
}
```

--------------------------------

### Implement Custom Event Hook in Go

Source: https://context7.com/wind-c/comqtt/llms.txt

Create a custom hook by embedding mqtt.HookBase and implementing the required interface methods to intercept connection and message events.

```go
package main

import (
    "bytes"
    "encoding/json"
    "fmt"
    "log"
    "net/http"

    "github.com/wind-c/comqtt/v2/mqtt"
    "github.com/wind-c/comqtt/v2/mqtt/hooks/auth"
    "github.com/wind-c/comqtt/v2/mqtt/listeners"
    "github.com/wind-c/comqtt/v2/mqtt/packets"
)

// Custom hook for event logging and message transformation
type EventLoggerHook struct {
    mqtt.HookBase
    webhookURL string
}

func (h *EventLoggerHook) ID() string {
    return "event-logger"
}

// Declare which hook methods this hook provides
func (h *EventLoggerHook) Provides(b byte) bool {
    return bytes.Contains([]byte{
        mqtt.OnConnect,
        mqtt.OnDisconnect,
        mqtt.OnSubscribed,
        mqtt.OnUnsubscribed,
        mqtt.OnPublish,
        mqtt.OnPublished,
    }, []byte{b})
}

// Initialize hook with configuration
func (h *EventLoggerHook) Init(config any) error {
    if cfg, ok := config.(map[string]string); ok {
        h.webhookURL = cfg["webhook_url"]
    }
    h.Log.Info("EventLoggerHook initialized", "webhook", h.webhookURL)
    return nil
}

// Called when a client connects
func (h *EventLoggerHook) OnConnect(cl *mqtt.Client, pk packets.Packet) error {
    h.Log.Info("client connecting",
        "client", cl.ID,
        "username", string(cl.Properties.Username),
        "remote", cl.Net.Remote,
    )

    // Optional: reject connection by returning an error
    // if cl.ID == "banned-client" {
    //     return packets.ErrNotAuthorized
    // }

    return nil
}

// Called when a client disconnects
func (h *EventLoggerHook) OnDisconnect(cl *mqtt.Client, err error, expire bool) {
    h.Log.Info("client disconnected",
        "client", cl.ID,
        "error", err,
        "session_expired", expire,
    )
}

// Called when a client subscribes
func (h *EventLoggerHook) OnSubscribed(cl *mqtt.Client, pk packets.Packet, reasonCodes []byte, counts []int) {
    for i, filter := range pk.Filters {
        h.Log.Info("client subscribed",
            "client", cl.ID,
            "filter", filter.Filter,
            "qos", reasonCodes[i],
            "subscriber_count", counts[i],
        )
    }
}

// Called when a client unsubscribes
func (h *EventLoggerHook) OnUnsubscribed(cl *mqtt.Client, pk packets.Packet, reasonCodes []byte, counts []int) {
    for _, filter := range pk.Filters {
        h.Log.Info("client unsubscribed", "client", cl.ID, "filter", filter.Filter)
    }
}

// Called when a client publishes - can modify the packet
func (h *EventLoggerHook) OnPublish(cl *mqtt.Client, pk packets.Packet) (packets.Packet, error) {
    h.Log.Info("message received",
        "client", cl.ID,
        "topic", pk.TopicName,
        "qos", pk.FixedHeader.Qos,
        "retain", pk.FixedHeader.Retain,
        "payload_size", len(pk.Payload),
    )

    // Example: transform messages on specific topics
    if pk.TopicName == "transform/uppercase" {
        pk.Payload = bytes.ToUpper(pk.Payload)
    }

    // Example: add metadata to JSON payloads
    if bytes.HasPrefix(pk.Payload, []byte("{")) {
        var data map[string]interface{}
        if json.Unmarshal(pk.Payload, &data) == nil {
            data["_processed_at"] = fmt.Sprintf("%d", time.Now().Unix())
            data["_client_id"] = cl.ID
            pk.Payload, _ = json.Marshal(data)
        }
    }

    // Example: reject messages to restricted topics
    // if strings.HasPrefix(pk.TopicName, "restricted/") {
    //     return pk, packets.ErrNotAuthorized
    // }

    return pk, nil
}

// Called after a message is published to subscribers
func (h *EventLoggerHook) OnPublished(cl *mqtt.Client, pk packets.Packet) {
    // Send to webhook for external processing
    if h.webhookURL != "" {
        go h.sendWebhook(cl.ID, pk.TopicName, pk.Payload)
    }
}

func (h *EventLoggerHook) sendWebhook(clientID, topic string, payload []byte) {
    data, _ := json.Marshal(map[string]interface{}{
        "client_id": clientID,
        "topic":     topic,
        "payload":   string(payload),
    })
    http.Post(h.webhookURL, "application/json", bytes.NewReader(data))
}

func main() {
    server := mqtt.New(nil)
    server.AddHook(new(auth.AllowHook), nil)

    // Add custom hook with configuration
    err := server.AddHook(new(EventLoggerHook), map[string]string{
        "webhook_url": "https://example.com/mqtt-events",
    })
    if err != nil {
        log.Fatal(err)
    }

    tcp := listeners.NewTCP("t1", ":1883", nil)
    server.AddListener(tcp)
    server.Serve()
}
```

--------------------------------

### Add Network Listeners

Source: https://context7.com/wind-c/comqtt/llms.txt

Configures multiple listener types including TCP, WebSocket, TLS-encrypted connections, and HTTP-based stats or health checks. Requires a valid TLS configuration for secure listeners.

```go
package main

import (
    "crypto/tls"
    "log"

    "github.com/wind-c/comqtt/v2/mqtt"
    "github.com/wind-c/comqtt/v2/mqtt/hooks/auth"
    "github.com/wind-c/comqtt/v2/mqtt/listeners"
)

func main() {
    server := mqtt.New(nil)
    server.AddHook(new(auth.AllowHook), nil)

    // TCP listener on standard MQTT port
    tcp := listeners.NewTCP("tcp1", ":1883", nil)
    if err := server.AddListener(tcp); err != nil {
        log.Fatal(err)
    }

    // WebSocket listener for browser clients
    ws := listeners.NewWebsocket("ws1", ":1882", nil)
    if err := server.AddListener(ws); err != nil {
        log.Fatal(err)
    }

    // TCP with TLS encryption
    tlsConfig := &tls.Config{
        Certificates: []tls.Certificate{loadCertificate()},
        MinVersion:   tls.VersionTLS12,
    }
    tcpTLS := listeners.NewTCP("tcp-tls", ":8883", &listeners.Config{
        TLSConfig: tlsConfig,
    })
    if err := server.AddListener(tcpTLS); err != nil {
        log.Fatal(err)
    }

    // WebSocket with TLS (WSS)
    wss := listeners.NewWebsocket("wss1", ":8884", &listeners.Config{
        TLSConfig: tlsConfig,
    })
    if err := server.AddListener(wss); err != nil {
        log.Fatal(err)
    }

    // HTTP Stats dashboard
    stats := listeners.NewHTTPStats("stats", ":8080", nil, server.Info)
    if err := server.AddListener(stats); err != nil {
        log.Fatal(err)
    }

    // Health check endpoint for load balancers
    health := listeners.NewHTTPHealthCheck("health", ":8081", nil)
    if err := server.AddListener(health); err != nil {
        log.Fatal(err)
    }

    server.Serve()
}

func loadCertificate() tls.Certificate {
    cert, _ := tls.LoadX509KeyPair("server.crt", "server.key")
    return cert
}
```

--------------------------------

### Configure Auth Ledger with Programmatic Rules

Source: https://context7.com/wind-c/comqtt/llms.txt

Defines authentication and ACL rules directly within the Go code. Rules are evaluated in order, and the first match determines the outcome.

```go
package main

import (
    "log"

    "github.com/wind-c/comqtt/v2/mqtt"
    "github.com/wind-c/comqtt/v2/mqtt/hooks/auth"
    "github.com/wind-c/comqtt/v2/mqtt/listeners"
)

func main() {
    server := mqtt.New(nil)

    // Configure auth ledger with rules
    err := server.AddHook(new(auth.Hook), &auth.Options{
        Ledger: &auth.Ledger{
            // Authentication rules - checked in order, first match wins
            Auth: auth.AuthRules{
                {Username: "admin", Password: "secret123", Allow: true},
                {Username: "sensor", Password: "sensor-pass", Allow: true},
                {Username: "readonly", Password: "read123", Allow: true},
                {Remote: "127.0.0.1:*", Allow: true},        // Allow localhost
                {Remote: "192.168.1.*:*", Allow: true},      // Allow local network
            },
            // ACL rules for topic access control
            ACL: auth.ACLRules{
                // Admin can access everything
                {Username: "admin"},
                // Sensors can only publish to sensor topics
                {
                    Username: "sensor",
                    Filters: auth.Filters{
                        "sensors/#":     auth.WriteOnly,
                        "sensors/+/cmd": auth.ReadOnly,
                    },
                },
                // Readonly user can only subscribe
                {
                    Username: "readonly",
                    Filters: auth.Filters{
                        "#": auth.ReadOnly,
                    },
                },
                // Default: deny publish to system topics
                {
                    Filters: auth.Filters{
                        "$SYS/#": auth.Deny,
                        "#":      auth.ReadWrite,
                    },
                },
            },
        },
    })
    if err != nil {
        log.Fatal(err)
    }

    tcp := listeners.NewTCP("t1", ":1883", nil)
    server.AddListener(tcp)
    server.Serve()
}
```

--------------------------------

### Configure Redis Storage Hook

Source: https://github.com/wind-c/comqtt/blob/main/README.md

Adds a Redis persistence hook to the server using the go-redis/v9 client.

```go
err := server.AddHook(new(redis.Hook), &redis.Options{
  Options: &rv8.Options{
    Addr:     "localhost:6379", // default redis address
    Password: "",               // your password
    DB:       0,                // your redis db
  },
})
if err != nil {
  log.Fatal(err)
}
```

--------------------------------

### Server Hooks

Source: https://github.com/wind-c/comqtt/blob/main/README.md

List of available hooks for managing client sessions, persistence, and authentication.

```APIDOC
## Server Hooks Overview

### Lifecycle Hooks
- **OnPacketIDExhausted**: Called when a client runs out of unused packet ids.
- **OnWill**: Called when a client disconnects and intends to issue a will message.
- **OnWillSent**: Called when an LWT message has been issued.
- **OnClientExpired**: Called when a client session has expired.
- **OnRetainedExpired**: Called when a retained message has expired.

### Persistence Hooks
- **StoredClients**: Returns clients from a persistent store.
- **StoredSubscriptions**: Returns client subscriptions from a persistent store.
- **StoredInflightMessages**: Returns inflight messages from a persistent store.
- **StoredRetainedMessages**: Returns retained messages from a persistent store.
- **StoredSysInfo**: Returns stored system info values.

### Authentication Hooks
- **OnACLCheck**: Used for ACL validation.
- **OnConnectAuthenticate**: Used for connection authentication.
```

--------------------------------

### Load Auth Ledger from Encoded Data

Source: https://github.com/wind-c/comqtt/blob/main/README.md

Initializes the Auth Ledger hook using pre-encoded JSON or YAML data.

```go
err = server.AddHook(new(auth.Hook), &auth.Options{
    Data: data, // build ledger from byte slice: yaml or json
})
```

--------------------------------

### Configure Auth Ledger Rules

Source: https://github.com/wind-c/comqtt/blob/main/README.md

Defines connection and ACL rules using the Auth Ledger hook. Rules are processed in index order, and Auth disallows all by default while ACL allows all by default.

```go
server := mqtt.New(nil)
err := server.AddHook(new(auth.Hook), &auth.Options{
    Ledger: &auth.Ledger{
    Auth: auth.AuthRules{ // Auth disallows all by default
      {Username: "peach", Password: "password1", Allow: true},
      {Username: "melon", Password: "password2", Allow: true},
      {Remote: "127.0.0.1:*", Allow: true},
      {Remote: "localhost:*", Allow: true},
    },
    ACL: auth.ACLRules{ // ACL allows all by default
      {Remote: "127.0.0.1:*"}, // local superuser allow all
      {
        // user melon can read and write to their own topic
        Username: "melon", Filters: auth.Filters{
          "melon/#":   auth.ReadWrite,
          "updates/#": auth.WriteOnly, // can write to updates, but can't read updates from others
        },
      },
      {
        // Otherwise, no clients have publishing permissions
        Filters: auth.Filters{
          "#":         auth.ReadOnly,
          "updates/#": auth.Deny,
        },
      },
    },
  }
})
```

--------------------------------

### Generate Bcrypt Password Hash in Go

Source: https://github.com/wind-c/comqtt/blob/main/README.md

Go snippet to generate a bcrypt hashed password for MQTT clients. Requires the 'golang.org/x/crypto/bcrypt' package.

```go
import "golang.org/x/crypto/bcrypt"
hashed, err := bcrypt.GenerateFromPassword(pwd, bcrypt.DefaultCost)
if err != nil {
	return 
}
println("Password hash for MQTT client: ", hashed)
```

--------------------------------

### Inject Publish Packet in Go

Source: https://context7.com/wind-c/comqtt/llms.txt

Use `InjectPacket()` to send custom MQTT packets, such as publish packets with v5 properties, as if they originated from a specific client. This requires creating an inline client first.

```go
package main

import (
    "log"
    "time"

    "github.com/wind-c/comqtt/v2/mqtt"
    "github.com/wind-c/comqtt/v2/mqtt/hooks/auth"
    "github.com/wind-c/comqtt/v2/mqtt/listeners"
    "github.com/wind-c/comqtt/v2/mqtt/packets"
)

func main() {
    server := mqtt.New(&mqtt.Options{
        InlineClient: true,
    })

    server.AddHook(new(auth.AllowHook), nil)
    tcp := listeners.NewTCP("t1", ":1883", nil)
    server.AddListener(tcp)

    go server.Serve()
    time.Sleep(time.Second)

    // Create an inline client for injection
    // Parameters: conn (nil for inline), listener, clientId, inline flag
    cl := server.NewClient(nil, "local", "system-publisher", true)

    // Inject a publish packet with MQTT v5 properties
    err := server.InjectPacket(cl, packets.Packet{
        FixedHeader: packets.FixedHeader{
            Type:   packets.Publish,
            Qos:    1,
            Retain: true,
        },
        TopicName: "system/announcements",
        Payload:   []byte(`{"message": "System maintenance at 3AM"}`),
        Properties: packets.Properties{
            ContentType:           "application/json",
            MessageExpiryInterval: 3600,
            User: []packets.UserProperty{
                {Key: "priority", Val: "high"},
                {Key: "source", Val: "system"},
            },
        },
    })
    if err != nil {
        log.Printf("inject error: %v", err)
    }

    select {}
}
```

--------------------------------

### Injecting MQTT Packets

Source: https://github.com/wind-c/comqtt/blob/main/README.md

Use the inline client flag to bypass ACL and topic validation checks when injecting custom MQTT packets directly into the runtime.

```go
cl := server.NewClient(nil, "local", "inline", true)
server.InjectPacket(cl, packets.Packet{
  FixedHeader: packets.FixedHeader{
    Type: packets.Publish,
  },
  TopicName: "direct/publish",
  Payload: []byte("scheduled message"),
})
```

--------------------------------

### Configure Redis Persistent Storage Hook in Go

Source: https://context7.com/wind-c/comqtt/llms.txt

Integrate the Redis hook for distributed persistent storage of client sessions, subscriptions, and messages. Ensure Redis is accessible at the specified address and credentials.

```go
package main

import (
    "log"

    "github.com/wind-c/comqtt/v2/mqtt"
    "github.com/wind-c/comqtt/v2/mqtt/hooks/auth"
    "github.com/wind-c/comqtt/v2/mqtt/hooks/storage/redis"
    "github.com/wind-c/comqtt/v2/mqtt/listeners"

    rv9 "github.com/redis/go-redis/v9"
)

func main() {
    server := mqtt.New(nil)
    server.AddHook(new(auth.AllowHook), nil)

    // Add Redis storage hook
    err := server.AddHook(new(redis.Hook), &redis.Options{
        HPrefix: "mqtt:", // Key prefix for all Redis keys
        Options: &rv9.Options{
            Addr:     "localhost:6379",
            Password: "your-redis-password",
            DB:       0,
            PoolSize: 10,
        },
    })
    if err != nil {
        log.Fatal(err)
    }

    tcp := listeners.NewTCP("t1", ":1883", nil)
    server.AddListener(tcp)
    server.Serve()
}

// Redis stores the following data:
// - mqtt:cl - Client sessions (hash)
// - mqtt:su - Subscriptions (hash)
// - mqtt:rt - Retained messages (hash)
// - mqtt:if - Inflight QoS messages (hash)
// - mqtt:sy - System info (hash)
```

--------------------------------

### Integrate REST API in Custom Go Server

Source: https://context7.com/wind-c/comqtt/llms.txt

Embed the Comqtt REST API into a custom Go application by registering handlers with an HTTP multiplexer. Ensure InlineClient is enabled in the MQTT options for REST publishing functionality.

```go
package main

import (
    "log"
    "net/http"

    "github.com/wind-c/comqtt/v2/mqtt"
    "github.com/wind-c/comqtt/v2/mqtt/hooks/auth"
    "github.com/wind-c/comqtt/v2/mqtt/listeners"
    "github.com/wind-c/comqtt/v2/mqtt/rest"
)

func main() {
    server := mqtt.New(&mqtt.Options{
        InlineClient: true, // Required for REST publish
    })
    server.AddHook(new(auth.AllowHook), nil)

    // Setup MQTT listener
    tcp := listeners.NewTCP("t1", ":1883", nil)
    server.AddListener(tcp)

    // Setup REST API
    restHandler := rest.New(server)
    mux := http.NewServeMux()

    // Register all REST endpoints
    for pattern, handler := range restHandler.GenHandlers() {
        mux.HandleFunc(pattern, handler)
    }

    // Add custom endpoints
    mux.HandleFunc("GET /api/custom/status", func(w http.ResponseWriter, r *http.Request) {
        w.Write([]byte("OK"))
    })

    // Start HTTP server
    go func() {
        log.Println("REST API listening on :8080")
        http.ListenAndServe(":8080", mux)
    }()

    server.Serve()
}
```
Tokens:
Raw