RxMQTTnet

Rxmqttnet

Perpanjangan ke proyek MQTTNET, untuk mengubah langganan menjadi yang dapat diamati dan untuk menerbitkan dari aliran yang dapat diamati.

Buat klien

Gunakan pabrik

Gunakan MQTTnet.MqttFactory dengan MQTTnet.Extensions.External.RxMQTT.Client.MqttFactoryExtensions .

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

Buat opsi

Gunakan opsi klien yang dikelola

Mulai klien

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

Berlangganan

Dapatkan IObservable<MqttApplicationMessageReceivedEventArgs> dengan menghubungkan ke klien RX dan menggunakan ekstensi untuk memproses pesan:

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

Akhiri langganan dengan membuang langganan.

 subscription . Dispose ( ) ;

Menerbitkan

Dari yang bisa diamati

Buat urutan yang dapat diamati dari MqttApplicationMessage s dan publikasikan ini melalui klien 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 ( ) ; 

Pesan tunggal

Gunakan metode publikasi klien MQTT.