MQTTBroker-cli

1.0.0

MQTTBroker-cli.ps1

<#PSScriptInfo
.VERSION 1.0.0
.GUID 9c4b2e71-6a3d-4f85-b1e0-7d5a8c3f2e96
.AUTHOR ConnectMQ Team
.COMPANYNAME ConnectMQ
.COPYRIGHT (c) 2026 ConnectMQ. MIT License.
.TAGS MQTT IoT Broker PublicBroker ConnectMQ PubSub Messaging TLS
.LICENSEURI https://opensource.org/licenses/MIT
.PROJECTURI https://connectmq.com/blog/free-public-mqtt-broker
.RELEASENOTES Initial release: publish, subscribe, test and broker info for broker.connectmq.com over TCP or TLS.
#>

<#
.SYNOPSIS
Single-file MQTT command-line client for the free ConnectMQ public broker (broker.connectmq.com).

.DESCRIPTION
Publish, subscribe and test MQTT 3.1.1 over TCP (1883) or TLS (8883) without installing anything else.
Works on Windows PowerShell 5.1 and PowerShell 7+. The public broker is anonymous: anything you publish
can be read by anyone, so use a random topic prefix. For production, create a private broker at https://connectmq.com.
The same functions are available as a module: Install-Module MQTTBroker

.PARAMETER Action
Test (default), Publish, Subscribe or Info.

.EXAMPLE
MQTTBroker-cli.ps1
Checks connectivity and message round-trip against broker.connectmq.com.

.EXAMPLE
MQTTBroker-cli.ps1 -Action Subscribe -Topic 'connectmq/demo/7f3a91/#' -Tls

.EXAMPLE
MQTTBroker-cli.ps1 -Action Publish -Topic 'connectmq/demo/7f3a91/temp' -Message '{"value":21.5}' -Qos 1 -Tls
#>
[CmdletBinding()]
param(
    [ValidateSet('Test', 'Publish', 'Subscribe', 'Info')][string]$Action = 'Test',
    [string[]]$Topic,
    [string]$Message,
    [ValidateSet(0, 1, 2)][int]$Qos = 0,
    [switch]$Retain,
    [switch]$Tls,
    [int]$Count = 0,
    [int]$TimeoutSeconds = 0,
    [string]$BrokerHost = 'broker.connectmq.com',
    [string]$Username,
    [securestring]$Password
)

#region ConnectMQ MQTT core
# Minimal, dependency-free MQTT 3.1.1 client for the ConnectMQ public broker (broker.connectmq.com).
# Works on Windows PowerShell 5.1 and PowerShell 7+. Transports: TCP (1883) and TLS (8883).

$script:ConnectMQDefaultHost = 'broker.connectmq.com'

function ConvertTo-CmqRemainingLength {
    param([int]$Length)
    $bytes = New-Object System.Collections.Generic.List[byte]
    do {
        $digit = $Length % 128
        $Length = [math]::Floor($Length / 128)
        if ($Length -gt 0) { $digit = $digit -bor 0x80 }
        $bytes.Add([byte]$digit)
    } while ($Length -gt 0)
    return , $bytes.ToArray()
}

function ConvertTo-CmqString {
    param([string]$Value)
    $data = [Text.Encoding]::UTF8.GetBytes($Value)
    return , ([byte[]]@([byte]($data.Length -shr 8), [byte]($data.Length -band 0xFF)) + $data)
}

function New-CmqPacket {
    param([byte]$Header, [byte[]]$Body = @())
    return , ([byte[]]@($Header) + (ConvertTo-CmqRemainingLength $Body.Length) + $Body)
}

function Write-CmqPacket {
    param($Connection, [byte[]]$Packet)
    $Connection.Stream.Write($Packet, 0, $Packet.Length)
    $Connection.Stream.Flush()
    $Connection.LastSent = [DateTime]::UtcNow
}

