soup_sdk/chat/
message.rs

1use std::{collections::HashMap, sync::Mutex};
2
3use tokio::sync::{broadcast, mpsc};
4
5use crate::{
6    Error, Result,
7    chat::{
8        Event,
9        commands::{Command, MessageType},
10        constants::message_codes::{self},
11        events::{
12            DecodedFrameDiagnostic, DiagnosticEvent, EventMeta, ParseFailedDiagnostic,
13            ParticipantListCompleteEvent, ParticipantListEvent, RawFrameDiagnostic,
14            UnknownCodeDiagnostic,
15        },
16        formatter::ChatFormatter,
17        parser::{
18            balloon::{
19                parse_ad_balloon_event, parse_balloon_event, parse_balloon_sub_event,
20                parse_station_ad_balloon_event, parse_video_balloon_event,
21                parse_vod_ad_balloon_event, parse_vod_balloon_event,
22            },
23            chat::{parse_chat_event, parse_manager_chat_event},
24            emoticon::parse_emoticon_event,
25            exit::parse_exit_event,
26            freeze::parse_freeze_event,
27            gift::{parse_ogq_gift_event, parse_quickview_gift_event, parse_subscribe_gift_event},
28            kick::parse_kick_cancel_event,
29            mission::{ParsedMissionEvent, parse_mission_event},
30            mute::parse_mute_event,
31            notification::parse_notification_event,
32            participant::{
33                ParticipantListChunk, parse_participant_list_chunk,
34                parse_participant_list_complete_event,
35            },
36            raw::{ParseError, RawMessage, parse_message},
37            slow::parse_slow_event,
38            sticker::{parse_sticker_event, parse_sticker_sub_event},
39            subscribe::{parse_subscribe_event, parse_subscribe_renew_event},
40        },
41        types::User,
42    },
43};
44
45pub struct MessageHandler {
46    pub formatter: ChatFormatter,
47    pub event_tx: mpsc::Sender<Event>,
48    pub command_tx: mpsc::Sender<Command>,
49    pub diagnostics: bool,
50    participant_list: Mutex<ParticipantListAccumulator>,
51}
52
53impl MessageHandler {
54    pub fn new(
55        formatter: &ChatFormatter,
56        event_tx: mpsc::Sender<Event>,
57        command_tx: mpsc::Sender<Command>,
58        diagnostics: bool,
59    ) -> Self {
60        Self {
61            formatter: formatter.clone(),
62            event_tx,
63            command_tx,
64            diagnostics,
65            participant_list: Mutex::new(ParticipantListAccumulator::default()),
66        }
67    }
68    /// 메시지를 처리하고 이벤트를 전송합니다.
69    pub fn handle(&self, raw: Vec<u8>) -> Result<Option<Vec<u8>>> {
70        let raw_arc: std::sync::Arc<[u8]> = raw.into();
71        self.broadcast_diagnostic(DiagnosticEvent::RawFrame(RawFrameDiagnostic {
72            meta: EventMeta {
73                received_time: chrono::Utc::now(),
74            },
75            byte_len: raw_arc.len(),
76            raw: raw_arc.clone(),
77        }))?;
78
79        match parse_message(raw_arc.clone()) {
80            Ok(message) => {
81                self.broadcast_diagnostic(DiagnosticEvent::DecodedFrame(DecodedFrameDiagnostic {
82                    meta: EventMeta {
83                        received_time: message.received_time,
84                    },
85                    code: message.code,
86                    ret_code: message.ret_code,
87                    field_count: message.body.len(),
88                    body: message.body.clone(),
89                    raw: message.raw.clone(),
90                }))?;
91                Ok(self.handle_message(message)?)
92            }
93            Err(error) => {
94                self.broadcast_parse_failed(error, raw_arc, None)?;
95                Ok(None)
96            }
97        }
98    }
99
100    fn broadcast(&self, event: Event) -> Result<()> {
101        self.event_tx
102            .try_send(event)
103            .map_err(|e| Error::ChannelClosed(e.to_string()))?;
104        Ok(())
105    }
106
107    fn broadcast_diagnostic(&self, event: DiagnosticEvent) -> Result<()> {
108        if self.diagnostics {
109            self.broadcast(Event::Diagnostic(event))?;
110        }
111        Ok(())
112    }
113
114    fn handle_message(&self, message: RawMessage) -> Result<Option<Vec<u8>>> {
115        // 메시지 처리 로직을 여기에 구현합니다.
116        // 예를 들어, raw 메시지를 파싱하고 필요한 이벤트를 생성할 수 있습니다.
117        let message_for_error = message.clone();
118        let res = match message.code {
119            message_codes::CONNECT => self.handle_connect(message),
120            message_codes::CHAT => self.handle_chat(message),
121            message_codes::EXIT => self.handle_participant_list_or_exit(message),
122            message_codes::PARTICIPANT_FOLLOW_STATE_LIST => Ok(None),
123            message_codes::PARTICIPANT_LIST_COMPLETE => {
124                self.handle_participant_list_complete(message)
125            }
126            message_codes::FREEZE => self.handle_freeze(message),
127            message_codes::MUTE => self.handle_mute(message),
128            message_codes::MANAGER_CHAT => self.handle_manager_message(message),
129            message_codes::EMOTICON => self.handle_emoticon_message(message),
130            message_codes::NOTIFICATION => self.handle_notification(message),
131            message_codes::BJ_STATE_CHANGE => self.handle_bj_state_change(message),
132            message_codes::SLOW => self.handle_slow(message),
133            message_codes::KICK_CANCEL => self.handle_kick_cancel(message),
134            message_codes::SUBSCRIBE => self.handle_subscribe(message),
135            message_codes::SUBSCRIBE_RENEW => self.handle_subscribe_renew(message),
136            // 스티커
137            message_codes::STICKER | message_codes::SUB_STICKER => self.handle_sticker(message),
138            // 선물
139            message_codes::OGQ_GIFTED
140            | message_codes::QUICKVIEW_GIFTED
141            | message_codes::SUBSCRIPTION_GIFTED => self.handle_gift(message),
142            // 미션
143            message_codes::MISSION_DONATION => self.handle_mission(message),
144            // 도네이션
145            message_codes::DONATION
146            | message_codes::ADBALLOON_DONATION
147            | message_codes::SUB_DONATION
148            | message_codes::VOD_AD_DONATION
149            | message_codes::VOD_DONATION
150            | message_codes::AD_STATION_DONATION
151            | message_codes::VIDEO_DONATION => self.handle_donation(message),
152
153            _ => {
154                // 다른 메시지 코드 처리
155                self.broadcast_unknown(&message);
156                Ok(None)
157            }
158        };
159
160        // 메시지에 대한 응답이 필요한 경우, Vec<u8>를 반환합니다.
161        match res {
162            Ok(response) => Ok(response),
163            Err(error) => {
164                self.broadcast_parse_failed(
165                    error,
166                    message_for_error.raw.clone(),
167                    Some(&message_for_error),
168                )?;
169                Ok(None)
170            }
171        }
172    }
173
174    fn handle_donation(
175        &self,
176        message: RawMessage,
177    ) -> std::result::Result<Option<Vec<u8>>, ParseError> {
178        let e = match message.code {
179            message_codes::DONATION => parse_balloon_event(message),
180            message_codes::SUB_DONATION => parse_balloon_sub_event(message),
181            message_codes::VOD_DONATION => parse_vod_balloon_event(message),
182            message_codes::VOD_AD_DONATION => parse_vod_ad_balloon_event(message),
183            message_codes::ADBALLOON_DONATION => parse_ad_balloon_event(message),
184            message_codes::AD_STATION_DONATION => parse_station_ad_balloon_event(message),
185            message_codes::VIDEO_DONATION => parse_video_balloon_event(message),
186            _ => return Ok(None),
187        }?;
188        let _ = self.broadcast(Event::Donation(e));
189        Ok(None)
190    }
191
192    fn handle_sticker(
193        &self,
194        message: RawMessage,
195    ) -> std::result::Result<Option<Vec<u8>>, ParseError> {
196        let e = match message.code {
197            message_codes::STICKER => parse_sticker_event(message),
198            message_codes::SUB_STICKER => parse_sticker_sub_event(message),
199            _ => return Ok(None),
200        }?;
201        let _ = self.broadcast(Event::Sticker(e));
202        Ok(None)
203    }
204
205    fn handle_gift(&self, message: RawMessage) -> std::result::Result<Option<Vec<u8>>, ParseError> {
206        let e = match message.code {
207            message_codes::OGQ_GIFTED => parse_ogq_gift_event(message),
208            message_codes::SUBSCRIPTION_GIFTED => parse_subscribe_gift_event(message),
209            message_codes::QUICKVIEW_GIFTED => parse_quickview_gift_event(message),
210            _ => return Ok(None),
211        }?;
212        let _ = self.broadcast(Event::Gift(e));
213        Ok(None)
214    }
215
216    fn handle_mission(
217        &self,
218        message: RawMessage,
219    ) -> std::result::Result<Option<Vec<u8>>, ParseError> {
220        match parse_mission_event(message)? {
221            ParsedMissionEvent::Mission(e) => {
222                let _ = self.broadcast(Event::MissionDonation(e));
223            }
224            ParsedMissionEvent::MissionTotal(e) => {
225                let _ = self.broadcast(Event::MissionTotal(e));
226            }
227            ParsedMissionEvent::BattleNotice(e) => {
228                let _ = self.broadcast(Event::BattleMissionResult(e));
229            }
230            ParsedMissionEvent::ChallengeNotice(e) => {
231                let _ = self.broadcast(Event::ChallengeMissionResult(e));
232            }
233        }
234
235        Ok(None)
236    }
237
238    fn handle_subscribe_renew(
239        &self,
240        message: RawMessage,
241    ) -> std::result::Result<Option<Vec<u8>>, ParseError> {
242        let _ = self.broadcast(Event::Subscribe(parse_subscribe_renew_event(message)?));
243        Ok(None)
244    }
245
246    fn handle_subscribe(
247        &self,
248        message: RawMessage,
249    ) -> std::result::Result<Option<Vec<u8>>, ParseError> {
250        let _ = self.broadcast(Event::Subscribe(parse_subscribe_event(message)?));
251        Ok(None)
252    }
253
254    fn handle_slow(&self, message: RawMessage) -> std::result::Result<Option<Vec<u8>>, ParseError> {
255        let _ = self.broadcast(Event::Slow(parse_slow_event(message)?));
256        Ok(None)
257    }
258
259    fn handle_bj_state_change(
260        &self,
261        _: RawMessage,
262    ) -> std::result::Result<Option<Vec<u8>>, ParseError> {
263        let _ = self.broadcast(Event::BJStateChange);
264        Ok(None)
265    }
266
267    fn handle_emoticon_message(
268        &self,
269        message: RawMessage,
270    ) -> std::result::Result<Option<Vec<u8>>, ParseError> {
271        let _ = self.broadcast(Event::Chat(parse_emoticon_event(message)?));
272        Ok(None)
273    }
274
275    fn handle_notification(
276        &self,
277        message: RawMessage,
278    ) -> std::result::Result<Option<Vec<u8>>, ParseError> {
279        let _ = self.broadcast(Event::Notification(parse_notification_event(message)?));
280        Ok(None)
281    }
282
283    fn handle_manager_message(
284        &self,
285        message: RawMessage,
286    ) -> std::result::Result<Option<Vec<u8>>, ParseError> {
287        let _ = self.broadcast(Event::Chat(parse_manager_chat_event(message)?));
288        Ok(None)
289    }
290
291    fn handle_kick_cancel(
292        &self,
293        message: RawMessage,
294    ) -> std::result::Result<Option<Vec<u8>>, ParseError> {
295        if let Some(e) = parse_kick_cancel_event(message)? {
296            let _ = self.broadcast(Event::KickCancel(e));
297        }
298        Ok(None)
299    }
300
301    fn handle_participant_list_or_exit(
302        &self,
303        message: RawMessage,
304    ) -> std::result::Result<Option<Vec<u8>>, ParseError> {
305        if let Some(chunk) = parse_participant_list_chunk(message.clone())? {
306            self.with_participant_list(|list| list.push_participant_chunk(chunk));
307            return Ok(None);
308        }
309
310        self.handle_exit(message)
311    }
312
313    fn handle_participant_list_complete(
314        &self,
315        message: RawMessage,
316    ) -> std::result::Result<Option<Vec<u8>>, ParseError> {
317        let complete = parse_participant_list_complete_event(message)?;
318        let event = self.with_participant_list(|list| list.finish(complete));
319        let _ = self.broadcast(Event::ParticipantList(event));
320        Ok(None)
321    }
322
323    fn handle_mute(&self, message: RawMessage) -> std::result::Result<Option<Vec<u8>>, ParseError> {
324        let _ = self.broadcast(Event::Mute(parse_mute_event(message)?));
325        Ok(None)
326    }
327
328    fn handle_freeze(
329        &self,
330        message: RawMessage,
331    ) -> std::result::Result<Option<Vec<u8>>, ParseError> {
332        let _ = self.broadcast(Event::Freeze(parse_freeze_event(message)?));
333        Ok(None)
334    }
335
336    fn handle_chat(&self, message: RawMessage) -> std::result::Result<Option<Vec<u8>>, ParseError> {
337        let _ = self.broadcast(Event::Chat(parse_chat_event(message)?));
338        Ok(None)
339    }
340
341    fn handle_exit(&self, message: RawMessage) -> std::result::Result<Option<Vec<u8>>, ParseError> {
342        if let Some((is_kick, e)) = parse_exit_event(message)? {
343            if is_kick {
344                let _ = self.broadcast(Event::Kick(e));
345            } else {
346                let _ = self.broadcast(Event::Exit(e));
347            };
348        }
349        Ok(None)
350    }
351
352    // CONNECT 메시지 처리 -> JOIN 메시지 전송
353    fn handle_connect(&self, _: RawMessage) -> std::result::Result<Option<Vec<u8>>, ParseError> {
354        let ret = self.formatter.format_message(MessageType::JOIN);
355        Ok(Some(ret))
356    }
357
358    fn broadcast_unknown(&self, message: &RawMessage) {
359        let _ = self.broadcast(Event::Unknown(message.code));
360        let _ = self.broadcast_diagnostic(DiagnosticEvent::UnknownCode(UnknownCodeDiagnostic {
361            meta: EventMeta {
362                received_time: message.received_time,
363            },
364            code: message.code,
365            ret_code: message.ret_code,
366            field_count: message.body.len(),
367            body: message.body.clone(),
368            raw: message.raw.clone(),
369        }));
370    }
371
372    fn broadcast_parse_failed(
373        &self,
374        error: ParseError,
375        raw: std::sync::Arc<[u8]>,
376        message: Option<&RawMessage>,
377    ) -> Result<()> {
378        let (field_index, expected_type) = match &error {
379            ParseError::MissingField { index, .. } => (Some(*index), None),
380            ParseError::InvalidNumber { index, target, .. } => {
381                (Some(*index), Some((*target).to_string()))
382            }
383            ParseError::InvalidJson { index, .. } => (Some(*index), Some("json".to_string())),
384            _ => (None, None),
385        };
386
387        let diagnostic = ParseFailedDiagnostic {
388            meta: EventMeta {
389                received_time: chrono::Utc::now(),
390            },
391            code: message.map(|message| message.code),
392            body_field_count: message.map(|message| message.body.len()),
393            field_index,
394            expected_type,
395            reason: error.to_string(),
396            raw,
397        };
398
399        self.broadcast_diagnostic(DiagnosticEvent::ParseFailed(diagnostic))
400    }
401
402    fn with_participant_list<T>(&self, f: impl FnOnce(&mut ParticipantListAccumulator) -> T) -> T {
403        let mut participant_list = self
404            .participant_list
405            .lock()
406            .unwrap_or_else(|poisoned| poisoned.into_inner());
407        f(&mut participant_list)
408    }
409}
410
411#[derive(Default)]
412struct ParticipantListAccumulator {
413    entries: Vec<User>,
414    entry_indexes: HashMap<String, usize>,
415}
416
417impl ParticipantListAccumulator {
418    fn push_participant_chunk(&mut self, chunk: ParticipantListChunk) {
419        for participant in chunk.entries {
420            if self.entry_indexes.contains_key(&participant.id) {
421                continue;
422            }
423
424            self.entry_indexes
425                .insert(participant.id.clone(), self.entries.len());
426            self.entries.push(participant);
427        }
428    }
429
430    fn finish(&mut self, complete: ParticipantListCompleteEvent) -> ParticipantListEvent {
431        let entries = std::mem::take(&mut self.entries);
432        self.entry_indexes.clear();
433
434        ParticipantListEvent {
435            meta: complete.meta,
436            result_code: complete.result_code,
437            count: entries.len(),
438            entries,
439        }
440    }
441}
442
443#[cfg(test)]
444mod tests {
445    use tokio::sync::{broadcast, mpsc};
446
447    use super::*;
448    use crate::{chat::constants::message_codes, models::LiveDetail};
449
450    fn frame(code: u32, body: &[u8]) -> Vec<u8> {
451        let mut bytes = Vec::new();
452        bytes.extend_from_slice(&[27, 9]);
453        bytes.extend_from_slice(format!("{code:04}").as_bytes());
454        bytes.extend_from_slice(format!("{:06}", body.len()).as_bytes());
455        bytes.extend_from_slice(b"00");
456        bytes.extend_from_slice(body);
457        bytes
458    }
459
460    fn handler_with_diagnostics(diagnostics: bool) -> (MessageHandler, broadcast::Receiver<Event>) {
461        let (event_tx, event_rx) = broadcast::channel(16);
462        let (command_tx, _) = mpsc::channel(4);
463        let formatter = ChatFormatter::new(
464            LiveDetail {
465                is_live: true,
466                ch_domain: "chat.example".to_string(),
467                ch_pt: 443,
468                bno: 123,
469                ch_no: "123".to_string(),
470                streamer_nick: "streamer".to_string(),
471                title: "title".to_string(),
472                categories: vec![],
473            },
474            String::new(),
475        );
476
477        (
478            MessageHandler::new(&formatter, event_tx, command_tx, diagnostics),
479            event_rx,
480        )
481    }
482
483    fn handler() -> (MessageHandler, broadcast::Receiver<Event>) {
484        handler_with_diagnostics(true)
485    }
486
487    #[test]
488    fn emits_unknown_code_diagnostic() {
489        let (handler, mut events) = handler();
490
491        handler.handle(frame(9999, b"\x0cnew")).unwrap();
492
493        assert!(matches!(
494            events.try_recv().unwrap(),
495            Event::Diagnostic(DiagnosticEvent::RawFrame(_))
496        ));
497        assert!(matches!(
498            events.try_recv().unwrap(),
499            Event::Diagnostic(DiagnosticEvent::DecodedFrame(_))
500        ));
501        assert!(matches!(events.try_recv().unwrap(), Event::Unknown(9999)));
502        assert!(matches!(
503            events.try_recv().unwrap(),
504            Event::Diagnostic(DiagnosticEvent::UnknownCode(_))
505        ));
506    }
507
508    #[test]
509    fn emits_parse_failed_diagnostic_for_malformed_chat() {
510        let (handler, mut events) = handler();
511
512        handler
513            .handle(frame(message_codes::CHAT, b"\x0conly-comment"))
514            .unwrap();
515
516        let mut saw_parse_failed = false;
517        while let Ok(event) = events.try_recv() {
518            if let Event::Diagnostic(DiagnosticEvent::ParseFailed(diagnostic)) = event {
519                saw_parse_failed = true;
520                assert_eq!(diagnostic.code, Some(message_codes::CHAT));
521                assert_eq!(diagnostic.field_index, Some(1));
522            }
523        }
524
525        assert!(saw_parse_failed);
526    }
527
528    #[test]
529    fn suppresses_diagnostic_events_when_disabled() {
530        let (handler, mut events) = handler_with_diagnostics(false);
531
532        handler.handle(frame(9999, b"\x0cnew")).unwrap();
533
534        assert!(matches!(events.try_recv().unwrap(), Event::Unknown(9999)));
535        assert!(events.try_recv().is_err());
536    }
537
538    #[test]
539    fn aggregates_participant_list_until_complete() {
540        let (handler, mut events) = handler_with_diagnostics(false);
541
542        handler
543            .handle(frame(
544                message_codes::EXIT,
545                b"\x0c1\x0cwoalsdl1234\x0c\xec\x9d\xb4\xed\x8c\x8c\xeb\xa6\xac\xec\x95\x84\x0c537477152|32768\x0cksch1028(3)\x0cuser\x0c268550176|557056\x0c",
546            ))
547            .unwrap();
548
549        assert!(events.try_recv().is_err());
550
551        handler
552            .handle(frame(
553                message_codes::PARTICIPANT_FOLLOW_STATE_LIST,
554                b"\x0cwoalsdl1234\x0cfw=-1&afw=-1\x0cksch1028(3)\x0cfw=15&afw=15\x0c",
555            ))
556            .unwrap();
557
558        assert!(events.try_recv().is_err());
559
560        handler
561            .handle(b"\x1b\t011000000300\x0c1\x0c".to_vec())
562            .unwrap();
563
564        match events.try_recv().unwrap() {
565            Event::ParticipantList(event) => {
566                assert_eq!(event.result_code, "1");
567                assert_eq!(event.count, 2);
568                assert_eq!(event.entries.len(), 2);
569                assert_eq!(event.entries[0].id, "woalsdl1234");
570                assert!(event.entries[0].subscribe.is_none());
571                assert_eq!(event.entries[1].id, "ksch1028");
572                assert!(event.entries[1].subscribe.is_none());
573            }
574            other => panic!("unexpected event: {other:?}"),
575        }
576    }
577
578    #[test]
579    fn keeps_existing_exit_handling_for_exit_shape() {
580        let (handler, mut events) = handler_with_diagnostics(false);
581
582        handler
583            .handle(frame(
584                message_codes::EXIT,
585                b"\x0c0\x0cuser1\x0cnick\x0c1\x0cunused\x0c65536|163840",
586            ))
587            .unwrap();
588
589        assert!(matches!(events.try_recv().unwrap(), Event::Exit(_)));
590    }
591
592    #[test]
593    fn discards_participant_follow_state_for_code_127() {
594        let (handler, mut events) = handler_with_diagnostics(false);
595
596        handler
597            .handle(frame(
598                message_codes::PARTICIPANT_FOLLOW_STATE_LIST,
599                b"\x0cwoalsdl1234\x0cfw=-1&afw=-1\x0cksch1028(3)\x0cfw=15&afw=15\x0c",
600            ))
601            .unwrap();
602
603        assert!(events.try_recv().is_err());
604    }
605
606    #[test]
607    fn discards_non_list_code_127() {
608        let (handler, mut events) = handler_with_diagnostics(false);
609
610        handler
611            .handle(frame(
612                message_codes::PARTICIPANT_FOLLOW_STATE_LIST,
613                b"\x0cuser1\x0cignored\x0cignored",
614            ))
615            .unwrap();
616
617        assert!(events.try_recv().is_err());
618    }
619
620    #[test]
621    fn emits_empty_participant_list_on_complete_without_chunks() {
622        let (handler, mut events) = handler_with_diagnostics(false);
623
624        handler
625            .handle(b"\x1b\t011000000300\x0c1\x0c".to_vec())
626            .unwrap();
627
628        match events.try_recv().unwrap() {
629            Event::ParticipantList(event) => {
630                assert_eq!(event.result_code, "1");
631                assert_eq!(event.count, 0);
632                assert!(event.entries.is_empty());
633            }
634            other => panic!("unexpected event: {other:?}"),
635        }
636    }
637}
soup_sdk/chat/message.rs

soup_sdk/chat/
message.rs
