MQTTBroker.psm1
|
#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 } Export-ModuleMember -Function Get-MQTTBrokerInfo, New-MQTTTopicPrefix, Connect-MQTTBroker, Disconnect-MQTTBroker, Publish-MQTTMessage, Receive-MQTTMessage, Test-MQTTBroker |