试着做一下思考题 use anyhow::Result; use async_prost::AsyncProstStream; use futures::prelude::*; use kv1::{CommandRequest, CommandResponse, Service, ServiceInner, SledDb}; use tokio::net::TcpListener; use tracing::info; use bytes::Bytes; use futures::{SinkExt, StreamExt}; use tokio_util::codec::{Framed, LengthDelimitedCodec}; use prost::Message; #[tokio::main] async fn main() -> Result<()> { tracing_subscriber::fmt::init(); let service: Service<SledDb> = ServiceInner::new(SledDb::new("/tmp/kvserver")) .fn_before_send(|res| match res.message.as_ref() { "" => res.message = "altered. Original message is empty.".into(), s => res.message = format!("altered: {}", s), }) .into(); let addr = "127.0.0.1:9527"; let listener = TcpListener::bind(addr).await?; loop { let (stream, addr) = listener.accept().await?; println!("accepted: {:?}", addr); // LengthDelimitedCodec 默认 4 字节长度 let mut stream = Framed::new(stream, LengthDelimitedCodec::new()); let svc = service.clone(); tokio::spawn(async move { // 接收到的消息会只包含消息主体（不包含长度） while let Some(Ok(data)) = stream.next().await { let cmd = CommandRequest::decode(data).unwrap(); info!("req: {:?} ", cmd); let res = svc.execute(cmd); info!("res: {:?} ", res); stream.send(Bytes::from(res.encode_to_vec())).await.unwrap(); } }); } }

思考题可以参考 dva 同学的实现，我这里也提供一个关于 client 的修改: ```rust use anyhow::Result; use bytes::Bytes; use futures::prelude::*; use kv::CommandRequest; use prost::Message; use tokio::net::TcpStream; use tokio_util::codec::{Framed, LengthDelimitedCodec}; use tracing::info; #[tokio::main] async fn main() -> Result<()> { tracing_subscriber::fmt().init(); let addr = "127.0.0.1:9527"; let stream = TcpStream::connect(addr).await?; let mut client = Framed::new(stream, LengthDelimitedCodec::new()); let cmd = CommandRequest::new_hset("table1", "hello", "world".into()); client.send(Bytes::from(cmd.encode_to_vec())).await?; if let Some(Ok(data)) = client.next().await { info!("Got response: {:?}", data); } Ok(()) } ```