Implement the "drain" feature for subscriptions and connections #1332
Conversation
jsudano
force-pushed
the
implement_drain
branch
from
fffcb1d
to
8ebaedb
Compare
async-nats/src/lib.rs
Outdated
| @@ -773,6 +793,27 @@ impl ConnectionHandler { | |||
| Command::Flush { observer } => { | |||
| self.flush_observers.push(observer); | |||
| } | |||
| Command::Drain { sid } => { | |||
| let mut drain_sub = |sid: &u64, sub: &mut Subscription| { | |||
There was a problem hiding this comment.
I would pass by value here:
| let mut drain_sub = |sid: &u64, sub: &mut Subscription| { | |
| let mut drain_sub = |sid: u64, sub: &mut Subscription| { |
async-nats/src/lib.rs
Outdated
| let mut drain_sub = |sid: &u64, sub: &mut Subscription| { | ||
| sub.is_draining = true; | ||
| self.connection.enqueue_write_op(&ClientOp::Unsubscribe { | ||
| sid: *sid, |
There was a problem hiding this comment.
which allows to remove the dereference here
| sid: *sid, | |
| sid: sid, |
async-nats/src/lib.rs
Outdated
|
|
||
| if let Some(sid) = sid { | ||
| if let Some(sub) = self.subscriptions.get_mut(&sid) { | ||
| drain_sub(&sid, sub); |
async-nats/src/lib.rs
Outdated
| } else { | ||
| // sid isn't set, so drain the whole client | ||
| self.is_draining = true; | ||
| for (sid, sub) in self.subscriptions.iter_mut() { |
There was a problem hiding this comment.
| for (sid, sub) in self.subscriptions.iter_mut() { | |
| for (&sid, sub) in self.subscriptions.iter_mut() { |
| // The entire connection is draining. This means we flushed outgoing messages in the previous | ||
| // call to this fn, we handled any remaining messages from the server in the loop above, and | ||
| // all subs were drained, so drain is complete and we should exit instead of processing any | ||
| // further messages |
There was a problem hiding this comment.
I wonder if we should add new Event variants - like Draining and Closed.
async-nats/src/lib.rs
Outdated
| @@ -1251,6 +1293,48 @@ impl Subscriber { | |||
| .await?; | |||
| Ok(()) | |||
| } | |||
|
|
|||
| /// Unsubscribes immediately but leaves the stream open to allow any in-flight messages on the | |||
There was a problem hiding this comment.
Maybe subscription instead of stream? Sounds more direct on what is affected.
async-nats/src/lib.rs
Outdated
| /// delivered | ||
| /// | ||
| /// # Examples | ||
| /// ``` |
There was a problem hiding this comment.
Please add no_run to examples - the demo server is not reliable enough :).
|
(CI was red because there was a typo in example. Should be fine after a rebase) |
|
@jsudano ping, if you missed the review 🙂 |
|
Apologies for the delay, was away from the computer last week. Really appreciate the review, will get the comments addressed this week! |
jsudano
force-pushed
the
implement_drain
branch
from
8ebaedb
to
aa13c20
Compare
| } | ||
| } else { | ||
| // sid isn't set, so drain the whole client | ||
| self.connector.events_tx.try_send(Event::Draining).ok(); |
There was a problem hiding this comment.
Assumed it was fine to ignore any errors here as they'd be pretty unlikely (the only scenario I could think of would be if the drain command was sent mid-shutdown, in which case it's fine to just ignore the error).
| ExitReason::Closed => break, | ||
| ExitReason::Closed => { | ||
| // | ||
| self.connector.events_tx.try_send(Event::Closed).ok(); |
There was a problem hiding this comment.
Similar to the other comment, this is more-or-less the last thing we'd do during a drain-shutdown so it's probably fine to ignore errors.
There was a problem hiding this comment.
It is, however please either remove the // or add a comment there that explains why its ok.
| ExitReason::Closed => break, | ||
| ExitReason::Closed => { | ||
| // | ||
| self.connector.events_tx.try_send(Event::Closed).ok(); |
There was a problem hiding this comment.
It is, however please either remove the // or add a comment there that explains why its ok.
…orgot to set handler.is_draining
…d` by value in closure; update doc comments
jsudano
force-pushed
the
implement_drain
branch
from
1fb87ab
to
77a3381
Compare
|
Oops, hopefully that last commit got all of them (I ran the commands from the validation runs locally and they passed at least) |
|
@jsudano Thanks! The test is missing |
Based on the proposal in #1325, implementing the functionality described in this NATS doc.
I would love suggestions for further tests. I tried adding tests for more complex timing/corner cases but they all wound up looking like exactly like the tests I have here