krpc 0.2.0

/// 用于接收RPC请求消息
async fn recv<Stream>(
	mut stream: Stream, sub_tx: mpsc::Sender<Content>, call_tx: Option<mpsc::Sender<Pair>>, tx: mpsc::Sender<Bytes>, id: u32,
) -> Result<(), Error>
where
	Stream: AsyncReadExt + std::marker::Unpin,
{
	let mut header = [0u8; RPC_HEADER_LEN];
	loop {
		let n = stream.read_exact(&mut header[..]).await?;
		if n == 0 {
			return Ok(()); //说明对端已关闭
		}
		let mut msg = Msg::decode(&header[..])?;
		if let Some(buf) = msg.body() {
			let _ = stream.read_exact(buf).await?; //读取消息体
		}
		match msg.mode() {
			Mode::Request => {
				//发送请求的msg到caller
				if let Some(_tx) = &call_tx {
					let _ = _tx.send(Pair { msg, tx: tx.clone() }).await?;
				} else {
					//说明根本没有bind闭包回调
					let _ = tx.send(msg.encode_without_body(Mode::NotFound)).await?;
				}
			}
			Mode::Subcribe => {
				let _ = sub_tx.send(Content::Sub(id, Pair { msg, tx: tx.clone() })).await?;
			}
			Mode::HeartBeat => {
				//心跳消息原路返回
				let _ = tx.send(Vec::from(&header[..])).await?;
			}
			_ => return Err(Error::new("消息模式不匹配")),
		}
	}
}

/// 用于发送RPC回复消息
async fn send<Stream>(mut stream: Stream, mut rx: mpsc::Receiver<Bytes>) -> Result<(), Error>
where
	Stream: AsyncWriteExt + std::marker::Unpin,
{
	while let Some(buf) = rx.recv().await {
		if buf.len() >= RPC_HEADER_LEN {
			stream.write_all(&buf[..]).await?;
		}
	}
	Ok(())
}