krpc/

client.rs

use super::msg::*;
use std::collections::HashMap;
use std::future::Future;
pub use std::ops::Deref;
use std::pin::Pin;
use std::sync::atomic::{AtomicU32, Ordering};
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::TcpStream;
#[cfg(unix)]
use tokio::net::UnixStream;
use tokio::sync::{mpsc, oneshot};

/// 异步trait的返回类型
type Return<'a, T> = Pin<Box<dyn Future<Output = T> + Send + 'a>>;
/// 异步trait主要用以发送RPC请求并接收服务端回复
pub trait Call {
	fn call<'a>(&'a self, id: u32, data: Bytes) -> Return<'a, Result<Msg, Error>>;
}

/// 用以订阅回调处理
pub trait SubcribeCallback<T> {
	/// 返回值为true表示一直订阅,返回false表示只订阅一次
	fn callback(&mut self, data: T) -> bool;
}

pub struct MyStream {
	tx: mpsc::Sender<(u32, oneshot::Sender<Result<Msg, Error>>, Bytes)>,
}

impl MyStream {
	/// 创建多线程流
	fn new<Stream: AsyncReadExt + AsyncWriteExt + std::marker::Unpin + Send + 'static>(mut stream: Stream) -> Self {
		let (tx, mut rx) = mpsc::channel::<(u32, oneshot::Sender<Result<Msg, Error>>, Bytes)>(1);
		tokio::spawn(async move {
			let mut header = [0u8; RPC_HEADER_LEN];
			let mut callers = HashMap::<u32, oneshot::Sender<Result<Msg, Error>>>::new();
			let error = loop {
				tokio::select! {
					Some((id, otx, data)) = rx.recv() => {
						callers.insert(id, otx);
						if let Err(err) = stream.write_all(&data[..]).await {
							break Error::from(err);
						}
					}
					ret = stream.read_exact(&mut header[..]) => {
						match ret {
							Ok(0) => break Error::new("对端已关闭读取数据长度为0"),
							Ok(_) => {
								if let Ok(mut msg) = Msg::decode(&header[..]) {
									match msg.mode() {
										Mode::Respond | Mode::Publish => {
											if let Some(buf) = msg.body() {
												//读取消息体
												if let Err(err) = stream.read_exact(buf).await {
													break Error::from(err);
												}
											}
										}
										_ => (),
									}
									if let Some(otx) = callers.remove(&msg.id()) {
										let _ = otx.send(Ok(msg));
									}
								}
							}
							Err(err) => break Error::from(err),
						}
					}
				}
			};
			//通知所有调用者出错了
			callers.into_iter().for_each(|(_, otx)| {
				let _ = otx.send(Err(error.clone()));
			});
		});
		Self { tx }
	}
}
/// 为多线程流实现Call trait
impl Call for MyStream {
	fn call<'a>(&'a self, id: u32, data: Bytes) -> Return<'a, Result<Msg, Error>> {
		Box::pin(async move {
			let (tx, rx) = oneshot::channel();
			let _ = self.tx.send((id, tx, data)).await;
			rx.await.unwrap()
		})
	}
}

pub struct Client<Stream: Call + Send> {
	stream: Stream,
	id: AtomicU32,
}

const HEARTBEAT: &'static str = "heartbeat";