function Read-CmqBytes {
    param($Stream, [int]$Count)
    $buffer = New-Object byte[] $Count
    $offset = 0
    while ($offset -lt $Count) {
        $read = $Stream.Read($buffer, $offset, $Count - $offset)
        if ($read -le 0) { throw 'Connection closed by broker.' }
        $offset += $read
    }
    return , $buffer
}

function Read-CmqPacketBody {
    # Reads remaining length + body after the fixed header byte has been received.
    param($Connection, [byte]$Header)
    $multiplier = 1; $length = 0
    do {
        $b = (Read-CmqBytes $Connection.Stream 1)[0]
        $length += ($b -band 0x7F) * $multiplier
        $multiplier *= 128
    } while ($b -band 0x80)
    $body = if ($length -gt 0) { Read-CmqBytes $Connection.Stream $length } else { [byte[]]@() }
    return [pscustomobject]@{ Type = $Header -shr 4; Flags = $Header -band 0x0F; Body = $body }
}

function Read-CmqPacket {
    # Waits up to TimeoutMs for the next packet, sending PINGREQ when keep-alive is due. Returns $null on timeout.
    param($Connection, [int]$TimeoutMs = 10000)
    $deadline = [DateTime]::UtcNow.AddMilliseconds($TimeoutMs)
    while ($true) {
        if (-not $Connection.PendingRead) {
            $Connection.HeaderBuffer = New-Object byte[] 1
            $Connection.PendingRead = $Connection.Stream.ReadAsync($Connection.HeaderBuffer, 0, 1)
        }
        $remaining = [int]($deadline - [DateTime]::UtcNow).TotalMilliseconds
        if ($remaining -le 0) { return $null }
        if ($Connection.PendingRead.Wait([math]::Min($remaining, 1000))) {
            $task = $Connection.PendingRead
            $Connection.PendingRead = $null
            if ($task.Result -le 0) { throw 'Connection closed by broker.' }
            return Read-CmqPacketBody $Connection $Connection.HeaderBuffer[0]
        }
        if ($Connection.KeepAlive -gt 0 -and ([DateTime]::UtcNow - $Connection.LastSent).TotalSeconds -ge ($Connection.KeepAlive * 0.75)) {
            Write-CmqPacket $Connection ([byte[]]@(0xC0, 0x00))
        }
    }
}

function Get-CmqPacketId {
    param($Connection)
    $Connection.NextPacketId = ($Connection.NextPacketId % 65535) + 1
    return $Connection.NextPacketId
}

function Wait-CmqPacket {
    # Waits for a packet of the given type (and packet id), dispatching incoming PUBLISH packets to the queue.
    param($Connection, [int]$Type, [int]$PacketId = -1, [int]$TimeoutMs = 10000)
    $deadline = [DateTime]::UtcNow.AddMilliseconds($TimeoutMs)
    while ($true) {
        $remaining = [int]($deadline - [DateTime]::UtcNow).TotalMilliseconds
        $packet = if ($remaining -gt 0) { Read-CmqPacket $Connection $remaining }
        if (-not $packet) { throw "Timed out waiting for MQTT packet type $Type." }
        if ($packet.Type -eq 3) { $msg = ConvertFrom-CmqPublish $Connection $packet; if ($msg) { $Connection.Queue.Enqueue($msg) }; continue }
        if ($packet.Type -ne $Type) { continue }
        if ($PacketId -ge 0 -and (($packet.Body[0] -shl 8) -bor $packet.Body[1]) -ne $PacketId) { continue }
        return $packet
    }
}

