update ricq-axum-api example (#85)

* update default client processor

* update default client processor

examples/ricq-axum-api/Cargo.toml

@@ -25,3 +25,5 @@ tracing = "0.1"
 base64 = "0.13"
 tower = { version = "0.4" }
 tower-http = { version = "0.3", features = ["fs"] }
+tracing-subscriber = { version = "0.3", features = ["fmt", "local-time"] }
+time = { version = "0.3", features = ["macros", "local-offset"] }

examples/ricq-axum-api/src/bin/main.rs

@@ -3,25 +3,147 @@
 use std::net::SocketAddr;
 use std::str::FromStr;
 use std::sync::Arc;
+use std::time::Duration;
 
 use axum::{
     routing::{get, get_service, post},
     Extension, Router,
 };
 use dashmap::DashMap;
+use tokio::sync::broadcast;
+use tokio::task::JoinHandle;
 use tower::ServiceBuilder;
 use tower_http::services::ServeDir;
+use tracing::Level;
+use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};
 
+use ricq::client::event::{FriendMessageEvent, GroupMessageEvent};
+use ricq::client::{DefaultConnector, NetworkStatus};
+use ricq::ext::common::after_login;
+use ricq::ext::reconnect::{auto_reconnect, Credential};
+use ricq::handler::QEvent;
 use ricq::Client;
 use ricq_axum_api::handler::{bot, password, qrcode};
-use ricq_axum_api::RicqAxumApi;
+use ricq_axum_api::processor::Processor;
+use ricq_axum_api::u8_protocol::U8Protocol;
+use ricq_axum_api::{ClientInfo, RicqAxumApi};
 
