/// Run the mini-redis server. /// /// Accepts connections from the supplied listener. For each inbound connection, /// a task is spawned to handle that connection. The server runs until the /// `shutdown` future completes, at which point the server shuts down /// gracefully. /// /// `tokio::signal::ctrl_c()` can be used as the `shutdown` argument. This will /// listen for a SIGINT signal. pubasyncfnrun(listener: TcpListener, shutdown: implFuture) { info!( "mini-redis server started listen on: {}", listener.local_addr().unwrap() );
// When the provided `shutdown` future completes, we must send a shutdown // message to all active connections. We use a broadcast channel for this // purpose. The call below ignores the receiver of the broadcast pair, and when // a receiver is needed, the subscribe() method on the sender is used to create one. let (notify_shutdown, _) = broadcast::channel(1); let (shutdown_complete_tx, shutdown_complete_rx) = mpsc::channel(1);
// Initialize the listener state letmut server = Listener { listener, db_holder: DbDropGuard::new(), limit_connections: Arc::new(Semaphore::new(MAX_CONNECTIONS)), notify_shutdown, shutdown_complete_tx, shutdown_complete_rx, };
// Concurrently run the server and listen for the `shutdown` signal. The // server task runs until an error is encountered, so under normal // circumstances, this `select!` statement runs until the `shutdown` signal // is received. tokio::select! { res = server.run() => { // If an error is received here, accepting connections from the TCP // listener failed multiple times and the server is giving up. // // Errors encountered when handling individual connections do not // bubble up to this point. ifletErr(err) = res { error!("failed to accept: {:?}", err); } } _ = shutdown => { // The shutdown signal has been received debug!("server is about to shutdown"); } }
// Extract the `shutdown_complete` receiver and transmitter // explicitly drop `shutdown_transmitter`. This is important, as the // `.await` below would otherwise never complete. letListener { mut shutdown_complete_rx, shutdown_complete_tx, notify_shutdown, .. } = server;
// When `notify_shutdown` is dropped, all tasks which have `subscribe`d will // receive the shutdown signal and can exit drop(notify_shutdown); // Drop final `Sender` so the `Receiver` below can complete drop(shutdown_complete_tx);
debug!("server is shutting down");
// Wait for all active connections to finish processing. As the `Sender` // handle held by the listener has been dropped above, the only remaining // `Sender` instances are held by connection handler tasks. When those drop, // the `mpsc` channel will close and `recv()` will return `None`. let_ = shutdown_complete_rx.recv().await; }
/// Server listener state. Created in the `run` call. It includes a `run` method /// which performs the TCP listening and initialization of per-connection state. #[derive(Debug)] pub(crate) structListener { /// Shared database handle. /// /// Contains the key / value store as well as the broadcast channels for /// pub/sub. /// /// This holds a wrapper around an `Arc`. The internal `Db` can be /// retrieved and passed into the per connection state (`Handler`). pub(crate) db_holder: DbDropGuard,
/// TCP listener supplied by the `run` caller. pub(crate) listener: TcpListener,
/// Limit the max number of connections. /// /// A `Semaphore` is used to limit the max number of connections. Before /// attempting to accept a new connection, a permit is acquired from the /// semaphore. If none are available, the listener waits for one. /// /// When handlers complete processing a connection, the permit is returned /// to the semaphore. pub(crate) limit_connections: Arc<Semaphore>,
/// Broadcasts a shutdown signal to all active connections. /// /// The initial `shutdown` trigger is provided by the `run` caller. The /// server is responsible for gracefully shutting down active connections. /// When a connection task is spawned, it is passed a broadcast receiver /// handle. When a graceful shutdown is initiated, a `()` value is sent via /// the broadcast::Sender. Each active connection receives it, reaches a /// safe terminal state, and completes the task. pub(crate) notify_shutdown: broadcast::Sender<()>,
/// Used as part of the graceful shutdown process to wait for client /// connections to complete processing. /// /// Tokio channels are closed once all `Sender` handles go out of scope. /// When a channel is closed, the receiver receives `None`. This is /// leveraged to detect all connection handlers completing. When a /// connection handler is initialized, it is assigned a clone of /// `shutdown_complete_tx`. When the listener shuts down, it drops the /// sender held by this `shutdown_complete_tx` field. Once all handler tasks /// complete, all clones of the `Sender` are also dropped. This results in /// `shutdown_complete_rx.recv()` completing with `None`. At this point, it /// is safe to exit the server process. pub(crate) shutdown_complete_rx: mpsc::Receiver<()>, pub(crate) shutdown_complete_tx: mpsc::Sender<()>, }
implListener { /// Run the server /// /// Listen for inbound connections. For each inbound connection, spawn a /// task to process that connection. /// /// # Errors /// /// Returns `Err` if accepting returns an error. This can happen for a /// number reasons that resolve over time. For example, if the underlying /// operating system has reached an internal limit for max number of /// sockets, accept will fail. /// /// The process is not able to detect when a transient error resolves /// itself. One strategy for handling this is to implement a back off /// strategy, which is what we do here. pub(crate) asyncfnrun(&mutself) ->Result<(), MiniRedisConnectionError> {
loop { // Wait for a permit to become available // // `acquire_owned` returns a permit that is bound to the semaphore. // When the permit value is dropped, it is automatically returned // to the semaphore. // // `acquire_owned()` returns `Err` when the semaphore has been // closed. We don't ever close the semaphore, so `unwrap()` is safe. letpermit = self .limit_connections .clone() .acquire_owned() .await .unwrap();
// Accept a new socket. This will attempt to perform error handling. // The `accept` method internally attempts to recover errors, so an // error here is non-recoverable. letsocket = self.accept().await?;
// Create the necessary per-connection handler state. letmut handler = Handler { // Get a handle to the shared database. db: self.db_holder.db(),
// Initialize the connection state. This allocates read/write // buffers to perform redis protocol frame parsing. connection: Connection::new(socket),
// Notifies the receiver half once all clones are dropped. _shutdown_complete: self.shutdown_complete_tx.clone(), };
// Spawn a new task to process the connections. Tokio tasks are like // asynchronous green threads and are executed concurrently. tokio::spawn(asyncmove { // Process the connection. If an error is encountered, log it. ifletErr(err) = handler.run().await { error!("connection error：{:?}", err); } // Move the permit into the task and drop it after completion. // This returns the permit back to the semaphore. drop(permit); }); } }
/// Accept an inbound connection. /// /// Errors are handled by backing off and retrying. An exponential backoff /// strategy is used. After the first failure, the task waits for 1 second. /// After the second failure, the task waits for 2 seconds. Each subsequent /// failure doubles the wait time. If accepting fails on the 6th try after /// waiting for 64 seconds, then this function returns with an error. asyncfnaccept(&mutself) ->Result<TcpStream, MiniRedisConnectionError> { letmut backoff = 1;
// Try to accept a few times loop { // Perform the accept operation. If a socket is successfully // accepted, return it. Otherwise, save the error. matchself.listener.accept().await { Ok((socket, _)) => returnOk(socket), Err(err) => { if backoff > 64 { // Accept has failed too many times. Return the error. error!("failed to accept socket after retry: {}", err); returnErr(err.into()); } else { error!("failed to accept socket: {}", err); } } }
// Pause execution until the back off period elapses. time::sleep(Duration::from_secs(backoff)).await;
/// Per-connection handler. Reads requests from `connection` and applies the /// commands to `db`. #[derive(Debug)] pub(crate) structHandler { /// Shared database handle. /// /// When a command is received from `connection`, it is applied with `db`. /// The implementation of the command is in the `cmd` module. Each command /// will need to interact with `db` in order to complete the work. pub(crate) db: Db,
/// The TCP connection decorated with the redis protocol encoder / decoder /// implemented using a buffered `TcpStream`. /// /// When `Listener` receives an inbound connection, the `TcpStream` is /// passed to `Connection::new`, which initializes the associated buffers. /// `Connection` allows the handler to operate at the "frame" level and keep /// the byte level protocol parsing details encapsulated in `Connection`. pub(crate) connection: Connection,
/// Listen for shutdown notifications. /// /// A wrapper around the `broadcast::Receiver` paired with the sender in /// `Listener`. The connection handler processes requests from the /// connection until the peer disconnects **or** a shutdown notification is /// received from `shutdown`. In the latter case, any in-flight work being /// processed for the peer is continued until it reaches a safe state, at /// which point the connection is terminated. pub(crate) shutdown: Shutdown,
/// Not used directly. Instead, when `Handler` is dropped pub(crate) _shutdown_complete: mpsc::Sender<()>, }
其中：
db：数据库的计数引用；
connection：客户端的 TCP 连接；
shutdown：服务器 shutdown 后的信号通知；
_shutdown_complete：Handler 处理完成后 Drop 此引用，通知 Listener 中的 channel；
implHandler { /// Process a single connection. /// /// Request frames are read from the socket and processed. Responses are /// written back to the socket. /// /// Currently, pipelining is not implemented. Pipelining is the ability to /// process more than one request concurrently per connection without /// interleaving frames. /// /// See for more details: /// https://redis.io/topics/pipelining /// /// When the shutdown signal is received, the connection is processed until /// it reaches a safe state, at which point it is terminated. pub(crate) asyncfnrun(&mutself) ->Result<(), MiniRedisConnectionError> { // As long as the shutdown signal has not been received, try to read a // new request frame. while !self.shutdown.is_shutdown() { // While reading a request frame, also listen for the shutdown // signal. letmaybe_frame = tokio::select! { res = self.connection.read_frame() => res?, _ = self.shutdown.recv() => { // If a shutdown signal is received, return from `run`. // This will result in the task terminating. returnOk(()); } };
// If `None` is returned from `read_frame()` then the peer closed // the socket. There is no further work to do and the task can be // terminated. letframe = match maybe_frame { Some(frame) => frame, None => { debug!("peer closed the socket, return"); returnOk(()); } };
// Convert the redis frame into a command struct. This returns an // error if the frame is not a valid redis command or it is an // unsupported command. letcmd = Command::from_frame(frame)?;
// Logs the `cmd` object. debug!("received command: {:?}", cmd);
// Perform the work needed to apply the command. This may mutate the // database state as a result. // // The connection is passed into the apply function which allows the // command to write response frames directly to the connection. In // the case of pub/sub, multiple frames may be send back to the // peer. cmd.apply(&self.db, &mutself.connection, &mutself.shutdown) .await?; }
Ok(()) } }
只要在 run 方法中没有收到 shutdown 信号或没有报错，就调用 self.connection.read_frame() 持续接收并解析来自客户端的数据；
/// Enumeration of supported Redis commands. /// /// Methods called on `Command` are delegated to the command implementation. #[derive(Debug)] pubenumCommand { Get(Get), Set(Set), Publish(Publish), Subscribe(Subscribe), Unsubscribe(Unsubscribe), Ping(Ping), Unknown(Unknown), }
implCommand { /// Parse a command from a received frame. /// /// The `Frame` must represent a Redis command supported by `mini-redis` and /// be the array variant. pubfnfrom_frame(frame: Frame) ->Result<Command, MiniRedisParseError> { letmut parse = Parse::new(frame)?;
// All redis commands begin with the command name as a string. The name // is read and converted to lower cases in order to do case sensitive // matching. letcommand_name = parse.next_string()?.to_lowercase();
// Match the command name, delegating the rest of the parsing to the // specific command. letcommand = match &command_name[..] { "get" => Command::Get(Get::parse_frames(&mut parse)?), "set" => Command::Set(Set::parse_frames(&mut parse)?), "publish" => Command::Publish(Publish::parse_frames(&mut parse)?), "subscribe" => Command::Subscribe(Subscribe::parse_frames(&mut parse)?), "unsubscribe" => Command::Unsubscribe(Unsubscribe::parse_frames(&mut parse)?), "ping" => Command::Ping(Ping::parse_frames(&mut parse)?), _ => { // The command is not recognized and an Unknown command is // returned. // // `return` is called here to skip the `finish()` call below. As // the command is not recognized, there is most likely // unconsumed fields remaining in the `Parse` instance. returnOk(Command::Unknown(Unknown::new(command_name))); } };
// Check if there is any remaining unconsumed fields in the `Parse` // value. If fields remain, this indicates an unexpected frame format // and an error is returned. parse.finish()?;
// The command has been successfully parsed Ok(command) }
/// Apply the command to the specified `Db` instance. /// /// The response is written to `dst`. This is called by the server in order /// to execute a received command. /// Apply the command to the specified `Db` instance. /// /// The response is written to `dst`. This is called by the server in order /// to execute a received command. pub(crate) asyncfnapply( self, db: &Db, dst: &mut Connection, shutdown: &mut Shutdown, ) ->Result<(), MiniRedisConnectionError> { use Command::*;
matchself { Ping(cmd) => cmd.apply(dst).await, Get(cmd) => cmd.apply(db, dst).await, Set(cmd) => cmd.apply(db, dst).await, Publish(cmd) => cmd.apply(db, dst).await, Subscribe(cmd) => cmd.apply(db, dst, shutdown).await, // `Unsubscribe` cannot be applied. It may only be received from the // context of a `Subscribe` command. Unsubscribe(_) => Err(MiniRedisConnectionError::CommandExecute( "`Unsubscribe` is unsupported in this context".into(), )), Unknown(cmd) => cmd.apply(dst).await, } }
/// Parse a `Ping` instance from a received frame. /// /// The `Parse` argument provides a cursor-like API to read fields from the /// `Frame`. At this point, the entire frame has already been received from /// the socket. /// /// The `PING` string has already been consumed. pub(crate) fnparse_frames(parse: &mut Parse) ->Result<Ping, MiniRedisParseError> { match parse.next_string() { Ok(msg) => Ok(Ping::new(Some(msg))), Err(MiniRedisParseError::EndOfStream) => Ok(Ping::default()), Err(e) => Err(e), } }
/// Apply the `Ping` command and return the message. /// /// The response is written to `dst`. This is called by the server in order /// to execute a received command. pub(crate) asyncfnapply(self, dst: &mut Connection) ->Result<(), MiniRedisConnectionError> { letresponse = matchself.msg { None => Frame::Simple("PONG".to_string()), Some(msg) => Frame::Bulk(Bytes::from(msg)), };
// Write the response back to the client dst.write_frame(&response).await?;
Ok(()) }
/// Converts the command into an equivalent `Frame`. /// /// This is called by the client when encoding a `Ping` command to send /// to the server. pub(crate) fninto_frame(self) ->Result<Frame, MiniRedisParseError> { letmut frame = Frame::array(); frame.push_bulk(Bytes::from("ping".as_bytes()))?; ifletSome(msg) = self.msg { frame.push_bulk(Bytes::from(msg))?; } Ok(frame) } }
/// Set `key` to hold the string `value`. /// /// If `key` already holds a value, it is overwritten, regardless of its type. /// Any previous time to live associated with the key is discarded on successful /// SET operation. #[derive(Debug)] pubstructSet { /// the lookup key key: String,
/// the value to be stored value: Bytes,
/// When to expire the key expire: Option<Duration>, }
implSet { pubfnnew(key: implToString, value: Bytes, expire: Option<Duration>) -> Set { Set { key: key.to_string(), value, expire, } }
/// Parse a `Set` instance from a received frame. /// /// The `Parse` argument provides a cursor-like API to read fields from the /// `Frame`. At this point, the entire frame has already been received from /// the socket. pub(crate) fnparse_frames(parse: &mut Parse) ->Result<Set, MiniRedisParseError> { // Read the key to set. This is a required field letkey = parse.next_string()?;
// Read the value to set. This is a required field. letvalue = parse.next_bytes()?;
// The expiration is optional. If nothing else follows, then it is `None`. letmut expire = None;
// Attempt to parse another string. match parse.next_string() { Ok(s) if s.to_uppercase() == "EX" => { // An expiration is specified in seconds. The next value is an // integer. letsecs = parse.next_int()?; expire = Some(Duration::from_secs(secs)); } Ok(s) if s.to_uppercase() == "PX" => { // An expiration is specified in milliseconds. The next value is // an integer. letms = parse.next_int()?; expire = Some(Duration::from_millis(ms)); } // Currently, mini-redis does not support any of the other SET // options. An error here results in the connection being // terminated. Other connections will continue to operate normally. Ok(s) => { warn!("unsupported SET option: {}", s); returnErr(MiniRedisParseError::Parse( "currently `SET` only supports the expiration option".into(), )); } // The `EndOfStream` error indicates there is no further data to // parse. In this case, it is a normal run time situation and // indicates there are no specified `SET` options. Err(MiniRedisParseError::EndOfStream) => { debug!("no extra SET option"); } // All other errors are bubbled up, resulting in the connection // being terminated. Err(err) => returnErr(err), }
Ok(Set { key, value, expire }) }
/// Apply the `Set` command to the specified `Db` instance. /// /// The response is written to `dst`. This is called by the server in order /// to execute a received command. pub(crate) asyncfnapply( self, db: &Db, dst: &mut Connection, ) ->Result<(), MiniRedisConnectionError> { // Set the value in the shared database state. db.set(self.key, self.value, self.expire);
// Create a success response and write it to `dst`. letresponse = Frame::Simple("OK".to_string()); debug!("applied set command response: {:?}", response);
dst.write_frame(&response).await?;
Ok(()) }
/// Converts the command into an equivalent `Frame`. /// /// This is called by the client when encoding a `Set` command to send to /// the server. pub(crate) fninto_frame(self) ->Result<Frame, MiniRedisParseError> { letmut frame = Frame::array(); frame.push_bulk(Bytes::from("set".as_bytes()))?; frame.push_bulk(Bytes::from(self.key.into_bytes()))?; frame.push_bulk(self.value)?; ifletSome(ms) = self.expire { // Expirations in Redis procotol can be specified in two ways // 1. SET key value EX seconds // 2. SET key value PX milliseconds // We implement the second option because it allows greater precision and // src/bin/cli.rs parses the expiration argument as milliseconds // in duration_from_ms_str() frame.push_bulk(Bytes::from("px".as_bytes()))?; frame.push_int(ms.as_millis() asu64)?; } Ok(frame) }
parse_frames 方法首先通过 parse 获取到了要 set 的 key 和 value，随后获取下一个 String：
如果为 EX：则通过下一个 int 类型获取秒单位的过期时间；
如果为 PX：则通过下一个 int 类型获取毫秒单位的过期时间；
否则说明命令格式有误，返回对应的错误；
Set 命令的 apply 方法和 Get 命令类似，直接通过操作数据库接口即可保存，随后向客户端返回结果即可；
Subscribe命令
Subscribe 命令较为复杂，下面我们来看；
Subscribe 命令定义如下：
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15
/// Subscribes the client to one or more channels. /// /// Once the client enters the subscribed state, it is not supposed to issue any /// other commands, except for additional SUBSCRIBE, PSUBSCRIBE, UNSUBSCRIBE, /// PUNSUBSCRIBE, PING and QUIT commands. #[derive(Debug)] pubstructSubscribe { channels: Vec<String>, }
/// Stream of messages. The stream receives messages from the /// `broadcast::Receiver`. We use `stream!` to create a `Stream` that consumes /// messages. Because `stream!` values cannot be named, we box the stream using /// a trait object. typeMessages = Pin<Box<dyn Stream<Item = Bytes> + Send>>;
pub(crate) fnparse_frames(parse: &mut Parse) ->Result<Subscribe, MiniRedisParseError> { // Extract the first string. If there is none, the the frame is // malformed and the error is bubbled up. letmut channels = vec![parse.next_string()?];
loop { match parse.next_string() { // A string has been consumed from the `parse`, push it into the // list of channels to subscribe to. Ok(s) => channels.push(s), // The `EndOfStream` error indicates there is no further data to // parse. Err(MiniRedisParseError::EndOfStream) => break, // All other errors are bubbled up, resulting in the connection // being terminated. Err(err) => returnErr(err), } }
pub(crate) asyncfnapply(mutself, db: &Db, dst: &mut Connection, shutdown: &mut Shutdown, ) ->Result<(), MiniRedisConnectionError> { // Each individual channel subscription is handled using a // `sync::broadcast` channel. Messages are then fanned out to all // clients currently subscribed to the channels. // // An individual client may subscribe to multiple channels and may // dynamically add and remove channels from its subscription set. To // handle this, a `StreamMap` is used to track active subscriptions. The // `StreamMap` merges messages from individual broadcast channels as // they are received. letmut subscriptions = StreamMap::new();
loop { // `self.channels` is used to track additional channels to subscribe // to. When new `SUBSCRIBE` commands are received during the // execution of `apply`, the new channels are pushed onto this vec. forchannel_nameinself.channels.drain(..) { Self::subscribe_to_channel(channel_name, &mut subscriptions, db, dst).await?; }
// Wait for one of the following to happen: // // - Receive a message from one of the subscribed channels. // - Receive a subscribe or unsubscribe command from the client. // - A server shutdown signal. select! { // Receive messages from subscribed channels Some((channel_name, msg)) = subscriptions.next() => { dst.write_frame(&make_message_frame(channel_name, msg)?).await?; } res = dst.read_frame() => { letframe = match res? { Some(frame) => frame, // This happens if the remote client has disconnected. None => { warn!("remote subscribe client disconnected"); returnOk(()) } };
asyncfnhandle_command( frame: Frame, subscribe_to: &mutVec<String>, subscriptions: &mut StreamMap<String, Messages>, dst: &mut Connection, ) ->Result<(), MiniRedisConnectionError> { // A command has been received from the client. // // Only `SUBSCRIBE` and `UNSUBSCRIBE` commands are permitted in this context. match Command::from_frame(frame)? { Command::Subscribe(subscribe) => { // The `apply` method will subscribe to the channels we add to this // vector. subscribe_to.extend(subscribe.channels.into_iter()); } Command::Unsubscribe(mut unsubscribe) => { // If no channels are specified, this requests unsubscribing from // **all** channels. To implement this, the `unsubscribe.channels` // vec is populated with the list of channels currently subscribed // to. if unsubscribe.channels.is_empty() { unsubscribe.channels = subscriptions .keys() .map(|channel_name| channel_name.to_string()) .collect(); }
use crate::connection::frame::Frame; use crate::connection::parse::Parse; use crate::error::MiniRedisParseError;
/// Unsubscribes the client from one or more channels. /// /// When no channels are specified, the client is unsubscribed from all the /// previously subscribed channels. #[derive(Clone, Debug)] pubstructUnsubscribe { pub(crate) channels: Vec<String>, }
pub(crate) fnparse_frames(parse: &mut Parse) ->Result<Unsubscribe, MiniRedisParseError> { // There may be no channels listed, so start with an empty vec. letmut channels = vec![];
// Each entry in the frame must be a string or the frame is malformed. // Once all values in the frame have been consumed, the command is fully // parsed. loop { match parse.next_string() { // A string has been consumed from the `parse`, push it into the // list of channels to unsubscribe from. Ok(s) => channels.push(s), // The `EndOfStream` error indicates there is no further data to // parse. Err(MiniRedisParseError::EndOfStream) => break, // All other errors are bubbled up, resulting in the connection // being terminated. Err(err) => returnErr(err), } }
pub(crate) asyncfnapply( self, db: &Db, dst: &mut Connection, ) ->Result<(), MiniRedisConnectionError> { // The shared state contains the `tokio::sync::broadcast::Sender` for // all active channels. Calling `db.publish` dispatches the message into // the appropriate channel. // // The number of subscribers currently listening on the channel is // returned. This does not mean that `num_subscriber` channels will // receive the message. Subscribers may drop before receiving the // message. Given this, `num_subscribers` should only be used as a // "hint". letnum_subscribers = db.publish(&self.channel, self.message);
// The number of subscribers is returned as the response to the publish // request. letresponse = Frame::Integer(num_subscribers asu64); debug!("apply command applied response: {}", response);
/// Responds to the client, indicating the command is not recognized. /// /// This usually means the command is not yet implemented by `mini-redis`. pub(crate) asyncfnapply(self, dst: &mut Connection) ->Result<(), MiniRedisConnectionError> { letresponse = Frame::Error(format!("err unknown command '{}'", self.command_name));
/// Listens for the server shutdown signal. /// /// Shutdown is signalled using a `broadcast::Receiver`. Only a single value is /// ever sent. Once a value has been sent via the broadcast channel, the server /// should shutdown. /// /// The `Shutdown` struct listens for the signal and tracks that the signal has /// been received. Callers may query for whether the shutdown signal has been /// received or not. #[derive(Debug)] pub(crate) structShutdown { /// `true` if the shutdown signal has been received shutdown: bool,
/// The receive half of the channel used to listen for shutdown. notify: broadcast::Receiver<()>, }
implShutdown { /// Create a new `Shutdown` backed by the given `broadcast::Receiver`. pub(crate) fnnew(notify: broadcast::Receiver<()>) -> Shutdown { Shutdown { shutdown: false, notify, } }
/// Returns `true` if the shutdown signal has been received. pub(crate) fnis_shutdown(&self) ->bool { self.shutdown }
/// Receive the shutdown notice, waiting if necessary. pub(crate) asyncfnrecv(&mutself) { // If the shutdown signal has already been received, then return immediately. ifself.shutdown { return; }
// Cannot receive a "lag error" as only one value is ever sent. let_ = self.notify.recv().await;
// Remember that the signal has been received. self.shutdown = true; } }