function ConvertFrom-CmqPublish {
    param($Connection, $Packet)
    $body = $Packet.Body
    $qos = ($Packet.Flags -shr 1) -band 0x03
    $topicLength = ($body[0] -shl 8) -bor $body[1]
    $topic = [Text.Encoding]::UTF8.GetString($body, 2, $topicLength)
    $index = 2 + $topicLength
    $packetId = 0
    if ($qos -gt 0) { $packetId = ($body[$index] -shl 8) -bor $body[$index + 1]; $index += 2 }
    $payload = New-Object byte[] ($body.Length - $index)
    [Array]::Copy($body, $index, $payload, 0, $payload.Length)
    $idBytes = [byte[]]@([byte]($packetId -shr 8), [byte]($packetId -band 0xFF))
    if ($qos -eq 1) { Write-CmqPacket $Connection (New-CmqPacket 0x40 $idBytes) }
    if ($qos -eq 2) {
        Write-CmqPacket $Connection (New-CmqPacket 0x50 $idBytes)
        [void](Wait-CmqPacket $Connection 6 $packetId)
        Write-CmqPacket $Connection (New-CmqPacket 0x70 $idBytes)
    }
    [pscustomobject]@{
        PSTypeName = 'MQTTBroker.Message'
        Topic      = $topic
        Payload    = [Text.Encoding]::UTF8.GetString($payload)
        Qos        = $qos
        Retain     = [bool]($Packet.Flags -band 0x01)
        Received   = Get-Date
        RawPayload = $payload
    }
}
#endregion

function Get-MQTTBrokerInfo {
    <#
    .SYNOPSIS
    Shows connection details and limits of the free ConnectMQ public MQTT broker.
    .EXAMPLE
    Get-MQTTBrokerInfo
    #>
    [CmdletBinding()]
    param()
    [pscustomobject]@{
        Host           = $script:ConnectMQDefaultHost
        Mqtt           = "mqtt://$($script:ConnectMQDefaultHost):1883"
        MqttTls        = "mqtts://$($script:ConnectMQDefaultHost):8883"
        WebSocket      = "ws://$($script:ConnectMQDefaultHost):8083/mqtt"
        SecureSocket   = "wss://$($script:ConnectMQDefaultHost):8084/mqtt"
        Authentication = 'Anonymous (non-empty client ID required)'
        MaxMessageSize = '64 KB'
        PublishRate    = '10 messages/second per client'
        Subscriptions  = '10 per client'
        MaxTopicDepth  = 8
        QosLevels      = '0, 1, 2'
        Note           = 'Public area: anyone can read your messages. Get a private broker at https://connectmq.com'
    }
}

function New-MQTTTopicPrefix {
    <#
    .SYNOPSIS
    Generates a random, collision-free topic prefix for testing on the public broker.
    .EXAMPLE
    $prefix = New-MQTTTopicPrefix # connectmq/demo/7f3a91
    #>
    [CmdletBinding()]
    [OutputType([string])]
    param([string]$Root = 'connectmq/demo')
    "$Root/$([guid]::NewGuid().ToString('N').Substring(0, 6))"
}