-type ClientProcessor = DashMap<(i64, u8), Arc<Client>>;
+// 默认处理器
+struct ClientProcessor(DashMap<(i64, u8), Arc<Client>>);
+
+#[async_trait::async_trait]
+impl Processor for ClientProcessor {
+    async fn on_login_success(
+        &self,
+        client: Arc<Client>,
+        mut event_receiver: broadcast::Receiver<QEvent>,
+        credential: Credential,
+        network_join_handle: JoinHandle<()>,
+    ) {
+        let uin = client.uin().await;
+        let protocol = client.version().await.protocol.to_u8();
+        self.0.insert((uin, protocol), client.clone());
+        after_login(&client).await;
+
+        tokio::spawn(async move {
+            while let Ok(event) = event_receiver.recv().await {
+                match event {
+                    QEvent::GroupMessage(e) => {
+                        let GroupMessageEvent {
+                            inner: message,
+                            client,
+                        } = e;
+                        tracing::info!(
+                            "GROUP_MSG, code: {}, content: {}",
+                            message.group_code,
+                            message.elements.to_string()
+                        );
+                        client
+                            .send_group_message(message.group_code, message.elements)
+                            .await
+                            .ok();
+                    }
+                    QEvent::FriendMessage(e) => {
+                        let FriendMessageEvent {
+                            inner: message,
+                            client,
+                        } = e;
+                        tracing::info!(
+                            "FRIEND_MSG, code: {}, content: {}",
+                            message.from_uin,
+                            message.elements.to_string()
+                        );
+                        client
+                            .send_friend_message(message.from_uin, message.elements)
+                            .await
+                            .ok();
+                    }
+                    other => {
+                        tracing::info!("{:?}", other)
+                    }
+                }
+            }
+        });
+
+        // DONT BLOCK
+        tokio::spawn(async move {
+            network_join_handle.await.ok();
+            auto_reconnect(
+                client,
+                credential,
+                Duration::from_secs(10),
+                10,
+                DefaultConnector,
+            )
+            .await;
+        });
+    }
+
+    async fn list_client(&self) -> Vec<ClientInfo> {
+        let mut infos = Vec::new();
+        for cli in self.0.iter() {
+            let (uin, protocol) = cli.key();
+            let client = cli.value();
+            infos.push(ClientInfo {
+                uin: *uin,
+                nick: client.account_info.read().await.nickname.clone(),
+                status: client.get_status(),
+                protocol: *protocol,
+            });
+        }
+        infos
+    }
+
+    async fn delete_client(&self, uin: i64, protocol: u8) {
+        if let Some((_, client)) = self.0.remove(&(uin, protocol)) {
+            client.stop(NetworkStatus::Stop);
+        }
+    }
+}
 
 #[tokio::main]
 async fn main() {
-    // 默认处理器，登录后什么也不做，仅作为容器
-    let processor = ClientProcessor::default();
+    // 初始化日志
+    tracing_subscriber::registry()
+        .with(
+            tracing_subscriber::fmt::layer()
+                .with_target(true)
+                .with_timer(tracing_subscriber::fmt::time::OffsetTime::new(
+                    time::UtcOffset::__from_hms_unchecked(8, 0, 0),
+                    time::macros::format_description!(
+                        "[year repr:last_two]-[month]-[day] [hour]:[minute]:[second]"
+                    ),
+                )),
+        )
+        .with(
+            tracing_subscriber::filter::Targets::new()
+                .with_target("main", Level::DEBUG)
+                .with_target("ricq", Level::DEBUG)
+                .with_target("ricq_axum_api", Level::DEBUG),
+        )
+        .init();
+
+    let processor = ClientProcessor(Default::default());
     let ricq_axum_api = Arc::new(RicqAxumApi::new(processor));
 
     let app = Router::new()
@@ -32,27 +154,33 @@ async fn main() {
                 .nest(
                     "/qrcode",
                     Router::new()
-                        .route("/create", post(qrcode::create))
-                        .route("/list", get(qrcode::list))
-                        .route("/delete", post(qrcode::delete))
-                        .route("/query", post(qrcode::query)),
+                        .route("/create", post(qrcode::create::<ClientProcessor>))
+                        .route("/list", get(qrcode::list::<ClientProcessor>))
+                        .route("/delete", post(qrcode::delete::<ClientProcessor>))
+                        .route("/query", post(qrcode::query::<ClientProcessor>)),
                 )
                 .nest(
                     "/password",
                     Router::new()
-                        .route("/create", post(password::login))
-                        .route("/request_sms", post(password::request_sms))
-                        .route("/submit_sms", post(password::submit_sms))
-                        .route("/submit_ticket", post(password::submit_ticket))
-                        .route("/list", get(password::list))
-                        .route("/delete", post(password::delete)),
+                        .route("/create", post(password::login::<ClientProcessor>))
+                        .route(
+                            "/request_sms",
+                            post(password::request_sms::<ClientProcessor>),
+                        )
+                        .route("/submit_sms", post(password::submit_sms::<ClientProcessor>))
+                        .route(
+                            "/submit_ticket",
+                            post(password::submit_ticket::<ClientProcessor>),
+                        )
+                        .route("/list", get(password::list::<ClientProcessor>))
+                        .route("/delete", post(password::delete::<ClientProcessor>)),
                 ),
         )
         .nest(
             "/bot",
             Router::new()
-                .route("/list", get(bot::list))
-                .route("/delete", post(bot::delete)),
+                .route("/list", get(bot::list::<ClientProcessor>))
+                .route("/delete", post(bot::delete::<ClientProcessor>)),
         )
         .fallback(get_service(ServeDir::new("static")).handle_error(handle_error))
         .layer(

examples/ricq-axum-api/src/handler/bot.rs

@@ -12,8 +12,8 @@ pub struct ListBotResp {
     pub bots: Vec<ClientInfo>,
 }
 
-pub async fn list(
-    ricq_axum_api: Extension<Arc<RicqAxumApi>>,
+pub async fn list<P: Processor>(
+    ricq_axum_api: Extension<Arc<RicqAxumApi<P>>>,
 ) -> Result<Json<ListBotResp>, StatusCode> {
     Ok(Json(ListBotResp {
         bots: ricq_axum_api.processor.list_client().await,
@@ -29,9 +29,9 @@ pub struct DeleteBotReq {
 #[derive(Debug, Default, Serialize, Deserialize)]
 pub struct DeleteBotResp {}
 
-pub async fn delete(
+pub async fn delete<P: Processor>(
     Json(req): Json<DeleteBotReq>,
-    ricq_axum_api: Extension<Arc<RicqAxumApi>>,
+    ricq_axum_api: Extension<Arc<RicqAxumApi<P>>>,
 ) -> Result<Json<DeleteBotResp>, StatusCode> {
     ricq_axum_api
         .processor

examples/ricq-axum-api/src/handler/password.rs

@@ -94,9 +94,9 @@ impl From<LoginResponse> for PasswordLoginResp {
     }
 }
 
-pub async fn login(
+pub async fn login<P: Processor>(
     Json(req): Json<CreateClientReq>,
-    ricq_axum_api: Extension<Arc<RicqAxumApi>>,
+    ricq_axum_api: Extension<Arc<RicqAxumApi<P>>>,
 ) -> Result<Json<PasswordLoginResp>, StatusCode> {
     let mut rand_seed = req.device_seed.unwrap_or(req.uin as u64);
     if rand_seed == 0 {
@@ -149,9 +149,9 @@ pub async fn login(
     Ok(Json(PasswordLoginResp::from(resp)))
 }
 
-pub async fn submit_ticket(
+pub async fn submit_ticket<P: Processor>(
     Json(req): Json<SubmitTicketReq>,
-    ricq_axum_api: Extension<Arc<RicqAxumApi>>,
+    ricq_axum_api: Extension<Arc<RicqAxumApi<P>>>,
 ) -> Result<Json<PasswordLoginResp>, StatusCode> {
     let mut resp = ricq_axum_api
         .password_clients
@@ -199,9 +199,9 @@ pub async fn submit_ticket(
     Ok(Json(PasswordLoginResp::from(resp)))
 }
 
-pub async fn request_sms(
+pub async fn request_sms<P: Processor>(
     Json(req): Json<RequestSmsReq>,
-    ricq_axum_api: Extension<Arc<RicqAxumApi>>,
+    ricq_axum_api: Extension<Arc<RicqAxumApi<P>>>,
 ) -> Result<Json<PasswordLoginResp>, StatusCode> {
     let resp = ricq_axum_api
         .password_clients
@@ -219,9 +219,9 @@ pub async fn request_sms(
     Ok(Json(PasswordLoginResp::from(resp)))
 }
 
-pub async fn submit_sms(
+pub async fn submit_sms<P: Processor>(
     Json(req): Json<SubmitSmsReq>,
-    ricq_axum_api: Extension<Arc<RicqAxumApi>>,
+    ricq_axum_api: Extension<Arc<RicqAxumApi<P>>>,
 ) -> Result<Json<PasswordLoginResp>, StatusCode> {
     let mut resp = ricq_axum_api
         .password_clients
@@ -281,8 +281,8 @@ pub struct ListClientRespClient {
     pub resp: PasswordLoginResp,
 }
 
-pub async fn list(
-    ricq_axum_api: Extension<Arc<RicqAxumApi>>,
+pub async fn list<P: Processor>(
+    ricq_axum_api: Extension<Arc<RicqAxumApi<P>>>,
 ) -> Result<Json<ListClientResp>, StatusCode> {
     let mut clients = Vec::new();
     for c in ricq_axum_api.password_clients.iter() {
@@ -304,9 +304,9 @@ pub struct DeleteClientReq {
 #[derive(Default, Serialize, Deserialize)]
 pub struct DeleteClientResp {}
 
-pub async fn delete(
+pub async fn delete<P: Processor>(
     Json(req): Json<DeleteClientReq>,
-    ricq_axum_api: Extension<Arc<RicqAxumApi>>,
+    ricq_axum_api: Extension<Arc<RicqAxumApi<P>>>,
 ) -> Result<Json<DeleteClientResp>, StatusCode> {
     if let Some((_, cli)) = ricq_axum_api
         .password_clients

examples/ricq-axum-api/src/handler/qrcode.rs

@@ -53,9 +53,9 @@ pub struct CreateClientResp {
     pub image: Vec<u8>,
 }
 
-pub async fn create(
+pub async fn create<P: Processor>(
     Json(req): Json<CreateClientReq>,
-    ricq_axum_api: Extension<Arc<RicqAxumApi>>,
+    ricq_axum_api: Extension<Arc<RicqAxumApi<P>>>,
 ) -> Result<Json<CreateClientResp>, StatusCode> {
     let rand_seed = req.device_seed.unwrap_or_else(rand::random);
     let device = Device::random_with_rng(&mut StdRng::seed_from_u64(rand_seed));
@@ -111,9 +111,9 @@ pub struct QueryQRCodeResp {
     pub state: String,
 }
 
-pub async fn query(
+pub async fn query<P: Processor>(
     Json(req): Json<QueryQRCodeReq>,
-    ricq_axum_api: Extension<Arc<RicqAxumApi>>,
+    ricq_axum_api: Extension<Arc<RicqAxumApi<P>>>,
 ) -> Result<Json<QueryQRCodeResp>, StatusCode> {
     let sig = Bytes::from(req.sig);
 
@@ -191,8 +191,8 @@ pub struct ListClientRespClient {
     pub state: String,
 }
 
-pub async fn list(
-    ricq_axum_api: Extension<Arc<RicqAxumApi>>,
+pub async fn list<P: Processor>(
+    ricq_axum_api: Extension<Arc<RicqAxumApi<P>>>,
 ) -> Result<Json<ListClientResp>, StatusCode> {
     let mut clients = Vec::new();
     for c in ricq_axum_api.qrcode_clients.iter() {
@@ -223,9 +223,9 @@ pub struct DeleteClientReq {
 #[derive(Default, Serialize, Deserialize)]
 pub struct DeleteClientResp {}
 
-pub async fn delete(
+pub async fn delete<P: Processor>(
     Json(req): Json<DeleteClientReq>,
-    ricq_axum_api: Extension<Arc<RicqAxumApi>>,
+    ricq_axum_api: Extension<Arc<RicqAxumApi<P>>>,
 ) -> Result<Json<DeleteClientResp>, StatusCode> {
     if let Some((_, cli)) = ricq_axum_api.qrcode_clients.remove(&Bytes::from(req.sig)) {
         cli.client.stop(NetworkStatus::Stop);

examples/ricq-axum-api/src/lib.rs

@@ -9,7 +9,7 @@ use tokio::task::JoinHandle;
 
 pub mod handler;
 pub mod processor;
-mod u8_protocol;
+pub mod u8_protocol;
 use serde::{Deserialize, Serialize};
 
 use crate::processor::Processor;
@@ -39,7 +39,7 @@ pub struct QRCodeClient {
     pub network_join_handle: JoinHandle<()>,
 }
 
-pub struct RicqAxumApi<P: Processor = DashMap<(i64, u8), Arc<Client>>> {
+pub struct RicqAxumApi<P: Processor> {
     // key: uin+protocol
     password_clients: DashMap<(i64, u8), PasswordClient>,