impl<Stream: Call + Send> Client<Stream> {
	/// 创建Client
	pub fn new(stream: Stream) -> Self {
		Self { stream, id: AtomicU32::new(0) }
	}
	/// 编码heartbeat请求
	fn encode_heartbeat(&self) -> (u32, Bytes) {
		let id = self.id.fetch_add(1, Ordering::SeqCst);
		let msg = Msg::new(id, HEARTBEAT);
		let data = msg.encode_without_body(Mode::HeartBeat);
		(id, data)
	}
	/// 编码无参数请求
	fn encode_without_arg(&self, name: &str) -> (u32, Bytes) {
		let id = self.id.fetch_add(1, Ordering::SeqCst);
		let msg = Msg::new(id, name);
		let data = msg.encode_without_body(Mode::Request);
		(id, data)
	}
	/// 编码有参数请求
	fn encode_with_arg<Args>(&self, name: &str, args: Args) -> (u32, Bytes)
	where
		Args: serde::ser::Serialize,
	{
		let id = self.id.fetch_add(1, Ordering::SeqCst);
		let msg = Msg::new(id, name);
		let data = msg.encode(Mode::Request, &args);
		(id, data)
	}
	/// 解码heartbeat回复
	fn decode_heartbeat(msg: Msg) -> Result<(), Error> {
		if msg.name() != HEARTBEAT.as_bytes() {
			return Err(Error::new("心跳回复函数名称不匹配"));
		}
		if !msg.headeronly() {
			return Err(Error::new("心跳不应当返回消息体"));
		}
		match msg.mode() {
			Mode::HeartBeat => Ok(()),
			_ => Err(Error::new("返回消息模式不正确")),
		}
	}
	/// 解码无返回值回复
	fn decode_without_ret(msg: Msg, name: &str) -> Result<(), Error> {
		if msg.name() != name.as_bytes() {
			return Err(Error::new("回复函数名称不匹配"));
		}
		if !msg.headeronly() {
			return Err(Error::new("不应当返回消息体"));
		}
		match msg.mode() {
			Mode::Respond => Ok(()),
			Mode::NotFound => Err(Error::new("没有找到相应的函数")),
			Mode::NotMatch => Err(Error::new("函数参数不匹配")),
			Mode::NoAccess => Err(Error::new("没有权限")),
			_ => Err(Error::new("返回消息模式不正确")),
		}
	}
	/// 解码有返回值回复
	fn decode_with_ret<Ret>(msg: Msg, name: &str) -> Result<Ret, Error>
	where
		Ret: for<'a> serde::de::Deserialize<'a>,
	{
		if msg.name() != name.as_bytes() {
			return Err(Error::new("回复函数名称不匹配"));
		}
		match msg.mode() {
			Mode::Respond => {
				if msg.headeronly() {
					Err(Error::new("没有消息体"))
				} else {
					msg.parse()
				}
			}
			Mode::NotFound => Err(Error::new("没有找到相应的函数")),
			Mode::NotMatch => Err(Error::new("函数参数不匹配")),
			Mode::NoAccess => Err(Error::new("没有权限")),
			_ => Err(Error::new("返回消息模式不正确")),
		}
	}

