RxMQTTnet

An extension to the MQTTnet project, to transform the subscriptions into observables and to publish from an observable stream.

Create a client

Use the factory

Use the MQTTnet.MqttFactory with the MQTTnet.Extensions.External.RxMQTT.Client.MqttFactoryExtensions.

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

Create the options

Use the managed client options

Start the client

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

Get an IObservable<MqttApplicationMessageReceivedEventArgs> by connecting to the rx client and use extensions to process the message:

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

End the subscription by disposing the subscription.

subscription.Dispose();

Publish

From observable

Create an observable sequence of MqttApplicationMessages and publish these via the rx client.

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

Single message

Use the mqtt client publish method.

fengdaokanhai/RxMQTTnet