function Connect-MQTTBroker {
    <#
    .SYNOPSIS
    Opens an MQTT 3.1.1 connection to broker.connectmq.com (or a private ConnectMQ broker).
    .DESCRIPTION
    The public broker accepts anonymous clients with a non-empty client ID. Use -Tls for port 8883.
    Pass -Username/-Password and -BrokerHost to connect to a private broker created on https://connectmq.com.
    .EXAMPLE
    $c = Connect-MQTTBroker -Tls
    .EXAMPLE
    $c = Connect-MQTTBroker -BrokerHost my-broker.connectmq.com -Tls -Username dev -Password (Read-Host -AsSecureString)
    #>
    [CmdletBinding()]
    param(
        [string]$BrokerHost = $script:ConnectMQDefaultHost,
        [int]$Port,
        [switch]$Tls,
        [string]$ClientId = "ps-$([guid]::NewGuid().ToString('N').Substring(0, 8))",
        [string]$Username,
        [object]$Password,
        [ValidateRange(0, 65535)][int]$KeepAlive = 60,
        [switch]$CleanSession = $true,
        [int]$TimeoutSeconds = 10
    )
    if (-not $Port) { $Port = if ($Tls) { 8883 } else { 1883 } }
    if (-not $ClientId) { throw 'ClientId must not be empty.' }

    $tcp = New-Object System.Net.Sockets.TcpClient
    $connectTask = $tcp.ConnectAsync($BrokerHost, $Port)
    if (-not $connectTask.Wait($TimeoutSeconds * 1000)) { $tcp.Close(); throw "Timed out connecting to $($BrokerHost):$Port." }
    $tcp.NoDelay = $true
    $stream = $tcp.GetStream()
    if ($Tls) {
        $ssl = New-Object System.Net.Security.SslStream($stream, $false)
        $ssl.AuthenticateAsClient($BrokerHost, $null, [System.Security.Authentication.SslProtocols]::Tls12, $true)
        $stream = $ssl
    }
    $stream.ReadTimeout = $TimeoutSeconds * 1000
    $stream.WriteTimeout = $TimeoutSeconds * 1000

    $flags = 0
    if ($CleanSession) { $flags = $flags -bor 0x02 }
    $payload = ConvertTo-CmqString $ClientId
    if ($Username) {
        $flags = $flags -bor 0x80
        $payload += ConvertTo-CmqString $Username
        if ($Password) {
            $flags = $flags -bor 0x40
            $plain = if ($Password -is [securestring]) { [Net.NetworkCredential]::new('', $Password).Password } else { [string]$Password }
            $payload += ConvertTo-CmqString $plain
        }
    }
    $variable = (ConvertTo-CmqString 'MQTT') + [byte[]]@(0x04, [byte]$flags, [byte]($KeepAlive -shr 8), [byte]($KeepAlive -band 0xFF))

    $connection = [pscustomobject]@{
        PSTypeName   = 'MQTTBroker.Connection'
        BrokerHost   = $BrokerHost
        Port         = $Port
        Tls          = [bool]$Tls
        ClientId     = $ClientId
        KeepAlive    = $KeepAlive
        Client       = $tcp
        Stream       = $stream
        LastSent     = [DateTime]::UtcNow
        NextPacketId = 0
        PendingRead  = $null
        HeaderBuffer = $null
        Queue        = New-Object System.Collections.Queue
    }
    Write-CmqPacket $connection (New-CmqPacket 0x10 ($variable + $payload))
    $connack = Wait-CmqPacket $connection 2 -TimeoutMs ($TimeoutSeconds * 1000)
    $code = $connack.Body[1]
    if ($code -ne 0) {
        $tcp.Close()
        $reason = @{ 1 = 'unacceptable protocol version'; 2 = 'identifier rejected'; 3 = 'server unavailable'; 4 = 'bad user name or password'; 5 = 'not authorized' }[[int]$code]
        throw "Broker refused connection (code $code): $reason"
    }
    Write-Verbose "Connected to $($BrokerHost):$Port as $ClientId"
    $connection
}

function Disconnect-MQTTBroker {
    <#
    .SYNOPSIS
    Sends DISCONNECT and closes the connection.
    .EXAMPLE
    Disconnect-MQTTBroker $c
    #>
    [CmdletBinding()]
    param([Parameter(Mandatory, ValueFromPipeline)]$Connection)
    process {
        try { Write-CmqPacket $Connection ([byte[]]@(0xE0, 0x00)) } catch { Write-Verbose $_ }
        $Connection.Stream.Dispose()
        $Connection.Client.Close()
    }
}