	/// 心跳检测
	#[inline]
	pub async fn heartbeat(&self) -> Result<(), Error> {
		let (id, data) = self.encode_heartbeat();
		let msg = self.stream.call(id, data).await?;
		Self::decode_heartbeat(msg)
	}
	/// 无参数无返回值的函数调用
	#[inline]
	pub async fn call_without_arg_ret(&self, name: &str) -> Result<(), Error> {
		let (id, data) = self.encode_without_arg(name);
		let msg = self.stream.call(id, data).await?;
		Self::decode_without_ret(msg, name)
	}
	/// 无参数有返回值的函数调用
	#[inline]
	pub async fn call_with_ret<Ret>(&self, name: &str) -> Result<Ret, Error>
	where
		Ret: for<'a> serde::de::Deserialize<'a>,
	{
		let (id, data) = self.encode_without_arg(name);
		let msg = self.stream.call(id, data).await?;
		Self::decode_with_ret(msg, name)
	}
	/// 有参数无返回值的函数调用
	#[inline]
	pub async fn call_with_arg<Args>(&self, name: &str, args: Args) -> Result<(), Error>
	where
		Args: serde::ser::Serialize,
	{
		let (id, data) = self.encode_with_arg(name, args);
		let msg = self.stream.call(id, data).await?;
		Self::decode_without_ret(msg, name)
	}
	/// 有参数也有返回值的函数调用
	#[inline]
	pub async fn call_with_arg_ret<Args, Ret>(&self, name: &str, args: Args) -> Result<Ret, Error>
	where
		Args: serde::ser::Serialize,
		Ret: for<'a> serde::de::Deserialize<'a>,
	{
		let (id, data) = self.encode_with_arg(name, args);
		let msg = self.stream.call(id, data).await?;
		Self::decode_with_ret(msg, name)
	}
	/// 订阅主题,通过lambda函数处理订阅数据
	pub async fn subcribe_with_lambda<Ret, F>(&self, topic: &str, mut f: F) -> Result<(), Error>
	where
		Ret: for<'a> serde::de::Deserialize<'a>,
		F: FnMut(Ret),
	{
		loop {
			let id = self.id.fetch_add(1, Ordering::SeqCst);
			let msg = Msg::new(id, topic);
			let data = msg.encode_without_body(Mode::Subcribe);
			let msg = self.stream.call(id, data).await?;
			if msg.name() != topic.as_bytes() {
				break Err(Error::new("订阅主题名称不匹配"));
			}
			match msg.mode() {
				Mode::Publish => {
					if msg.headeronly() {
						break Err(Error::new("没有订阅到消息体"));
					} else {
						f(msg.parse()?);
						continue;
					}
				}
				Mode::NotFound => break Err(Error::new("没有找到相应的函数")),
				Mode::NotMatch => break Err(Error::new("函数参数不匹配")),
				Mode::NoAccess => break Err(Error::new("没有权限")),
				_ => break Err(Error::new("返回消息模式不正确")),
			}
		}
	}
	/// 订阅主题,通过trait处理订阅数据
	pub async fn subcribe_with_trait<Ret, T>(&self, topic: &str, t: &mut T) -> Result<(), Error>
	where
		Ret: for<'a> serde::de::Deserialize<'a>,
		T: SubcribeCallback<Ret>,
	{
		loop {
			let id = self.id.fetch_add(1, Ordering::SeqCst);
			let msg = Msg::new(id, topic);
			let data = msg.encode_without_body(Mode::Subcribe);
			let msg = self.stream.call(id, data).await?;
			if msg.name() != topic.as_bytes() {
				break Err(Error::new("订阅主题名称不匹配"));
			}
			match msg.mode() {
				Mode::Publish => {
					if msg.headeronly() {
						break Err(Error::new("没有订阅到消息体"));
					} else {
						if t.callback(msg.parse()?) {
							continue; //callback返回true说明需要一直订阅
						} else {
							break Ok(()); //callback返回false说明只订阅一次直接退出loop
						}
					}
				}
				Mode::NotFound => break Err(Error::new("没有找到相应的函数")),
				Mode::NotMatch => break Err(Error::new("函数参数不匹配")),
				Mode::NoAccess => break Err(Error::new("没有权限")),
				_ => break Err(Error::new("返回消息模式不正确")),
			}
		}
	}
}

/// TCP client
pub type TCPClient = Client<MyStream>;
/// unix client
#[cfg(unix)]
pub type UnixClient = Client<MyStream>;

/// 创建TCP客户端
#[inline]
pub async fn new_tcp_client(addr: &str) -> std::io::Result<TCPClient> {
	Ok(Client::new(MyStream::new(TcpStream::connect(addr).await?)))
}
/// unix客户端
#[inline]
#[cfg(unix)]
pub async fn new_unix_client(path: &str) -> std::io::Result<UnixClient> {
	Ok(Client::new(MyStream::new(UnixStream::connect(path).await?)))
}

