krpc 0.2.0

#[derive(Debug)]
struct Pair {
	msg: Msg,
	tx: mpsc::Sender<Bytes>,
}

#[derive(Debug)]
enum Content {
	Sub(u32, Pair),
	Pub(&'static str, Vec<u8>),
	Del(u32),
}

struct PubSub {
	subs: HashMap<String, HashMap<u32, Option<Pair>>>,
}

impl PubSub {
	// 添加订阅者
	async fn add(&mut self, id: u32, pair: Pair, on_subcribe: &mut (impl FnMut(&str) -> Option<Bytes> + Send + 'static)) {
		let name = String::from_utf8(pair.msg.name().to_vec()).unwrap();
		if let Some(pairs) = self.subs.get_mut(&name) {
			if pairs.get(&id).is_none() {
				if let Some(data) = on_subcribe(&name) {
					let _ = pair.tx.send(pair.msg.encode_with_bytes(Mode::Publish, &data)).await;
					pairs.insert(id, None);
					return;
				}
			}
			pairs.insert(id, Some(pair));
		} else {
			if let Some(data) = on_subcribe(&name) {
				let _ = pair.tx.send(pair.msg.encode_with_bytes(Mode::Publish, &data)).await;
				self.subs.insert(name, HashMap::from([(id, None)]));
			} else {
				self.subs.insert(name, HashMap::from([(id, Some(pair))]));
			}
		}
	}
	fn del(&mut self, id: u32) {
		for (_, pairs) in &mut self.subs {
			let _ = pairs.remove(&id);
		}
	}
	/// 若是发布内容过快可能会丢失,因为每次发布之后需要接收到订阅方的订阅信息才能再次发布
	async fn publish(&mut self, name: &'static str, data: Vec<u8>) {
		if let Some(pairs) = self.subs.get_mut(name) {
			for (_, pair) in pairs {
				if let Some(pair) = pair.take() {
					let _ = pair.tx.send(pair.msg.encode_with_bytes(Mode::Publish, &data)).await;
				}
			}
		}
	}
}

#[derive(Clone)]
pub struct Publisher {
	tx: mpsc::Sender<Content>,
}

impl Publisher {
	#[allow(unused_must_use)]
	pub fn push<Args>(&self, topic: &'static str, args: Args)
	where
		Args: serde::ser::Serialize,
	{
		let data = rmps::encode::to_vec(&args).unwrap();
		let tx = self.tx.clone();
		tokio::spawn(async move {
			tx.send(Content::Pub(topic, data)).await;
		});
	}
}