nc, err := nats.Connect("demo.nats.io")
if err != nil {
log.Fatal(err)
}
defer nc.Close()
done := sync.WaitGroup{}
done.Add(1)
count := 0
errCh := make(chan error, 1)
msgAfterDrain := "not this one"
// Just to not collide using the demo server with other users.
subject := nats.NewInbox()
// This callback will process each message slowly
sub, err := nc.Subscribe(subject, func(m *nats.Msg) {
if string(m.Data) == msgAfterDrain {
errCh <- fmt.Errorf("Should not have received this message")
return
}
time.Sleep(100 * time.Millisecond)
count++
if count == 2 {
done.Done()
}
})
// Send 2 messages
for i := 0; i < 2; i++ {
nc.Publish(subject, []byte("hello"))
}
// Call Drain on the subscription. It unsubscribes but
// wait for all pending messages to be processed.
if err := sub.Drain(); err != nil {
log.Fatal(err)
}
// Send one more message, this message should not be received
nc.Publish(subject, []byte(msgAfterDrain))
// Wait for the subscription to have processed the 2 messages.
done.Wait()
// Now check that the 3rd message was not received
select {
case e := <-errCh:
log.Fatal(e)
case <-time.After(200 * time.Millisecond):
// OK!
}
Connection nc = Nats.connect("nats://demo.nats.io:4222");
// Use a latch to wait for a message to arrive
CountDownLatch latch = new CountDownLatch(1);
// Create a dispatcher and inline message handler
Dispatcher d = nc.createDispatcher((msg) -> {
String str = new String(msg.getData(), StandardCharsets.UTF_8);
System.out.println(str);
latch.countDown();
});
// Subscribe
d.subscribe("updates");
// Wait for a message to come in
latch.await();
// Messages that have arrived will be processed
CompletableFuture<Boolean> drained = d.drain(Duration.ofSeconds(10));
// Wait for the drain to complete
drained.get();
// Close the connection
nc.close();
let nc = NATS.connect({url: "nats://demo.nats.io:4222"});
let inbox = createInbox();
let counter = 0;
let sid = nc.subscribe(inbox, () => {
counter++;
});
nc.publish(inbox);
nc.drainSubscription(sid, (err)=> {
if(err) {
t.log(err);
}
t.log('processed', counter, 'messages');
});
nc.flush(() => {
nc.close();
t.pass();
resolve();
});
import asyncio
from nats.aio.client import Client as NATS
async def example(loop):
nc = NATS()
await nc.connect("nats://127.0.0.1:4222", loop=loop)
async def handler(msg):
print("[Received] ", msg)
await nc.publish(msg.reply, b'I can help')
# Can check whether client is in draining state
if nc.is_draining:
print("Connection is draining")
sid = await nc.subscribe("help", "workers", cb=handler)
await nc.flush()
# Gracefully unsubscribe the subscription
await nc.drain(sid)
# There is currently no API to drain a single subscription, the whole connection can be drained though via NATS.drain
let sub = await nc.subscribe('updates', (err, msg) => {
t.log('worker got message', msg.data);
}, {queue: "workers"});