/// # RPC调用
/// - 第一个参数表示使用的传输协议,可选值**tcp**:使用tcp通信, **unix**:使用unix域套接字通信)
/// - 第二个参数为通信地址,类型为<font color=blue>&str</font>,例如:`"127.0.0.1:9000"`,`"/tmp/local/unix"`
/// - 剩下的参数为函数调用,格式如下所示:
/// 	- `rpc1()`,无参数无返回值
/// 	- `rpc2(id:i32)`,有参数无返回值
/// 	- `rpc3()->i32`,无参数有返回值
/// 	- `rpc4(a:i32,b:bool)->String`,既有参数也有返回值
/// - 返回值类型为**Result<R,Error>**,R为返回值类型,当无返回值时R为()
/// # example
/// ```
/// let _ = call!(unix, "/tmp/local/unix").await?;	//心跳检测
/// let _ = call!(unix, "/tmp/local/unix", test_notargs_and_notret()).await?;
/// let _ = call!(unix, "/tmp/local/unix", test_notret("hello client")).await?;
/// let ret = call!(tcp, "127.0.0.1:9000", test_notargs() -> String).await?;
/// assert_eq!(&ret, "test_notargs called!");
/// let ret = call!(tcp, "127.0.0.1:9000", test_has_args_and_ret(10000, 1000, 10, true) -> String).await?;
/// assert_eq!(&ret, "10000 1000 10 true");
/// ```
#[macro_export]
macro_rules! call {
	/* 以下四个匹配用以匹配函数名有空格的情况,匹配模型如下所示,若有需要可打开注释
	let _ = call!(tcp, "127.0.0.1:9000", "test notargs and notret").await?;
	let _ = call!(tcp, "127.0.0.1:9000", "test notret", ("hello client",)).await?;
	let ret = call!(tcp, "127.0.0.1:9000", "test notargs" -> String).await?;
	let ret = call!(tcp, "127.0.0.1:9000", "test has args and ret", (10000, 1000, 10, true) -> String).await?;
	//无参数且无返回值
	(@call $func:ident, $addr:tt, $name:tt) => {
		async move {
			match $crate::$func($addr).await {
				Ok(client) => client.call_without_arg_ret($name).await,
				Err(e) => Err($crate::msg::Error::from(e)),
			}
		}
	};
	//无参数有返回值
	(@call $func:ident, $addr:tt, $name:tt -> $ret:ty) => {
		async move {
			match $crate::$func($addr).await {
				Ok(client) => {
					let result: $ret = client.call_with_ret($name).await?;
					Ok(result)
				}
				Err(e) => Err($crate::msg::Error::from(e)),
			}
		}
	};
	//有参数无返回值
	(@call $func:ident, $addr:tt, $name:tt, $args:tt) => {
		async move {
			match $crate::$func($addr).await {
				Ok(client) => client.call_with_arg($name, $args).await,
				Err(e) => Err($crate::msg::Error::from(e)),
			}
		}
	};
	//有参数有返回值
	(@call $func:ident, $addr:tt, $name:tt, $args:tt -> $ret:ty) => {
		async move {
			match $crate::$func($addr).await {
				Ok(client) => {
					let result: $ret = client.call_with_arg_ret($name, $args).await?;
					Ok(result)
				}
				Err(e) => Err($crate::msg::Error::from(e)),
			}
		}
	};
 */
	//心跳检测
	(@call $connect:ident, $addr:expr) => {
		async move {
			match $crate::$connect($addr).await {
				Ok(client) => client.heartbeat().await,
				Err(e) => Err($crate::msg::Error::from(e)),
			}
		}
	};
	//无参数且无返回值
	(@call $connect:ident, $addr:expr, $func:ident()) => {
		async move {
			match $crate::$connect($addr).await {
				Ok(client) => client.call_without_arg_ret(stringify!($func)).await,
				Err(e) => Err($crate::msg::Error::from(e)),
			}
		}
	};
	//无参数有返回值
	(@call $connect:ident, $addr:expr, $func:ident() -> $ret:ty) => {
		async move {
			match $crate::$connect($addr).await {
				Ok(client) => {
					let result: $ret = client.call_with_ret(stringify!($func)).await?;
					Ok(result)
				}
				Err(e) => Err($crate::msg::Error::from(e)),
			}
		}
	};
	//有参数无返回值
	(@call $connect:ident, $addr:expr, $func:ident($($arg:expr),+)) => {
		async move {
			match $crate::$connect($addr).await {
				Ok(client) => client.call_with_arg(stringify!($func), ($($arg,)+)).await,
				Err(e) => Err($crate::msg::Error::from(e)),
			}
		}
	};
	//有参数有返回值
	(@call $connect:ident, $addr:expr, $func:ident($($arg:expr),+) -> $ret:ty) => {
		async move {
			match $crate::$connect($addr).await {
				Ok(client) => {
					let result: $ret = client.call_with_arg_ret(stringify!($func), ($($arg,)+)).await?;
					Ok(result)
				}
				Err(e) => Err($crate::msg::Error::from(e)),
			}
		}
	};
	(tcp, $($var:tt)+) => {
		async move {
			call!(@call new_tcp_client, $($var)+).await
		}
	};
	(unix, $($var:tt)+) => {
		async move {
			call!(@call new_unix_client, $($var)+).await
		}
	};
}