function Publish-MQTTMessage {
    <#
    .SYNOPSIS
    Publishes a message to an MQTT topic on broker.connectmq.com.
    .DESCRIPTION
    Without -Connection a temporary connection is opened and closed automatically.
    .EXAMPLE
    Publish-MQTTMessage -Topic connectmq/demo/7f3a91/temp -Message '{"value":21.5}'
    .EXAMPLE
    $c = Connect-MQTTBroker -Tls; 1..5 | % { Publish-MQTTMessage -Connection $c -Topic demo/x -Message $_ -Qos 1 }; Disconnect-MQTTBroker $c
    #>
    [CmdletBinding()]
    param(
        [Parameter(Mandatory)][string]$Topic,
        [Parameter(Mandatory, ValueFromPipeline)][AllowEmptyString()][object]$Message,
        [ValidateSet(0, 1, 2)][int]$Qos = 0,
        [switch]$Retain,
        $Connection,
        [switch]$Tls,
        [string]$BrokerHost = $script:ConnectMQDefaultHost
    )
    begin {
        $own = -not $Connection
        if ($own) { $Connection = Connect-MQTTBroker -BrokerHost $BrokerHost -Tls:$Tls }
    }
    process {
        $payload = if ($Message -is [byte[]]) { $Message } else { [Text.Encoding]::UTF8.GetBytes([string]$Message) }
        $body = ConvertTo-CmqString $Topic
        $packetId = 0
        if ($Qos -gt 0) {
            $packetId = Get-CmqPacketId $Connection
            $body += [byte[]]@([byte]($packetId -shr 8), [byte]($packetId -band 0xFF))
        }
        $header = 0x30 -bor ($Qos -shl 1)
        if ($Retain) { $header = $header -bor 0x01 }
        Write-CmqPacket $Connection (New-CmqPacket ([byte]$header) ($body + $payload))
        if ($Qos -eq 1) { [void](Wait-CmqPacket $Connection 4 $packetId) }
        if ($Qos -eq 2) {
            [void](Wait-CmqPacket $Connection 5 $packetId)
            Write-CmqPacket $Connection (New-CmqPacket 0x62 ([byte[]]@([byte]($packetId -shr 8), [byte]($packetId -band 0xFF))))
            [void](Wait-CmqPacket $Connection 7 $packetId)
        }
        Write-Verbose "Published $($payload.Length) bytes to $Topic (QoS $Qos)"
    }
    end { if ($own) { Disconnect-MQTTBroker $Connection } }
}

function Receive-MQTTMessage {
    <#
    .SYNOPSIS
    Subscribes to one or more topic filters and streams incoming messages to the pipeline.
    .DESCRIPTION
    Runs until -Count messages are received, -TimeoutSeconds elapses, or Ctrl+C is pressed.
    Without -Connection a temporary connection is opened and closed automatically.
    .EXAMPLE
    Receive-MQTTMessage -Topic 'connectmq/demo/7f3a91/#'
    .EXAMPLE
    Receive-MQTTMessage -Topic 'connectmq/demo/7f3a91/temp' -Count 1 -TimeoutSeconds 30 | Select-Object -Expand Payload
    #>
    [CmdletBinding()]
    param(
        [Parameter(Mandatory)][string[]]$Topic,
        [ValidateSet(0, 1, 2)][int]$Qos = 0,
        [int]$Count = 0,
        [int]$TimeoutSeconds = 0,
        $Connection,
        [switch]$Tls,
        [string]$BrokerHost = $script:ConnectMQDefaultHost
    )
    $own = -not $Connection
    if ($own) { $Connection = Connect-MQTTBroker -BrokerHost $BrokerHost -Tls:$Tls }
    try {
        $packetId = Get-CmqPacketId $Connection
        $body = [byte[]]@([byte]($packetId -shr 8), [byte]($packetId -band 0xFF))
        foreach ($filter in $Topic) { $body += (ConvertTo-CmqString $filter) + [byte[]]@([byte]$Qos) }
        Write-CmqPacket $Connection (New-CmqPacket 0x82 $body)
        $suback = Wait-CmqPacket $Connection 9 $packetId
        for ($i = 2; $i -lt $suback.Body.Length; $i++) {
            if ($suback.Body[$i] -eq 0x80) { throw "Subscription to '$($Topic[$i - 2])' was rejected by the broker." }
        }
        Write-Verbose "Subscribed to $($Topic -join ', ')"

        $received = 0
        $deadline = if ($TimeoutSeconds -gt 0) { [DateTime]::UtcNow.AddSeconds($TimeoutSeconds) } else { [DateTime]::MaxValue }
        while (($Count -eq 0 -or $received -lt $Count) -and [DateTime]::UtcNow -lt $deadline) {
            if ($Connection.Queue.Count -gt 0) { $Connection.Queue.Dequeue(); $received++; continue }
            $wait = [int][math]::Min(1000, ($deadline - [DateTime]::UtcNow).TotalMilliseconds)
            $packet = Read-CmqPacket $Connection $wait
            if ($packet -and $packet.Type -eq 3) { ConvertFrom-CmqPublish $Connection $packet; $received++ }
        }
    }
    finally {
        if ($own) { Disconnect-MQTTBroker $Connection }
    }
}

