socket-bot/ica-rs/src/ica.rs

pub mod client;
pub mod events;

use futures_util::FutureExt;
use rust_socketio::asynchronous::{Client, ClientBuilder};
use rust_socketio::{Event, Payload, TransportType};
use tracing::{event, span, Level};

use crate::config::IcaConfig;
use crate::error::{ClientResult, IcaError};
use crate::{wrap_any_callback, wrap_callback, StopGetter};

const ICA_PROTOCOL_VERSION: &str = "2.12.2";

pub async fn start_ica(config: &IcaConfig, stop_reciver: StopGetter) -> ClientResult<(), IcaError> {
    let span = span!(Level::INFO, "Icalingua Client");
    let _enter = span.enter();

    event!(Level::INFO, "ica-async-rs v{} initing", crate::ICA_VERSION);

    let socket = match ClientBuilder::new(config.host.clone())
        .transport_type(TransportType::Websocket)
        .on_any(wrap_any_callback!(events::any_event))
        .on("requireAuth", wrap_callback!(client::sign_callback))
        .on("message", wrap_callback!(events::connect_callback))
        .on("authSucceed", wrap_callback!(events::connect_callback))
        .on("authFailed", wrap_callback!(events::connect_callback))
        .on("messageSuccess", wrap_callback!(events::succes_message))
        .on("messageFailed", wrap_callback!(events::failed_message))
        .on("onlineData", wrap_callback!(events::get_online_data))
        .on("setAllRooms", wrap_callback!(events::update_all_room))
        .on("setMessages", wrap_callback!(events::set_messages))
        .on("addMessage", wrap_callback!(events::add_message))
        .on("deleteMessage", wrap_callback!(events::delete_message))
        .connect()
        .await
    {
        Ok(client) => {
            event!(Level::INFO, "socketio connected");
            client
        }
        Err(e) => {
            event!(Level::ERROR, "socketio connect failed: {}", e);
            return Err(IcaError::SocketIoError(e));
        }
    };

    if config.notice_start {
        for room in config.notice_room.iter() {
            let startup_msg = crate::data_struct::ica::messages::SendMessage::new(
                format!("shenbot v {}\nica-async-rs v{}", crate::VERSION, crate::ICA_VERSION),
                *room,
                None,
            );
            tokio::time::sleep(std::time::Duration::from_secs(1)).await;

            event!(Level::INFO, "发送启动消息到房间: {}", room);

            if let Err(e) =
                socket.emit("sendMessage", serde_json::to_value(startup_msg).unwrap()).await
            {
                event!(Level::INFO, "启动信息发送失败 房间:{}|e:{}", room, e);
            }
        }
    }
    // 等待停止信号
    stop_reciver.await.ok();
    event!(Level::INFO, "socketio client stopping");
    match socket.disconnect().await {
        Ok(_) => {
            event!(Level::INFO, "socketio client stopped");
            Ok(())
        }
        Err(e) => {
            // 单独处理 SocketIoError(IncompleteResponseFromEngineIo(WebsocketError(AlreadyClosed)))
            match e {
                rust_socketio::Error::IncompleteResponseFromEngineIo(inner_e) => {
                    if inner_e.to_string().contains("AlreadyClosed") {
                        event!(Level::INFO, "socketio client stopped");
                        return Ok(());
                    } else {
                        event!(Level::ERROR, "socketio client stopped with error: {:?}", inner_e);
                        Err(IcaError::SocketIoError(
                            rust_socketio::Error::IncompleteResponseFromEngineIo(inner_e),
                        ))
                    }
                }
                e => {
                    event!(Level::ERROR, "socketio client stopped with error: {}", e);
                    Err(IcaError::SocketIoError(e))
                }
            }
        }
    }
}