/// # 订阅主题
/// - 第一个参数表示使用的传输协议,可选值**tcp**:使用tcp通信, **unix**:使用unix域套接字通信)
/// - 第二个参数为通信地址,类型为<font color=blue>&str</font>,例如:`"127.0.0.1:9000"`,`"/tmp/local/unix"`
/// - 第三个参数为订阅主题,类型为<font color=blue>&str</font>,例如:`"onMessage"`
/// - 第四个参数为订阅数据回调处理,支持两种格式:
/// 	- `|id:i32|{})`,lambda回调处理,lambda函数可以有多个参数但每个参数需实现**serde::de::Deserialize**,另外lambda函数返回值类型为<font color=blue>()</font>,且lambda函数体**需用`{}`包裹**
/// 	- `var`,var类型为**&mut**,且需要实现**SubcribeCallback<T>**,T实现**serde::de::Deserialize**,T即为订阅的数据类型
/// - 返回值类型为**Result<(),Error>**,<font color=red>此订阅会一直运行只有出错时才会返回</font>
/// # example
/// ```
/// //通过lambda进行回调处理
/// let _ = subcribe!(unix, "/tmp/local/unix", "sub", |s: String, v: i32| {
/// 	println!("({}, {})", s, v);
/// }).await;
///
/// //实现SubcribeCallback trait进行回调处理
/// struct MySub;
/// impl SubcribeCallback<(String, i32)> for MySub {
/// 	fn callback(&mut self, data: (String, i32)) -> bool {
/// 		false
/// 	}
/// }
///
/// let mut sub = MySub;
/// let _ = subcribe!(unix, "/tmp/local/unix", "sub", sub).await;
/// ```
#[macro_export]
macro_rules! subcribe {
	//订阅主题并通过lambda函数回调处理
	(@sub $connect:ident, $addr:expr, $topic:expr, |$($arg:ident:$argType:ty),+|$body:block) => {
		async move {
			match $crate::$connect($addr).await {
				Ok(client) => client.subcribe_with_lambda($topic, |($($arg,)+):($($argType,)+)|$body).await,
				Err(e) => Err($crate::msg::Error::from(e)),
			}
		}
	};
	//订阅主题并通过实现SubcribeCallback trait回调处理
	(@sub $connect:ident, $addr:expr, $topic:expr, $var:expr) => {
		async move {
			match $crate::$connect($addr).await {
				Ok(client) => client.subcribe_with_trait($topic, $var).await,
				Err(e) => Err($crate::msg::Error::from(e)),
			}
		}
	};
	(tcp, $($var:tt)+) => {
		async move {
			subcribe!(@sub new_tcp_client, $($var)+).await
		}
	};
	(unix, $($var:tt)+) => {
		async move {
			subcribe!(@sub new_unix_client, $($var)+).await
		}
	};
}