function Test-MQTTBroker {
    <#
    .SYNOPSIS
    Checks that the ConnectMQ public broker is reachable and round-trips a message.
    .EXAMPLE
    Test-MQTTBroker
    .EXAMPLE
    Test-MQTTBroker -Tls
    #>
    [CmdletBinding()]
    param([switch]$Tls, [string]$BrokerHost = $script:ConnectMQDefaultHost)
    $watch = [Diagnostics.Stopwatch]::StartNew()
    $result = [pscustomobject]@{ BrokerHost = $BrokerHost; Port = $(if ($Tls) { 8883 } else { 1883 }); Tls = [bool]$Tls; Connected = $false; RoundTrip = $false; ConnectMs = $null; RoundTripMs = $null; Error = $null }
    $c = $null
    try {
        $c = Connect-MQTTBroker -BrokerHost $BrokerHost -Tls:$Tls
        $result.Connected = $true
        $result.ConnectMs = $watch.ElapsedMilliseconds
        $topic = "$(New-MQTTTopicPrefix)/ping"
        $token = [guid]::NewGuid().ToString()
        $watch.Restart()
        $packetId = Get-CmqPacketId $c
        Write-CmqPacket $c (New-CmqPacket 0x82 ([byte[]]@([byte]($packetId -shr 8), [byte]($packetId -band 0xFF)) + (ConvertTo-CmqString $topic) + [byte[]]@(0)))
        [void](Wait-CmqPacket $c 9 $packetId)
        Publish-MQTTMessage -Connection $c -Topic $topic -Message $token
        $deadline = [DateTime]::UtcNow.AddSeconds(10)
        while ([DateTime]::UtcNow -lt $deadline) {
            $packet = Read-CmqPacket $c 1000
            if ($packet -and $packet.Type -eq 3 -and (ConvertFrom-CmqPublish $c $packet).Payload -eq $token) {
                $result.RoundTrip = $true; $result.RoundTripMs = $watch.ElapsedMilliseconds; break
            }
        }
    }
    catch { $result.Error = $_.Exception.Message }
    finally { if ($c) { Disconnect-MQTTBroker $c } }
    $result
}

$common = @{ BrokerHost = $BrokerHost; Tls = $Tls }
if ($Username) { $common.Username = $Username; $common.Password = $Password }

switch ($Action) {
    'Info' { Get-MQTTBrokerInfo }
    'Test' { Test-MQTTBroker -BrokerHost $BrokerHost -Tls:$Tls }
    'Publish' {
        if (-not $Topic) { throw '-Topic is required for Publish.' }
        $c = Connect-MQTTBroker @common
        try { foreach ($t in $Topic) { Publish-MQTTMessage -Connection $c -Topic $t -Message $Message -Qos $Qos -Retain:$Retain } }
        finally { Disconnect-MQTTBroker $c }
        Write-Host "Published to $($Topic -join ', ')"
    }
    'Subscribe' {
        if (-not $Topic) { throw '-Topic is required for Subscribe.' }
        $c = Connect-MQTTBroker @common
        Write-Host "Subscribed to $($Topic -join ', ') on $($c.BrokerHost):$($c.Port). Press Ctrl+C to stop."
        try {
            Receive-MQTTMessage -Connection $c -Topic $Topic -Qos $Qos -Count $Count -TimeoutSeconds $TimeoutSeconds |
                ForEach-Object { '{0:HH:mm:ss} {1} {2}' -f $_.Received, $_.Topic, $_.Payload }
        }
        finally { Disconnect-MQTTBroker $c }
    }
}