diff_service

diff --git a/39/kv/examples/server.rs b/39/kv/examples/server.rs
index d8f3dc2..25a13c4 100644
--- a/39/kv/examples/server.rs
+++ b/39/kv/examples/server.rs
@@ -21,8 +21,10 @@ async fn main() -> Result<()> {
                 AsyncProstStream::<_, CommandRequest, CommandResponse, _>::from(stream).for_async();
             while let Some(Ok(cmd)) = stream.next().await {
                 info!("Got a new command: {:?}", cmd);
-                let res = svc.execute(cmd);
-                stream.send(res).await.unwrap();
+                let mut res = svc.execute(cmd);
+                while let Some(data) = res.next().await {
+                    stream.send((*data).clone()).await.unwrap();
+                }
             }
             info!("Client {:?} disconnected", addr);
         });
diff --git a/39/kv/examples/server_with_codec.rs b/39/kv/examples/server_with_codec.rs
index ae6fb8c..c454732 100644
--- a/39/kv/examples/server_with_codec.rs
+++ b/39/kv/examples/server_with_codec.rs
@@ -1,4 +1,5 @@
 use anyhow::Result;
+use bytes::BytesMut;
 use futures::prelude::*;
 use kv5::{CommandRequest, MemTable, Service, ServiceInner};
 use prost::Message;
@@ -22,10 +23,14 @@ async fn main() -> Result<()> {
             while let Some(Ok(mut buf)) = stream.next().await {
                 let cmd = CommandRequest::decode(&buf[..]).unwrap();
                 info!("Got a new command: {:?}", cmd);
-                let res = svc.execute(cmd);
+                let mut res = svc.execute(cmd);
+
                 buf.clear();
-                res.encode(&mut buf).unwrap();
-                stream.send(buf.freeze()).await.unwrap();
+                while let Some(data) = res.next().await {
+                    let mut buf = BytesMut::new();
+                    data.encode(&mut buf).unwrap();
+                    stream.send(buf.freeze()).await.unwrap();
+                }
             }
             info!("Client {:?} disconnected", addr);
         });
diff --git a/39/kv/examples/server_with_sled.rs b/39/kv/examples/server_with_sled.rs
index 9d626c7..5e78c15 100644
--- a/39/kv/examples/server_with_sled.rs
+++ b/39/kv/examples/server_with_sled.rs
@@ -26,8 +26,10 @@ async fn main() -> Result<()> {
                 AsyncProstStream::<_, CommandRequest, CommandResponse, _>::from(stream).for_async();
             while let Some(Ok(cmd)) = stream.next().await {
                 info!("Got a new command: {:?}", cmd);
-                let res = svc.execute(cmd);
-                stream.send(res).await.unwrap();
+                let mut res = svc.execute(cmd);
+                while let Some(data) = res.next().await {
+                    stream.send((*data).clone()).await.unwrap();
+                }
             }
             info!("Client {:?} disconnected", addr);
         });
diff --git a/39/kv/src/network/mod.rs b/39/kv/src/network/mod.rs
index c7ddb1b..554e9b5 100644
--- a/39/kv/src/network/mod.rs
+++ b/39/kv/src/network/mod.rs
@@ -40,8 +40,10 @@ where
         let stream = &mut self.inner;
         while let Some(Ok(cmd)) = stream.next().await {
             info!("Got a new command: {:?}", cmd);
-            let res = self.service.execute(cmd);
-            stream.send(res).await.unwrap();
+            let mut res = self.service.execute(cmd);
+            while let Some(data) = res.next().await {
+                stream.send(&data).await.unwrap();
+            }
         }
         // info!("Client {:?} disconnected", self.addr);
         Ok(())