/// 用以定义新类型
#[macro_export]
macro_rules! define_new_type {
	//定义无参数无返回值方法
	(@method fn $name:ident$(<$generic:tt>)?()) => {
		pub async fn $name$(<$generic>)?(&self) -> Result<(), $crate::msg::Error> {
			self.0.call_without_arg_ret(stringify!($name)).await
		}
	};
	//定义有参数无返回值方法
	(@method fn $name:ident$(<$generic:tt>)?($($arg:ident:$argType:ty,)+)) => {
		pub async fn $name$(<$generic>)?(&self, $($arg:$argType,)+) -> Result<(), $crate::msg::Error>
		where
			$($argType: serde::ser::Serialize,)+
		{
			self.0.call_with_arg(stringify!($name), ($($arg,)+)).await
		}
	};
	//定义无参数有返回值方法
	(@method fn $name:ident$(<$generic:tt>)?()->$ret:ty) => {
		pub async fn $name$(<$generic>)?(&self) -> Result<$ret, $crate::msg::Error>
		where
			$ret: for<'a> serde::de::Deserialize<'a>,
		{
			self.0.call_with_ret(stringify!($name)).await
		}
	};
	//定义有参数有返回值方法
	(@method fn $name:ident$(<$generic:tt>)?($($arg:ident:$argType:ty,)+)->$ret:ty) => {
		pub async fn $name$(<$generic>)?(&self, $($arg:$argType,)+) -> Result<$ret, $crate::msg::Error>
		where
			$($argType: serde::ser::Serialize,)+
			$ret: for<'a> serde::de::Deserialize<'a>,
		{
			self.0.call_with_arg_ret(stringify!($name), ($($arg,)+)).await
		}
	};
	//定义sub onOpen(topic:&str,f:impl FnMut(ArgType))格式的订阅方法
	(@method sub $name:ident($topic:ident:$topicType:ty, $f:ident:$(impl)? FnMut($ArgType:ty))) => {
		pub async fn $name<F>(&self, $topic:$topicType, $f:F) -> Result<(), $crate::msg::Error>
		where
			F: FnMut($ArgType),
			$topicType: Deref<Target = str>,
			$ArgType: for<'a> serde::de::Deserialize<'a>,
		{
			self.0.subcribe_with_lambda(&$topic, $f).await
		}
	};
	//定义sub onClose(topic:&str,var:&mut V)格式的订阅方法
	(@method sub $name:ident($topic:ident:$topicType:ty, $var:ident:&mut $ArgType:ty)) => {
		pub async fn $name<Ret>(&self, $topic:&str, $var:&mut $ArgType) -> Result<(), $crate::msg::Error>
		where
			Ret: for<'a> serde::de::Deserialize<'a>,
			$topicType: Deref<Target = str>,
			$ArgType: SubcribeCallback<Ret>,
		{
			self.0.subcribe_with_trait(&$topic, $var).await
		}
	};
	//定义只有fn方法声明的新类型
	($f:ident, $t:ident, $StructName:ident, $(fn $name:ident$(<$generic:tt>)?($($arg:ident:$argType:ty),*)$(->$ret:ty)?),+) => {
		struct $StructName($t);
		impl $StructName {
			pub async fn new(path:&str) -> std::io::Result<Self> {
				Ok(Self($f(path).await?))
			}
			pub async fn heartbeat(&self) -> Result<(), $crate::msg::Error> {
				self.0.heartbeat().await
			}
			$(define_new_type!(@method fn $name$(<$generic>)?($($arg:$argType,)*)$(->$ret)?);)+
		}
	};
	//定义只有sub方法声明的新类型
	($f:ident, $t:ident, $StructName:ident, $(sub $name:ident($topic:ident:$topicType:ty, $arg:ident:$($argType:tt)+)),+) => {
		struct $StructName($t);
		impl $StructName {
			pub async fn new(path:&str) -> std::io::Result<Self> {
				Ok(Self($f(path).await?))
			}
			pub async fn heartbeat(&self) -> Result<(), $crate::msg::Error> {
				self.0.heartbeat().await
			}
			$(define_new_type!(@method sub $name($topic:$topicType,$arg:$($argType)+));)+
		}
	};
	//定义同时包含fn和sub方法声明的新类型
	($f:ident, $t:ident, $StructName:ident, $(fn $name:ident($($arg:ident:$argType:ty),*)$(->$ret:ty)?),+, $(sub $name2:ident($topic:ident:$topicType:ty, $arg2:ident:$($argType2:tt)+)),+) => {
		struct $StructName($t);
		impl $StructName {
			pub async fn new(path:&str) -> std::io::Result<Self> {
				Ok(Self($f(path).await?))
			}
			pub async fn heartbeat(&self) -> Result<(), $crate::msg::Error> {
				self.0.heartbeat().await
			}
			$(define_new_type!(@method fn $name($($arg:$argType,)*)$(->$ret)?);)+
			$(define_new_type!(@method sub $name2($topic:$topicType,$arg2:$($argType2)+));)+
		}
	};
}

