|
|
@@ -23,6 +23,7 @@ use crate::response_base::ResponseData;
|
|
|
pub enum HeartbeatType {
|
|
|
Ping,
|
|
|
Pong,
|
|
|
+ Custom(String),
|
|
|
}
|
|
|
|
|
|
pub struct AbstractWsMode {}
|
|
|
@@ -56,123 +57,199 @@ impl AbstractWsMode {
|
|
|
};
|
|
|
|
|
|
loop {
|
|
|
- let (ws_stream, _) = connect_async(address_url.clone(), proxy).await?;
|
|
|
- trace!("WebSocket 握手完成。");
|
|
|
- let (write, mut read) = ws_stream.split();
|
|
|
-
|
|
|
- let write_arc = Arc::new(Mutex::new(write));
|
|
|
- let write_clone = Arc::clone(&write_arc);
|
|
|
- //订阅写入(包括订阅信息 )
|
|
|
- trace!("订阅内容：{:?}",subscribe_array.clone());
|
|
|
- for s in &subscribe_array {
|
|
|
- let mut write_lock = write_clone.lock().await;
|
|
|
- write_lock.send(Message::Text(s.parse().unwrap())).await?;
|
|
|
- }
|
|
|
+ match connect_async(address_url.clone(), proxy).await {
|
|
|
+ Ok((ws_stream, _)) => {
|
|
|
+ trace!("WebSocket 握手完成。");
|
|
|
+ let (write, mut read) = ws_stream.split();
|
|
|
+
|
|
|
+ let write_arc = Arc::new(Mutex::new(write));
|
|
|
+ let write_clone = Arc::clone(&write_arc);
|
|
|
+ //订阅写入(包括订阅信息 )
|
|
|
+ trace!("订阅内容：{:?}",subscribe_array.clone());
|
|
|
+ for s in &subscribe_array {
|
|
|
+ let mut write_lock = write_clone.lock().await;
|
|
|
+ write_lock.send(Message::Text(s.parse().unwrap())).await?;
|
|
|
+ }
|
|
|
|
|
|
- //将socket 的写操作与 写通道链接起来，将数据以ok的结构体封装进行传递
|
|
|
- // let stdin_to_ws = write_rx.map(Ok).forward(write);
|
|
|
- // Writing task
|
|
|
- let write_clone2 = Arc::clone(&write_arc);
|
|
|
- let stdin_to_ws = async {
|
|
|
- while let Some(message) = write_rx.next().await {
|
|
|
- let mut write_lock2 = write_clone2.lock().await;
|
|
|
- write_lock2.send(message).await?;
|
|
|
- }
|
|
|
- Ok::<(), tokio_tungstenite::tungstenite::Error>(())
|
|
|
- };
|
|
|
- let write_clone3 = Arc::clone(&write_arc);
|
|
|
- let ws_to_stdout = async {
|
|
|
- while let Some(message) = read.next().await {
|
|
|
- let mut write_lock3 = write_clone3.lock().await;
|
|
|
- let response_data = AbstractWsMode::analysis_message(message, message_text, message_ping, message_pong);
|
|
|
- // let response_data = func(message);
|
|
|
- if response_data.is_some() {
|
|
|
- let mut data = response_data.unwrap();
|
|
|
- data.label = lable.clone();
|
|
|
- let code = data.code.clone();
|
|
|
- /*
|
|
|
- 200 -正确返回
|
|
|
- -200 -登录成功
|
|
|
- -201 -订阅成功
|
|
|
- -300 -客户端收到服务器心跳ping，需要响应
|
|
|
- -301 -客户端收到服务器心跳pong，需要响应
|
|
|
- -302 -客户端收到服务器心跳自定义，需要响应自定义
|
|
|
- */
|
|
|
- match code.as_str() {
|
|
|
- "200" => {
|
|
|
- if bool_v1.load(Ordering::Relaxed) {
|
|
|
- read_tx.unbounded_send(data).unwrap();
|
|
|
- }
|
|
|
- }
|
|
|
- "-200" => {
|
|
|
- //登录成功
|
|
|
- trace!("登录成功：{:?}", data);
|
|
|
- }
|
|
|
- "-201" => {
|
|
|
- //订阅成功
|
|
|
- trace!("订阅成功：{:?}", data);
|
|
|
- }
|
|
|
- "-300" => {
|
|
|
- //服务器发送心跳 ping 给客户端，客户端需要pong回应
|
|
|
- trace!("服务器响应-ping");
|
|
|
- if data.data.len() > 0 {
|
|
|
- write_lock3.send(Message::Pong(Vec::from(data.data))).await?;
|
|
|
- trace!("客户端回应服务器-pong");
|
|
|
- }
|
|
|
- }
|
|
|
- "-301" => {
|
|
|
- //服务器发送心跳 pong 给客户端，客户端需要ping回应
|
|
|
- trace!("服务器响应-pong");
|
|
|
- if data.data.len() > 0 {
|
|
|
- write_lock3.send(Message::Ping(Vec::from(data.data))).await?;
|
|
|
- trace!("客户端回应服务器-ping");
|
|
|
+ //将socket 的写操作与 写通道链接起来，将数据以ok的结构体封装进行传递
|
|
|
+ // let stdin_to_ws = write_rx.map(Ok).forward(write);
|
|
|
+ // Writing task
|
|
|
+ let write_clone2 = Arc::clone(&write_arc);
|
|
|
+ let stdin_to_ws = async {
|
|
|
+ while let Some(message) = write_rx.next().await {
|
|
|
+ let mut write_lock2 = write_clone2.lock().await;
|
|
|
+ write_lock2.send(message).await?;
|
|
|
+ }
|
|
|
+ Ok::<(), tokio_tungstenite::tungstenite::Error>(())
|
|
|
+ };
|
|
|
+ let write_clone3 = Arc::clone(&write_arc);
|
|
|
+ let ws_to_stdout = async {
|
|
|
+ while let Some(message) = read.next().await {
|
|
|
+ let mut write_lock3 = write_clone3.lock().await;
|
|
|
+ let response_data = AbstractWsMode::analysis_message(message, message_text, message_ping, message_pong);
|
|
|
+ // let response_data = func(message);
|
|
|
+ if response_data.is_some() {
|
|
|
+ let mut data = response_data.unwrap();
|
|
|
+ data.label = lable.clone();
|
|
|
+ let code = data.code.clone();
|
|
|
+ /*
|
|
|
+ 200 -正确返回
|
|
|
+ -200 -登录成功
|
|
|
+ -201 -订阅成功
|
|
|
+ -300 -客户端收到服务器心跳ping，需要响应
|
|
|
+ -301 -客户端收到服务器心跳pong，需要响应
|
|
|
+ -302 -客户端收到服务器心跳自定义，需要响应自定义
|
|
|
+ */
|
|
|
+ match code.as_str() {
|
|
|
+ "200" => {
|
|
|
+ if bool_v1.load(Ordering::Relaxed) {
|
|
|
+ read_tx.unbounded_send(data).unwrap();
|
|
|
+ }
|
|
|
+ }
|
|
|
+ "-200" => {
|
|
|
+ //登录成功
|
|
|
+ trace!("登录成功：{:?}", data);
|
|
|
+ }
|
|
|
+ "-201" => {
|
|
|
+ //订阅成功
|
|
|
+ trace!("订阅成功：{:?}", data);
|
|
|
+ }
|
|
|
+ "-300" => {
|
|
|
+ //服务器发送心跳 ping 给客户端，客户端需要pong回应
|
|
|
+ trace!("服务器响应-ping");
|
|
|
+ if data.data.len() > 0 {
|
|
|
+ write_lock3.send(Message::Pong(Vec::from(data.data))).await?;
|
|
|
+ trace!("客户端回应服务器-pong");
|
|
|
+ }
|
|
|
+ }
|
|
|
+ "-301" => {
|
|
|
+ //服务器发送心跳 pong 给客户端，客户端需要ping回应
|
|
|
+ trace!("服务器响应-pong");
|
|
|
+ if data.data.len() > 0 {
|
|
|
+ write_lock3.send(Message::Ping(Vec::from(data.data))).await?;
|
|
|
+ trace!("客户端回应服务器-ping");
|
|
|
+ }
|
|
|
+ }
|
|
|
+ "-302" => {
|
|
|
+ //客户端收到服务器心跳自定义，需要响应自定义
|
|
|
+ trace!("特定字符心跳，特殊响应：{:?}", data);
|
|
|
+ write_lock3.send(Message::Text(data.data)).await?;
|
|
|
+ trace!("特殊字符心跳-回应完成");
|
|
|
+ }
|
|
|
+ _ => {
|
|
|
+ trace!("未知：{:?}",data);
|
|
|
+ }
|
|
|
}
|
|
|
}
|
|
|
- "-302" => {
|
|
|
- //客户端收到服务器心跳自定义，需要响应自定义
|
|
|
- trace!("特定字符心跳，特殊响应：{:?}", data);
|
|
|
- write_lock3.send(Message::Text(data.data)).await?;
|
|
|
- trace!("特殊字符心跳-回应完成");
|
|
|
- }
|
|
|
- _ => {
|
|
|
- trace!("未知：{:?}",data);
|
|
|
- }
|
|
|
}
|
|
|
- }
|
|
|
+ Ok::<(), tokio_tungstenite::tungstenite::Error>(())
|
|
|
+ };
|
|
|
+
|
|
|
+ //必须操作。，因为不同于其他的高级语言，有自动内存管理，所以为了防范地址改变，所以需要做此处理
|
|
|
+ pin_mut!(stdin_to_ws, ws_to_stdout,);
|
|
|
+ future::select(stdin_to_ws, ws_to_stdout).await;
|
|
|
+ }
|
|
|
+ Err(e) => {
|
|
|
+ trace!("WebSocket 握手失败：{:?}",e);
|
|
|
}
|
|
|
- Ok::<(), tokio_tungstenite::tungstenite::Error>(())
|
|
|
- };
|
|
|
- // let ws_to_stdout = {
|
|
|
- // trace!("---1");
|
|
|
- // //读，循环读取，然后拿到 message,，然后开启异步处理 message，
|
|
|
- // let result = read.for_each(|message| async {
|
|
|
- // let response_data = func(message);
|
|
|
+ }
|
|
|
+ trace!("---5");
|
|
|
+ trace!("---4");
|
|
|
+ trace!("重启...");
|
|
|
+
|
|
|
+ // let (ws_stream, _) = connect_async(address_url.clone(), proxy).await.unwrap();
|
|
|
+ // trace!("WebSocket 握手完成。");
|
|
|
+ // let (write, mut read) = ws_stream.split();
|
|
|
+ //
|
|
|
+ // let write_arc = Arc::new(Mutex::new(write));
|
|
|
+ // let write_clone = Arc::clone(&write_arc);
|
|
|
+ // //订阅写入(包括订阅信息 )
|
|
|
+ // trace!("订阅内容：{:?}",subscribe_array.clone());
|
|
|
+ // for s in &subscribe_array {
|
|
|
+ // let mut write_lock = write_clone.lock().await;
|
|
|
+ // write_lock.send(Message::Text(s.parse().unwrap())).await?;
|
|
|
+ // }
|
|
|
+ //
|
|
|
+ // //将socket 的写操作与 写通道链接起来，将数据以ok的结构体封装进行传递
|
|
|
+ // // let stdin_to_ws = write_rx.map(Ok).forward(write);
|
|
|
+ // // Writing task
|
|
|
+ // let write_clone2 = Arc::clone(&write_arc);
|
|
|
+ // let stdin_to_ws = async {
|
|
|
+ // while let Some(message) = write_rx.next().await {
|
|
|
+ // let mut write_lock2 = write_clone2.lock().await;
|
|
|
+ // write_lock2.send(message).await?;
|
|
|
+ // }
|
|
|
+ // Ok::<(), tokio_tungstenite::tungstenite::Error>(())
|
|
|
+ // };
|
|
|
+ // let write_clone3 = Arc::clone(&write_arc);
|
|
|
+ // let ws_to_stdout = async {
|
|
|
+ // while let Some(message) = read.next().await {
|
|
|
+ // let mut write_lock3 = write_clone3.lock().await;
|
|
|
+ // let response_data = AbstractWsMode::analysis_message(message, message_text, message_ping, message_pong);
|
|
|
+ // // let response_data = func(message);
|
|
|
// if response_data.is_some() {
|
|
|
// let mut data = response_data.unwrap();
|
|
|
// data.label = lable.clone();
|
|
|
// let code = data.code.clone();
|
|
|
- // if code.as_str() == "-1" {
|
|
|
- // // let close_frame = CloseFrame {
|
|
|
- // // code: CloseCode::Normal,
|
|
|
- // // reason: Cow::Borrowed("Bye bye"),
|
|
|
- // // };
|
|
|
- // // let close_message = Message::Close(Some(close_frame));
|
|
|
- // // write.send(close_message);
|
|
|
- // } else if code.as_str() == "200" {
|
|
|
- // read_tx.unbounded_send(data).unwrap();
|
|
|
+ // /*
|
|
|
+ // 200 -正确返回
|
|
|
+ // -200 -登录成功
|
|
|
+ // -201 -订阅成功
|
|
|
+ // -300 -客户端收到服务器心跳ping，需要响应
|
|
|
+ // -301 -客户端收到服务器心跳pong，需要响应
|
|
|
+ // -302 -客户端收到服务器心跳自定义，需要响应自定义
|
|
|
+ // */
|
|
|
+ // match code.as_str() {
|
|
|
+ // "200" => {
|
|
|
+ // if bool_v1.load(Ordering::Relaxed) {
|
|
|
+ // read_tx.unbounded_send(data).unwrap();
|
|
|
+ // }
|
|
|
+ // }
|
|
|
+ // "-200" => {
|
|
|
+ // //登录成功
|
|
|
+ // trace!("登录成功：{:?}", data);
|
|
|
+ // }
|
|
|
+ // "-201" => {
|
|
|
+ // //订阅成功
|
|
|
+ // trace!("订阅成功：{:?}", data);
|
|
|
+ // }
|
|
|
+ // "-300" => {
|
|
|
+ // //服务器发送心跳 ping 给客户端，客户端需要pong回应
|
|
|
+ // trace!("服务器响应-ping");
|
|
|
+ // if data.data.len() > 0 {
|
|
|
+ // write_lock3.send(Message::Pong(Vec::from(data.data))).await?;
|
|
|
+ // trace!("客户端回应服务器-pong");
|
|
|
+ // }
|
|
|
+ // }
|
|
|
+ // "-301" => {
|
|
|
+ // //服务器发送心跳 pong 给客户端，客户端需要ping回应
|
|
|
+ // trace!("服务器响应-pong");
|
|
|
+ // if data.data.len() > 0 {
|
|
|
+ // write_lock3.send(Message::Ping(Vec::from(data.data))).await?;
|
|
|
+ // trace!("客户端回应服务器-ping");
|
|
|
+ // }
|
|
|
+ // }
|
|
|
+ // "-302" => {
|
|
|
+ // //客户端收到服务器心跳自定义，需要响应自定义
|
|
|
+ // trace!("特定字符心跳，特殊响应：{:?}", data);
|
|
|
+ // write_lock3.send(Message::Text(data.data)).await?;
|
|
|
+ // trace!("特殊字符心跳-回应完成");
|
|
|
+ // }
|
|
|
+ // _ => {
|
|
|
+ // trace!("未知：{:?}",data);
|
|
|
+ // }
|
|
|
// }
|
|
|
// }
|
|
|
- // });
|
|
|
- // trace!("---3");
|
|
|
- // result
|
|
|
+ // }
|
|
|
+ // Ok::<(), tokio_tungstenite::tungstenite::Error>(())
|
|
|
// };
|
|
|
-
|
|
|
- //必须操作。，因为不同于其他的高级语言，有自动内存管理，所以为了防范地址改变，所以需要做此处理
|
|
|
- pin_mut!(stdin_to_ws, ws_to_stdout,);
|
|
|
- future::select(stdin_to_ws, ws_to_stdout).await;
|
|
|
- trace!("---5");
|
|
|
- trace!("---4");
|
|
|
- trace!("重启...");
|
|
|
+ //
|
|
|
+ // //必须操作。，因为不同于其他的高级语言，有自动内存管理，所以为了防范地址改变，所以需要做此处理
|
|
|
+ // pin_mut!(stdin_to_ws, ws_to_stdout,);
|
|
|
+ // future::select(stdin_to_ws, ws_to_stdout).await;
|
|
|
+ // trace!("---5");
|
|
|
+ // trace!("---4");
|
|
|
+ // trace!("重启...");
|
|
|
}
|
|
|
// return Ok(());
|
|
|
}
|
|
|
@@ -190,6 +267,9 @@ impl AbstractWsMode {
|
|
|
HeartbeatType::Pong => {
|
|
|
Message::Pong(Vec::from("Pong"))
|
|
|
}
|
|
|
+ HeartbeatType::Custom(ref str) => {
|
|
|
+ Message::Text(str.parse().unwrap())
|
|
|
+ }
|
|
|
}
|
|
|
).expect("发送失败");
|
|
|
trace!("发送指令-心跳：{:?}",h_type);
|
