v3: Subscriptions

[adam-fowler]

Subscription API should be of the form

extension MQTTNewConnection {
    func subscribe<Value>(
        to topics: [String], 
        operation: (MQTTSubscription) async throws -> sending Value
    )
}

This should send the relevant subscribe packet create a MQTTSubscription which is essentially a wrapper for an AsyncStream. Internally setup state so that when a PUBLISH packet is received the relevant AsyncStreams are fed the packet. When the operation exits then the relevant UNSUBSCRIBE commands are sent.

MQTT v3.1.1 subscriptions have a slight problem in that there is no way to associate a specific PUBLISH message with the equivalent SUBSCRIBE so you'll have to keep state recording how many SUBSCRIBE's have been called. And do topic name comparisons to make sure you are sending the correct PUBLISH message to the correct place.

MQTT v5 brings in the concept of subscription ids which allow us to tag a subscription with an id and the a PUBLISH packet we receive will then include the id's it is relevant for.