/// # 定义RPC Client的新类型
/// - 第一个参数表示使用的传输协议,可选值**tcp**:使用tcp通信, **unix**:使用unix域套接字通信)
/// - 第二个参数为新类型的名称,例如`MyStruct`
/// - 剩下的参数为新类型的方法声明(可以声明多个),支持格式如下:
/// 	- `fn rpc1()`,无参数无返回值声明
/// 	- `fn rpc2(id:i32)`,有参数无返回值声明
/// 	- `fn rpc3()->i32`,无参数有返回值声明
/// 	- `fn rpc4(a:i32,b:bool)->String`,既有参数也有返回值声明
/// 	- `sub onOpen(topic:TopicType,f:impl FnMut(ArgType))`,订阅主题通过lambda函数回调处理,其中TopicType必须实现**Deref<Target = str>**,ArgType需实现**serde::de::Deserialize**,另外lambda函数返回值类型为**()**
/// 	- `sub onClose(topic:TopicType,var:&mut V)`,订阅主题通过SubcribeCallback trait处理,其中TopicType必须实现**Deref<Target = str>**,var类型必须为**&mut**,V需要实现**SubcribeCallback<T>**,T实现**serde::de::Deserialize**,T即为订阅的数据类型
/// - 注意:
/// 	- 此宏会默认一个pub async fn new(path:&str) -> std::io::Result<Self>的方法,用以创建新定义的类型
/// 	- 此宏会默认一个pub async fn heartbeat(&self) -> Result<(), Error>的方法,用以检测服务端是否存在
/// 	- 不需要在方法参数表第一个参数指明self,此宏会自己生成
/// 	- 当同时存在fn与sub方法声明时,以**fn**开头的所有方法声明必须在以**sub**的前面,即先**fn声明在前,sub声明在后**
/// # example
/// ```
/// //实现SubcribeCallback trait进行回调处理
/// struct MySub;
/// impl SubcribeCallback<(String, i32)> for MySub {
/// 	fn callback(&mut self, data: (String, i32)) -> bool {
/// 		false
/// 	}
/// }
/// 只定义rpc方法
/// define!(unix, Test1,
/// 	fn rpc1(),
/// 	fn rpc2(id:i32),
/// 	fn rpc3()->i32,
/// 	fn rpc4(a:i32,b:bool)->String
/// );
/// 只定义订阅方法
/// define!(unix, Test2,
/// 	sub onOpen(topic:&'static str, f:impl FnMut((String,i32))),
/// 	sub onClose(topic:&'static str, var:&mut MySub)
/// );
/// 定义rpc方法和订阅方法
/// define!(unix, Test,
/// 	fn rpc1(),
/// 	fn rpc2(id:i32),
/// 	fn rpc3()->i32,
/// 	fn rpc4(a:i32,b:bool)->String,
/// 	sub onOpen(topic:&'static str, f:impl FnMut((String,i32))),
/// 	sub onClose(topic:&'static str, var:&mut MySub)
/// );
/// //如上定义会生成一个新类型Test,包含方法rpc1、rpc2、rpc3、rpc4、onOpen、onClose生成的新类型代码大致如下所示:
/// struct Test(T);
/// impl Test {
/// 	pub async fn new(path:&str) -> std::io::Result<Self> {...} //默认生成的方法
/// 	pub async fn heartbeat(&self) -> Result<(), Error> {...} //默认生成的方法,用以检测服务端是否存在
/// 	pub async fn rpc1(&self) -> Result<(), Error> {...}
/// 	pub async fn rpc2(&self, id:i32) -> Result<(), Error> {...}
/// 	pub async fn rpc3(&self) -> Result<i32, Error> {...}
/// 	pub async fn rpc4(&self, a:i32, b:bool) -> Result<String, Error> {...}
/// 	pub async fn onOpen(topic:&'static str, f:impl FnMut(String,i32)) -> Result<(), Error> {...} //此方法会一直执行直到出错才返回
/// 	pub async fn onClose(topic:&'static str, var:&mut MySub) -> Result<(), Error> {...} //此方法若var.callback()的返回值是true就一直执行直到出错才返回,若返回值是false则立即返回Ok(())
/// }
/// ```
#[macro_export]
macro_rules! define {
	(tcp, $($var:tt)+) => {
		define_new_type!(new_tcp_client, TCPClient, $($var)+);
	};
	(unix, $($var:tt)+) => {
		define_new_type!(new_unix_client, UnixClient, $($var)+);
	};
}