MyNatsClient.Rx 0.12.0-pre1

Convenience package that installs MyNatsClient and ReactiveExtensions (System.Reactive). MyNatsClient, provides a simple, effective sync and async library for interacting with NATS Server using .NET and .NET Core. It uses IObservable so it is ReactiveX (RX) friendly.

This is a prerelease version of MyNatsClient.Rx.
There is a newer version of this package available.
See the version list below for details.
Install-Package MyNatsClient.Rx -Version 0.12.0-pre1
dotnet add package MyNatsClient.Rx --version 0.12.0-pre1
paket add MyNatsClient.Rx --version 0.12.0-pre1
The NuGet Team does not provide support for this client. Please contact its maintainers for support.

Pub-Sub sample

Simple pub-sub sample showing one client that publishes and one that subscribes. This can of course be the same client and you can also have more clients subscribing etc.

Publisher

var cnInfo = new ConnectionInfo("192.168.1.10");
var client = new NatsClient(cnInfo);

await client.PubAsync("tick", GetNextTick());

//or using an encoding package e.g. Json
await client.PubAsJsonAsync("tickItem", new Tick { Value = GetNextTick() });

Subscriber

var cnInfo = new ConnectionInfo("192.168.1.10");
var client = new NatsClient(cnInfo);

await client.SubAsync("tick", stream => stream.Subscribe(msg => {
    Console.WriteLine($"Clock ticked. Tick is {msg.GetPayloadAsString()}");
}));

//or using an encoding package e.g Json
await client.SubAsync("tickItem", stream => stream.Subscribe(msg => {
    Console.WriteLine($"Clock ticked. Tick is {msg.FromJson<TestItem>().Value}");
}))

Stream.Subscribe vs Stream.SubscribeSafe

If you subscribe to e.g. the MessageOpStream using Stream.Subscribe and your handler is throwing an exception. That handler will get OnError invoked and then removed.

await client.SubAsync("mySubject", stream => stream.Subscribe(msg => DoSomething(msg)));

If you instead subscribe using Stream.SubscribeSafe any unhandled exception will get swallowed.

await client.SubAsync("mySubject", stream => stream.SubscribeSafe(msg => DoSomething(msg)));

Request-Response sample

Simple request-response sample. This sample also makes use of two clients. It can of course be the same client requesting and responding, you can also have more responders forming a queue group. Where one will be giving the answer.

Requester

var cnInfo = new ConnectionInfo("192.168.1.10");
var client = new NatsClient(cnInfo);

var response = await client.RequestAsync("getTemp", "stockholm@sweden");
Console.WriteLine($"Temp in Stockholm is {response.GetPayloadAsString()}");

Responder

var cnInfo = new ConnectionInfo("192.168.1.10");
var client = new NatsClient(cnInfo);

await client.SubAsync("getTemp", stream.Subscribe(msg => {
    client.Pub(msg.ReplyTo, getTemp(msg.GetPayloadAsString()));
}));

Pub-Sub sample

Simple pub-sub sample showing one client that publishes and one that subscribes. This can of course be the same client and you can also have more clients subscribing etc.

Publisher

var cnInfo = new ConnectionInfo("192.168.1.10");
var client = new NatsClient(cnInfo);

await client.PubAsync("tick", GetNextTick());

//or using an encoding package e.g. Json
await client.PubAsJsonAsync("tickItem", new Tick { Value = GetNextTick() });

Subscriber

var cnInfo = new ConnectionInfo("192.168.1.10");
var client = new NatsClient(cnInfo);

await client.SubAsync("tick", stream => stream.Subscribe(msg => {
    Console.WriteLine($"Clock ticked. Tick is {msg.GetPayloadAsString()}");
}));

//or using an encoding package e.g Json
await client.SubAsync("tickItem", stream => stream.Subscribe(msg => {
    Console.WriteLine($"Clock ticked. Tick is {msg.FromJson<TestItem>().Value}");
}))

Stream.Subscribe vs Stream.SubscribeSafe

If you subscribe to e.g. the MessageOpStream using Stream.Subscribe and your handler is throwing an exception. That handler will get OnError invoked and then removed.

await client.SubAsync("mySubject", stream => stream.Subscribe(msg => DoSomething(msg)));

If you instead subscribe using Stream.SubscribeSafe any unhandled exception will get swallowed.

await client.SubAsync("mySubject", stream => stream.SubscribeSafe(msg => DoSomething(msg)));

Request-Response sample

Simple request-response sample. This sample also makes use of two clients. It can of course be the same client requesting and responding, you can also have more responders forming a queue group. Where one will be giving the answer.

Requester

var cnInfo = new ConnectionInfo("192.168.1.10");
var client = new NatsClient(cnInfo);

var response = await client.RequestAsync("getTemp", "stockholm@sweden");
Console.WriteLine($"Temp in Stockholm is {response.GetPayloadAsString()}");

Responder

var cnInfo = new ConnectionInfo("192.168.1.10");
var client = new NatsClient(cnInfo);

await client.SubAsync("getTemp", stream.Subscribe(msg => {
    client.Pub(msg.ReplyTo, getTemp(msg.GetPayloadAsString()));
}));

Release Notes

https://github.com/danielwertheim/mynatsclient/blob/master/ReleaseNotes.md

Version History

Version Downloads Last updated
0.13.0 50 6/29/2018
0.12.0 41 6/29/2018
0.12.0-pre1 (current) 129 5/4/2018
0.11.0 4,876 1/26/2017
0.10.0 21,258 1/10/2017
0.8.0 156 1/4/2017
0.7.0 223 12/23/2016
0.6.0 162 12/4/2016
0.5.1 178 10/18/2016
0.5.0 160 10/17/2016
0.4.0 156 10/16/2016
0.3.0 162 10/8/2016
0.3.0-b1 155 10/6/2016
0.2.1 243 6/14/2016