Skip to content

Commit

Permalink
wip remove E
Browse files Browse the repository at this point in the history
  • Loading branch information
Itsusinn逸新 committed Feb 7, 2024
1 parent a31e7e1 commit f29b7d9
Show file tree
Hide file tree
Showing 2 changed files with 27 additions and 31 deletions.
42 changes: 19 additions & 23 deletions src/obc/app_obc/app_ws.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@ use crate::{

use std::sync::Arc;

use color_eyre::eyre;
use futures_util::{SinkExt, StreamExt};
use serde::{Deserialize, Serialize};
use tokio::net::TcpListener;
Expand All @@ -25,23 +26,21 @@ use tokio_tungstenite::tungstenite::http::{header::USER_AGENT, Request};
use tokio_tungstenite::tungstenite::Message as WsMsg;
use tokio_tungstenite::WebSocketStream;
use tracing::{info, warn};
use color_eyre::eyre;

impl<A, R> AppOBC<A, R>
where
A: ProtocolItem,
R: ProtocolItem,
{
pub(crate) async fn ws<E, AH, EH>(
pub(crate) async fn ws<AH, EH>(
&self,
ob: &Arc<OneBot<AH, EH>>,
config: Vec<WebSocketClient>,
tasks: &mut Vec<JoinHandle<()>>,
) -> WalleResult<()>
where
E: ProtocolItem + GetSelf + Clone,
AH: ActionHandler<E, A, R> + Send + Sync + 'static,
EH: EventHandler<E, A, R> + Send + Sync + 'static,
AH: ActionHandler<Event, A, R> + Send + Sync + 'static,
EH: EventHandler<Event, A, R> + Send + Sync + 'static,
{
for wsc in config {
info!(target: super::OBC, "Start try connect to {}", wsc.url);
Expand Down Expand Up @@ -77,16 +76,15 @@ where
}
Ok(())
}
pub(crate) async fn wsr<E, AH, EH>(
pub(crate) async fn wsr<AH, EH>(
&self,
ob: &Arc<OneBot<AH, EH>>,
config: Vec<WebSocketServer>,
tasks: &mut Vec<JoinHandle<()>>,
) -> WalleResult<()>
where
E: ProtocolItem + GetSelf + Clone,
AH: ActionHandler<E, A, R> + Send + Sync + 'static,
EH: EventHandler<E, A, R> + Send + Sync + 'static,
AH: ActionHandler<Event, A, R> + Send + Sync + 'static,
EH: EventHandler<Event, A, R> + Send + Sync + 'static,
{
for wss in config {
let addr = std::net::SocketAddr::new(wss.host, wss.port);
Expand Down Expand Up @@ -123,18 +121,17 @@ where
}
}

async fn ws_loop<E, A, R, AH, EH>(
async fn ws_loop<A, R, AH, EH>(
ob: Arc<OneBot<AH, EH>>,
mut ws_stream: WebSocketStream<TcpStream>,
echo_map: EchoMap<R>,
bot_map: Arc<BotMap<A>>,
// implt: String,
) where
E: ProtocolItem + GetSelf + Clone,
A: ProtocolItem,
R: ProtocolItem,
AH: ActionHandler<E, A, R> + Send + Sync + 'static,
EH: EventHandler<E, A, R> + Send + Sync + 'static,
AH: ActionHandler<Event, A, R> + Send + Sync + 'static,
EH: EventHandler<Event, A, R> + Send + Sync + 'static,
{
let (seq, mut action_rx) = bot_map.new_connect();
let mut signal_rx = ob.get_signal_rx().unwrap(); //todo
Expand Down Expand Up @@ -171,7 +168,7 @@ async fn ws_loop<E, A, R, AH, EH>(
bot_map.connect_closs(&seq);
}

async fn ws_recv<E, A, R, AH, EH>(
async fn ws_recv<A, R, AH, EH>(
msg: WsMsg,
ob: &Arc<OneBot<AH, EH>>,
ws_stream: &mut WebSocketStream<TcpStream>,
Expand All @@ -181,20 +178,19 @@ async fn ws_recv<E, A, R, AH, EH>(
implt: &mut Option<String>,
) -> bool
where
E: ProtocolItem + Clone + GetSelf,
A: ProtocolItem,
R: ProtocolItem,
AH: ActionHandler<E, A, R> + Send + Sync + 'static,
EH: EventHandler<E, A, R> + Send + Sync + 'static,
AH: ActionHandler<Event, A, R> + Send + Sync + 'static,
EH: EventHandler<Event, A, R> + Send + Sync + 'static,
{
#[derive(Debug, Deserialize, Serialize)]
#[serde(untagged)]
enum ReceiveItem<E, R> {
Event(E),
enum ReceiveItem<R> {
Event(Event),
Resp(Echo<R>),
}

let handle_ok = |item: Result<ReceiveItem<E, R>, eyre::Report>| async move {
let handle_ok = |item: Result<ReceiveItem<R>, eyre::Report>| async move {
match item {
Ok(ReceiveItem::Event(event)) => {
let ob = ob.clone();
Expand All @@ -210,7 +206,7 @@ where
}
};

let event_process = |meta: Result<ComEvent, eyre::Report>| async move {
let meta_process = |meta: Result<ComEvent, eyre::Report>| async move {
if let Ok(event) = meta.and_then(|e: ComEvent| {
let e = e.to_v12();
<MetaDetailEvent as TryFrom<Event>>::try_from(e).map_err(|e| e.into())
Expand All @@ -229,11 +225,11 @@ where

match msg {
WsMsg::Text(text) => {
event_process(ProtocolItem::json_decode(&text)).await;
meta_process(ProtocolItem::json_decode(&text)).await;
handle_ok(ProtocolItem::json_decode(&text)).await;
}
WsMsg::Binary(b) => {
event_process(ProtocolItem::rmp_decode(&b)).await;
meta_process(ProtocolItem::rmp_decode(&b)).await;
handle_ok(ProtocolItem::rmp_decode(&b)).await;
}
WsMsg::Ping(b) => {
Expand Down
16 changes: 8 additions & 8 deletions src/obc/app_obc/mod.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@ use std::sync::Arc;

use super::OBC;
use crate::ah::GetSelfs;
use crate::event::Event;
use crate::prelude::{Bot, Version};
use crate::structs::Selft;
use crate::util::{Echo, EchoInner, EchoS, GetSelf, ProtocolItem};
Expand Down Expand Up @@ -65,9 +66,8 @@ impl<A, R> AppOBC<A, R> {
}
}

impl<E, A, R> ActionHandler<E, A, R> for AppOBC<A, R>
impl<A, R> ActionHandler<Event, A, R> for AppOBC<A, R>
where
E: ProtocolItem + Clone + GetSelf,
A: ProtocolItem + GetSelf,
R: ProtocolItem,
{
Expand All @@ -78,8 +78,8 @@ where
config: crate::config::AppConfig,
) -> WalleResult<Vec<JoinHandle<()>>>
where
AH: ActionHandler<E, A, R> + Send + Sync + 'static,
EH: EventHandler<E, A, R> + Send + Sync + 'static,
AH: ActionHandler<Event, A, R> + Send + Sync + 'static,
EH: EventHandler<Event, A, R> + Send + Sync + 'static,
{
let mut tasks = vec![];
#[cfg(feature = "websocket")]
Expand All @@ -96,8 +96,8 @@ where
}
async fn call<AH, EH>(&self, action: A, _ob: &Arc<OneBot<AH, EH>>) -> WalleResult<R>
where
AH: ActionHandler<E, A, R> + Send + Sync + 'static,
EH: EventHandler<E, A, R> + Send + Sync + 'static,
AH: ActionHandler<Event, A, R> + Send + Sync + 'static,
EH: EventHandler<Event, A, R> + Send + Sync + 'static,
{
match self.bots.get_bot(&action.get_self()) {
Some(action_txs) => {
Expand Down Expand Up @@ -132,9 +132,9 @@ where
}
async fn before_call_event<AH, EH>(
&self,
event: E,
event: Event,
_ob: &Arc<OneBot<AH, EH>>,
) -> WalleResult<E> {
) -> WalleResult<Event> {
if self._block_meta_event.load(Ordering::Relaxed) {
use core::any::Any;
let event: Box<dyn Any> = Box::new(event.clone());
Expand Down

0 comments on commit f29b7d9

Please sign in to comment.