RxMQTTnet

RXMQTTNET

MQTTNET项目的扩展，将订阅转换为可观察到的，并从可观察的流发布。

创建客户

使用工厂

将MQTTnet.MqttFactory与MQTTnet.Extensions.External.RxMQTT.Client.MqttFactoryExtensions一起使用。

 var client = new MqttFactory ( ) . CreateRxMqttClient ( ) ; 

创建选项

使用托管客户选项

启动客户

 await client . StartAsync ( options ) . ConfigureAwait ( false ) ;

订阅

通过连接到RX客户端并使用扩展IObservable<MqttApplicationMessageReceivedEventArgs>处理消息：

 var subscription = rxMqttClinet
    . Connect ( "RxClientTest/#" )
    . SelectPayload ( )
    . Subscribe ( Console . WriteLine ) ;

通过处置订阅来结束订阅。

 subscription . Dispose ( ) ;

发布

来自可观察的

创建一个可观察到的MqttApplicationMessage s序列，并通过RX客户端发布它们。

 Observable . Interval ( TimeSpan . FromMilliseconds ( 1000 ) )
    . Select ( i => new MqttApplicationMessageBuilder ( )
        . WithTopic ( "RxClientTest" )
        . WithPayload ( "Time: " + DateTime . Now . ToLongTimeString ( ) )
        . WithQualityOfServiceLevel ( MqttQualityOfServiceLevel . ExactlyOnce )
        . WithRetainFlag ( )
        . Build ( ) )
    . PublishOn ( mqttClient )
    . Subscribe ( ) ; 

单个消息

使用MQTT客户端发布方法。