mirror of
https://github.com/taigrr/nats.docs
synced 2025-01-18 04:03:23 -08:00
188 lines
6.9 KiB
HTML
188 lines
6.9 KiB
HTML
|
|
<div class="tab-wrap">
|
|
|
|
|
|
<input type="radio" id="drain_sub_go" name="drain_sub" class="tab" checked>
|
|
|
|
<label for="drain_sub_go" class="api-lang" data-language="go">Go</label>
|
|
|
|
|
|
<input type="radio" id="drain_sub_java" name="drain_sub" class="tab">
|
|
|
|
<label for="drain_sub_java" class="api-lang" data-language="java">Java</label>
|
|
|
|
|
|
<input type="radio" id="drain_sub_js" name="drain_sub" class="tab">
|
|
|
|
<label for="drain_sub_js" class="api-lang" data-language="js">JavaScript</label>
|
|
|
|
|
|
<input type="radio" id="drain_sub_py" name="drain_sub" class="tab">
|
|
|
|
<label for="drain_sub_py" class="api-lang" data-language="py">Python</label>
|
|
|
|
|
|
<input type="radio" id="drain_sub_ruby" name="drain_sub" class="tab">
|
|
|
|
<label for="drain_sub_ruby" class="api-lang" data-language="ruby">Ruby</label>
|
|
|
|
|
|
<input type="radio" id="drain_sub_ts" name="drain_sub" class="tab">
|
|
|
|
<label for="drain_sub_ts" class="api-lang" data-language="ts">TypeScript</label>
|
|
|
|
|
|
|
|
<div class="tab__content">
|
|
<pre id="drain_sub_go_content"><a class="toolbar-icons pull-right" target="_blank" href="https://github.com/nats-io/go-nats-examples/blob/master/api-examples/drain_sub/main.go#L13-70"><i class="mdi mdi-github-circle" title="View on GitHub"></i></a><a class="toolbar-icons pull-right"><i class="mdi mdi-content-copy js-copy" title="Copy to Clipboard"></i></a><span class="copy-msg pull-right"></span><code class="language-go">
|
|
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!
|
|
}
|
|
|
|
</code></pre>
|
|
</div>
|
|
|
|
<div class="tab__content">
|
|
<pre id="drain_sub_java_content"><a class="toolbar-icons pull-right" target="_blank" href="https://github.com/nats-io/java-nats-examples/blob/master/src/main/java/io/nats/examples/DrainSub.java#L16-43"><i class="mdi mdi-github-circle" title="View on GitHub"></i></a><a class="toolbar-icons pull-right"><i class="mdi mdi-content-copy js-copy" title="Copy to Clipboard"></i></a><span class="copy-msg pull-right"></span><code class="language-java">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();
|
|
</code></pre>
|
|
</div>
|
|
|
|
<div class="tab__content">
|
|
<pre id="drain_sub_js_content"><a class="toolbar-icons pull-right" target="_blank" href="https://github.com/nats-io/node-nats-examples/blob/master/src/subscriber_samples.js#L232-252"><i class="mdi mdi-github-circle" title="View on GitHub"></i></a><a class="toolbar-icons pull-right"><i class="mdi mdi-content-copy js-copy" title="Copy to Clipboard"></i></a><span class="copy-msg pull-right"></span><code class="language-javascript">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();
|
|
});
|
|
</code></pre>
|
|
</div>
|
|
|
|
<div class="tab__content">
|
|
<pre id="drain_sub_py_content"><a class="toolbar-icons pull-right" target="_blank" href="https://github.com/nats-io/asyncio-nats-examples/blob/master/drain_sub.py#L1-24"><i class="mdi mdi-github-circle" title="View on GitHub"></i></a><a class="toolbar-icons pull-right"><i class="mdi mdi-content-copy js-copy" title="Copy to Clipboard"></i></a><span class="copy-msg pull-right"></span><code class="language-python">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)
|
|
|
|
</code></pre>
|
|
</div>
|
|
|
|
<div class="tab__content">
|
|
<pre id="drain_sub_ruby_content"><a class="toolbar-icons pull-right" target="_blank" href="https://github.com/nats-io/ruby-nats-examples/blob/master/notapplicable.txt#L9-11"><i class="mdi mdi-github-circle" title="View on GitHub"></i></a><a class="toolbar-icons pull-right"><i class="mdi mdi-content-copy js-copy" title="Copy to Clipboard"></i></a><span class="copy-msg pull-right"></span><code class="language-ruby"># There is currently no API to drain a single subscription, the whole connection can be drained though via NATS.drain
|
|
</code></pre>
|
|
</div>
|
|
|
|
<div class="tab__content">
|
|
<pre id="drain_sub_ts_content"><a class="toolbar-icons pull-right" target="_blank" href="https://github.com/nats-io/ts-nats-examples/blob/master/src/subscriber_samples.ts#L208-212"><i class="mdi mdi-github-circle" title="View on GitHub"></i></a><a class="toolbar-icons pull-right"><i class="mdi mdi-content-copy js-copy" title="Copy to Clipboard"></i></a><span class="copy-msg pull-right"></span><code class="language-javascript">let sub = await nc.subscribe('updates', (err, msg) => {
|
|
t.log('worker got message', msg.data);
|
|
}, {queue: "workers"});
|
|
</code></pre>
|
|
</div>
|
|
|
|
</div>
|