connectmq.com-mqtt.ps1
|
<#PSScriptInfo .VERSION 1.0.0 .GUID 4e8a1d27-93c5-4b6f-8f0e-2a7d5c19b3e0 .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 connectmq.com .PARAMETER Action Test (default), Publish, Subscribe or Info. .EXAMPLE connectmq.com-mqtt.ps1 Checks connectivity and message round-trip against broker.connectmq.com. .EXAMPLE connectmq.com-mqtt.ps1 -Action Subscribe -Topic 'connectmq/demo/7f3a91/#' -Tls .EXAMPLE connectmq.com-mqtt.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 = 'ConnectMQ.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-ConnectMQBroker { <# .SYNOPSIS Shows connection details and limits of the free ConnectMQ public MQTT broker. .EXAMPLE Get-ConnectMQBroker #> [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-ConnectMQTopicPrefix { <# .SYNOPSIS Generates a random, collision-free topic prefix for testing on the public broker. .EXAMPLE $prefix = New-ConnectMQTopicPrefix # connectmq/demo/7f3a91 #> [CmdletBinding()] [OutputType([string])] param([string]$Root = 'connectmq/demo') "$Root/$([guid]::NewGuid().ToString('N').Substring(0, 6))" } function Connect-ConnectMQ { <# .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-ConnectMQ -Tls .EXAMPLE $c = Connect-ConnectMQ -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 = 'ConnectMQ.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-ConnectMQ { <# .SYNOPSIS Sends DISCONNECT and closes the connection. .EXAMPLE Disconnect-ConnectMQ $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-ConnectMQMessage { <# .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-ConnectMQMessage -Topic connectmq/demo/7f3a91/temp -Message '{"value":21.5}' .EXAMPLE $c = Connect-ConnectMQ -Tls; 1..5 | % { Publish-ConnectMQMessage -Connection $c -Topic demo/x -Message $_ -Qos 1 }; Disconnect-ConnectMQ $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-ConnectMQ -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-ConnectMQ $Connection } } } function Receive-ConnectMQMessage { <# .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-ConnectMQMessage -Topic 'connectmq/demo/7f3a91/#' .EXAMPLE Receive-ConnectMQMessage -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-ConnectMQ -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-ConnectMQ $Connection } } } function Test-ConnectMQBroker { <# .SYNOPSIS Checks that the ConnectMQ public broker is reachable and round-trips a message. .EXAMPLE Test-ConnectMQBroker .EXAMPLE Test-ConnectMQBroker -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-ConnectMQ -BrokerHost $BrokerHost -Tls:$Tls $result.Connected = $true $result.ConnectMs = $watch.ElapsedMilliseconds $topic = "$(New-ConnectMQTopicPrefix)/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-ConnectMQMessage -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-ConnectMQ $c } } $result } $common = @{ BrokerHost = $BrokerHost; Tls = $Tls } if ($Username) { $common.Username = $Username; $common.Password = $Password } switch ($Action) { 'Info' { Get-ConnectMQBroker } 'Test' { Test-ConnectMQBroker -BrokerHost $BrokerHost -Tls:$Tls } 'Publish' { if (-not $Topic) { throw '-Topic is required for Publish.' } $c = Connect-ConnectMQ @common try { foreach ($t in $Topic) { Publish-ConnectMQMessage -Connection $c -Topic $t -Message $Message -Qos $Qos -Retain:$Retain } } finally { Disconnect-ConnectMQ $c } Write-Host "Published to $($Topic -join ', ')" } 'Subscribe' { if (-not $Topic) { throw '-Topic is required for Subscribe.' } $c = Connect-ConnectMQ @common Write-Host "Subscribed to $($Topic -join ', ') on $($c.BrokerHost):$($c.Port). Press Ctrl+C to stop." try { Receive-ConnectMQMessage -Connection $c -Topic $Topic -Qos $Qos -Count $Count -TimeoutSeconds $TimeoutSeconds | ForEach-Object { '{0:HH:mm:ss} {1} {2}' -f $_.Received, $_.Topic, $_.Payload } } finally { Disconnect-ConnectMQ $c } } } |