@@ -60,7 +62,7 @@ where

     pub async fn execute(&mut self, cmd: CommandRequest) -> Result<CommandResponse, KvError> {
         let stream = &mut self.inner;
-        stream.send(cmd).await?;
+        stream.send(&cmd).await?;

         match stream.next().await {
             Some(v) => v,
diff --git a/39/kv/src/network/stream.rs b/39/kv/src/network/stream.rs
index 6c241d5..dc521f1 100644
--- a/39/kv/src/network/stream.rs
+++ b/39/kv/src/network/stream.rs
@@ -54,7 +54,7 @@ where
 }

 /// 当调用 send() 时，会把 Out 发出去
-impl<S, In, Out> Sink<Out> for ProstStream<S, In, Out>
+impl<S, In, Out> Sink<&Out> for ProstStream<S, In, Out>
 where
     S: AsyncRead + AsyncWrite + Unpin,
     In: Unpin + Send,
@@ -67,7 +67,7 @@ where
         Poll::Ready(Ok(()))
     }

-    fn start_send(self: Pin<&mut Self>, item: Out) -> Result<(), Self::Error> {
+    fn start_send(self: Pin<&mut Self>, item: &Out) -> Result<(), Self::Error> {
         let this = self.get_mut();
         item.encode_frame(&mut this.wbuf)?;

@@ -140,7 +140,7 @@ mod tests {
         let cmd = CommandRequest::new_hdel("t1", "k1");

         // 使用 ProstStream 发送数据
-        stream.send(cmd.clone()).await?;
+        stream.send(&cmd).await?;

         // 使用 ProstStream 接收数据
         if let Some(Ok(s)) = stream.next().await {
diff --git a/39/kv/src/pb/mod.rs b/39/kv/src/pb/mod.rs
index 7534974..bf784aa 100644
--- a/39/kv/src/pb/mod.rs
+++ b/39/kv/src/pb/mod.rs
@@ -90,6 +90,30 @@ impl CommandRequest {
         }
     }

+    pub fn new_subscribe(name: impl Into<String>) -> Self {
+        Self {
+            request_data: Some(RequestData::Subscribe(Subscribe { topic: name.into() })),
+        }
+    }
+
+    pub fn new_unsubscribe(name: impl Into<String>, id: u32) -> Self {
+        Self {
+            request_data: Some(RequestData::Unsubscribe(Unsubscribe {
+                topic: name.into(),
+                id,
+            })),
+        }
+    }
+
+    pub fn new_publish(name: impl Into<String>, data: Vec<Value>) -> Self {
+        Self {
+            request_data: Some(RequestData::Publish(Publish {
+                topic: name.into(),
+                data,
+            })),
+        }
+    }
+
     /// 转换成 string 做错误处理
     pub fn format(&self) -> String {
         format!("{:?}", self)
@@ -97,6 +121,19 @@ impl CommandRequest {
 }

 impl CommandResponse {
+    pub fn ok() -> Self {
+        let mut result = CommandResponse::default();
+        result.status = StatusCode::OK.as_u16() as _;
+        result
+    }
+
+    pub fn internal_error(msg: String) -> Self {
+        let mut result = CommandResponse::default();
+        result.status = StatusCode::INTERNAL_SERVER_ERROR.as_u16() as _;
+        result.message = msg;
+        result
+    }
+
     /// 转换成 string 做错误处理
     pub fn format(&self) -> String {
         format!("{:?}", self)
diff --git a/39/kv/src/service/mod.rs b/39/kv/src/service/mod.rs
index 8f524b4..e39d9e9 100644
--- a/39/kv/src/service/mod.rs
+++ b/39/kv/src/service/mod.rs
@@ -1,13 +1,16 @@
 use crate::{
     command_request::RequestData, CommandRequest, CommandResponse, KvError, MemTable, Storage,
 };
+use futures::stream;
 use std::sync::Arc;
 use tracing::debug;

 mod command_service;
 mod topic;
+mod topic_service;

 pub use topic::{Broadcaster, Topic};
+pub use topic_service::{StreamingResponse, TopicService};

 /// 对 Command 的处理的抽象
 pub trait CommandService {
@@ -46,12 +49,14 @@ impl<Arg> NotifyMut<Arg> for Vec<fn(&mut Arg)> {
 /// Service 数据结构
 pub struct Service<Store = MemTable> {
     inner: Arc<ServiceInner<Store>>,
+    broadcaster: Arc<Broadcaster>,
 }

 impl<Store> Clone for Service<Store> {
     fn clone(&self) -> Self {
         Self {
             inner: Arc::clone(&self.inner),
+            broadcaster: Arc::clone(&self.broadcaster),
         }
     }
 }
@@ -101,27 +106,33 @@ impl<Store: Storage> From<ServiceInner<Store>> for Service<Store> {
     fn from(inner: ServiceInner<Store>) -> Self {
         Self {
             inner: Arc::new(inner),
+            broadcaster: Default::default(),
         }
     }
 }

 impl<Store: Storage> Service<Store> {
-    pub fn execute(&self, cmd: CommandRequest) -> CommandResponse {
+    pub fn execute(&self, cmd: CommandRequest) -> StreamingResponse {
         debug!("Got request: {:?}", cmd);
         self.inner.on_received.notify(&cmd);
-        let mut res = dispatch(cmd, &self.inner.store);
-        debug!("Executed response: {:?}", res);
-        self.inner.on_executed.notify(&res);
-        self.inner.on_before_send.notify(&mut res);
-        if !self.inner.on_before_send.is_empty() {
-            debug!("Modified response: {:?}", res);
+        let mut res = dispatch(cmd.clone(), &self.inner.store);
+
+        if res == CommandResponse::default() {
+            dispatch_stream(cmd, Arc::clone(&self.broadcaster))
+        } else {
+            debug!("Executed response: {:?}", res);
+            self.inner.on_executed.notify(&res);
+            self.inner.on_before_send.notify(&mut res);
+            if !self.inner.on_before_send.is_empty() {
+                debug!("Modified response: {:?}", res);
+            }
+
+            Box::pin(stream::once(async { Arc::new(res) }))
         }
-
-        res
     }
 }

-// 从 Request 中得到 Response，目前处理 HGET/HGETALL/HSET
+/// 从 Request 中得到 Response，目前处理所有 HGET/HSET/HDEL/HEXIST
 pub fn dispatch(cmd: CommandRequest, store: &impl Storage) -> CommandResponse {
     match cmd.request_data {
         Some(RequestData::Hget(param)) => param.execute(store),
@@ -134,31 +145,33 @@ pub fn dispatch(cmd: CommandRequest, store: &impl Storage) -> CommandResponse {
         Some(RequestData::Hexist(param)) => param.execute(store),
         Some(RequestData::Hmexist(param)) => param.execute(store),
         None => KvError::InvalidCommand("Request has no data".into()).into(),
-        _ => todo!(),
+        // 处理不了的返回一个啥都不包括的 Response，这样后续可以用 dispatch_stream 处理
+        _ => CommandResponse::default(),
     }
 }

-// pub fn dispatch_stream<T: Channel + Send>(cmd: CommandRequest, chan: T) -> StreamingResponse {
-//     match cmd.request_data {
-//         Some(RequestData::Publish(param)) => param.execute(chan),
-//         Some(RequestData::Subscribe(param)) => param.execute(chan),
-//         Some(RequestData::Unsubscribe(param)) => param.execute(chan),
-//         _ => unreachable!(),
-//     }
-// }
+/// 从 Request 中得到 Response，目前处理所有 PUBLISH/SUBSCRIBE/UNSUBSCRIBE
+pub fn dispatch_stream(cmd: CommandRequest, topic: impl Topic) -> StreamingResponse {
+    match cmd.request_data {
+        Some(RequestData::Publish(param)) => param.execute(topic),
+        Some(RequestData::Subscribe(param)) => param.execute(topic),
+        Some(RequestData::Unsubscribe(param)) => param.execute(topic),
+        // 如果走到这里，就是代码逻辑的问题，直接 crash 出来
+        _ => unreachable!(),
+    }
+}

 #[cfg(test)]
 mod tests {
-    use std::thread;
-
     use http::StatusCode;
+    use tokio_stream::StreamExt;
     use tracing::info;

     use super::*;
     use crate::{MemTable, Value};

-    #[test]
-    fn service_should_works() {
+    #[tokio::test]
+    async fn service_should_works() {
         // 我们需要一个 service 结构至少包含 Storage
         let service: Service = ServiceInner::new(MemTable::default()).into();

@@ -166,19 +179,22 @@ mod tests {
         let cloned = service.clone();

         // 创建一个线程，在 table t1 中写入 k1, v1
-        let handle = thread::spawn(move || {
-            let res = cloned.execute(CommandRequest::new_hset("t1", "k1", "v1".into()));
-            assert_res_ok(&res, &[Value::default()], &[]);
-        });
-        handle.join().unwrap();
+        tokio::spawn(async move {
+            let mut res = cloned.execute(CommandRequest::new_hset("t1", "k1", "v1".into()));
+            let data = res.next().await.unwrap();
+            assert_res_ok(&data, &[Value::default()], &[]);
+        })
+        .await
+        .unwrap();

         // 在当前线程下读取 table t1 的 k1，应该返回 v1
-        let res = service.execute(CommandRequest::new_hget("t1", "k1"));
-        assert_res_ok(&res, &["v1".into()], &[]);
+        let mut res = service.execute(CommandRequest::new_hget("t1", "k1"));
+        let data = res.next().await.unwrap();
+        assert_res_ok(&data, &["v1".into()], &[]);
     }

-    #[test]
-    fn event_registration_should_work() {
+    #[tokio::test]
+    async fn event_registration_should_work() {
         fn b(cmd: &CommandRequest) {
             info!("Got {:?}", cmd);
         }
@@ -200,10 +216,11 @@ mod tests {
             .fn_after_send(e)
             .into();

-        let res = service.execute(CommandRequest::new_hset("t1", "k1", "v1".into()));
-        assert_eq!(res.status, StatusCode::CREATED.as_u16() as u32);
-        assert_eq!(res.message, "");
-        assert_eq!(res.values, vec![Value::default()]);
+        let mut res = service.execute(CommandRequest::new_hset("t1", "k1", "v1".into()));
+        let data = res.next().await.unwrap();
+        assert_eq!(data.status, StatusCode::CREATED.as_u16() as u32);
+        assert_eq!(data.message, "");
+        assert_eq!(data.values, vec![Value::default()]);
     }
 }

diff --git a/39/kv/src/service/topic.rs b/39/kv/src/service/topic.rs
index 3037227..3c5874e 100644
--- a/39/kv/src/service/topic.rs
+++ b/39/kv/src/service/topic.rs
@@ -19,7 +19,7 @@ fn get_next_subscription_id() -> u32 {
     NEXT_ID.fetch_add(1, Ordering::Relaxed)
 }

-pub trait Topic {
+pub trait Topic: Send + Sync + 'static {
     /// 订阅某个主题
     fn subscribe(self, name: String) -> mpsc::Receiver<Arc<CommandResponse>>;
     /// 取消对主题的订阅
diff --git a/39/kv/src/service/topic_service.rs b/39/kv/src/service/topic_service.rs
new file mode 100644
index 0000000..c7f49c6
--- /dev/null
+++ b/39/kv/src/service/topic_service.rs
@@ -0,0 +1,33 @@
+use futures::{stream, Stream};
+use std::{pin::Pin, sync::Arc};
+use tokio_stream::wrappers::ReceiverStream;
+
+use crate::{CommandResponse, Publish, Subscribe, Topic, Unsubscribe};
+
+pub type StreamingResponse = Pin<Box<dyn Stream<Item = Arc<CommandResponse>> + Send>>;
+
+pub trait TopicService {
+    /// 处理 Command，返回 Response
+    fn execute(self, topic: impl Topic) -> StreamingResponse;
+}
+
+impl TopicService for Subscribe {
+    fn execute(self, topic: impl Topic) -> StreamingResponse {
+        let rx = topic.subscribe(self.topic);
+        Box::pin(ReceiverStream::new(rx))
+    }
+}
+
+impl TopicService for Unsubscribe {
+    fn execute(self, topic: impl Topic) -> StreamingResponse {
+        topic.unsubscribe(self.topic, self.id);
+        Box::pin(stream::once(async { Arc::new(CommandResponse::ok()) }))
+    }
+}
+
+impl TopicService for Publish {
+    fn execute(self, topic: impl Topic) -> StreamingResponse {
+        topic.publish(self.topic, Arc::new(self.data.into()));
+        Box::pin(stream::once(async { Arc::new(CommandResponse::ok()) }))
+    }
+}