use std::{collections::HashMap, sync::Mutex};
use tokio::sync::{broadcast, mpsc};
use crate::{
Error, Result,
chat::{
Event,
commands::{Command, MessageType},
constants::message_codes::{self},
events::{
DecodedFrameDiagnostic, DiagnosticEvent, EventMeta, ParseFailedDiagnostic,
ParticipantListCompleteEvent, ParticipantListEvent, RawFrameDiagnostic,
UnknownCodeDiagnostic,
},
formatter::ChatFormatter,
parser::{
balloon::{
parse_ad_balloon_event, parse_balloon_event, parse_balloon_sub_event,
parse_station_ad_balloon_event, parse_video_balloon_event,
parse_vod_ad_balloon_event, parse_vod_balloon_event,
},
chat::{parse_chat_event, parse_manager_chat_event},
emoticon::parse_emoticon_event,
exit::parse_exit_event,
freeze::parse_freeze_event,
gift::{parse_ogq_gift_event, parse_quickview_gift_event, parse_subscribe_gift_event},
kick::parse_kick_cancel_event,
mission::{ParsedMissionEvent, parse_mission_event},
mute::parse_mute_event,
notification::parse_notification_event,
participant::{
ParticipantListChunk, parse_participant_list_chunk,
parse_participant_list_complete_event,
},
raw::{ParseError, RawMessage, parse_message},
slow::parse_slow_event,
sticker::{parse_sticker_event, parse_sticker_sub_event},
subscribe::{parse_subscribe_event, parse_subscribe_renew_event},
},
types::User,
},
};
pub struct MessageHandler {
pub formatter: ChatFormatter,
pub event_tx: broadcast::Sender<Event>,
pub command_tx: mpsc::Sender<Command>,
pub diagnostics: bool,
participant_list: Mutex<ParticipantListAccumulator>,
}
impl MessageHandler {
pub fn new(
formatter: &ChatFormatter,
event_tx: broadcast::Sender<Event>,
command_tx: mpsc::Sender<Command>,
diagnostics: bool,
) -> Self {
Self {
formatter: formatter.clone(),
event_tx,
command_tx,
diagnostics,
participant_list: Mutex::new(ParticipantListAccumulator::default()),
}
}
pub fn handle(&self, raw: Vec<u8>) -> Result<Option<Vec<u8>>> {
let raw_arc: std::sync::Arc<[u8]> = raw.into();
self.broadcast_diagnostic(DiagnosticEvent::RawFrame(RawFrameDiagnostic {
meta: EventMeta {
received_time: chrono::Utc::now(),
},
byte_len: raw_arc.len(),
raw: raw_arc.clone(),
}))?;
match parse_message(raw_arc.clone()) {
Ok(message) => {
self.broadcast_diagnostic(DiagnosticEvent::DecodedFrame(DecodedFrameDiagnostic {
meta: EventMeta {
received_time: message.received_time,
},
code: message.code,
ret_code: message.ret_code,
field_count: message.body.len(),
body: message.body.clone(),
raw: message.raw.clone(),
}))?;
Ok(self.handle_message(message)?)
}
Err(error) => {
self.broadcast_parse_failed(error, raw_arc, None)?;
Ok(None)
}
}
}
fn broadcast(&self, event: Event) -> Result<()> {
self.event_tx
.send(event)
.map_err(|e| Error::ChannelClosed(e.to_string()))?;
Ok(())
}
fn broadcast_diagnostic(&self, event: DiagnosticEvent) -> Result<()> {
if self.diagnostics {
self.broadcast(Event::Diagnostic(event))?;
}
Ok(())
}
fn handle_message(&self, message: RawMessage) -> Result<Option<Vec<u8>>> {
let message_for_error = message.clone();
let res = match message.code {
message_codes::CONNECT => self.handle_connect(message),
message_codes::CHAT => self.handle_chat(message),
message_codes::EXIT => self.handle_participant_list_or_exit(message),
message_codes::PARTICIPANT_FOLLOW_STATE_LIST => Ok(None),
message_codes::PARTICIPANT_LIST_COMPLETE => {
self.handle_participant_list_complete(message)
}
message_codes::FREEZE => self.handle_freeze(message),
message_codes::MUTE => self.handle_mute(message),
message_codes::MANAGER_CHAT => self.handle_manager_message(message),
message_codes::EMOTICON => self.handle_emoticon_message(message),
message_codes::NOTIFICATION => self.handle_notification(message),
message_codes::BJ_STATE_CHANGE => self.handle_bj_state_change(message),
message_codes::SLOW => self.handle_slow(message),
message_codes::KICK_CANCEL => self.handle_kick_cancel(message),
message_codes::SUBSCRIBE => self.handle_subscribe(message),
message_codes::SUBSCRIBE_RENEW => self.handle_subscribe_renew(message),
message_codes::STICKER | message_codes::SUB_STICKER => self.handle_sticker(message),
message_codes::OGQ_GIFTED
| message_codes::QUICKVIEW_GIFTED
| message_codes::SUBSCRIPTION_GIFTED => self.handle_gift(message),
message_codes::MISSION_DONATION => self.handle_mission(message),
message_codes::DONATION
| message_codes::ADBALLOON_DONATION
| message_codes::SUB_DONATION
| message_codes::VOD_AD_DONATION
| message_codes::VOD_DONATION
| message_codes::AD_STATION_DONATION
| message_codes::VIDEO_DONATION => self.handle_donation(message),
_ => {
self.broadcast_unknown(&message);
Ok(None)
}
};
match res {
Ok(response) => Ok(response),
Err(error) => {
self.broadcast_parse_failed(
error,
message_for_error.raw.clone(),
Some(&message_for_error),
)?;
Ok(None)
}
}
}
fn handle_donation(
&self,
message: RawMessage,
) -> std::result::Result<Option<Vec<u8>>, ParseError> {
let e = match message.code {
message_codes::DONATION => parse_balloon_event(message),
message_codes::SUB_DONATION => parse_balloon_sub_event(message),
message_codes::VOD_DONATION => parse_vod_balloon_event(message),
message_codes::VOD_AD_DONATION => parse_vod_ad_balloon_event(message),
message_codes::ADBALLOON_DONATION => parse_ad_balloon_event(message),
message_codes::AD_STATION_DONATION => parse_station_ad_balloon_event(message),
message_codes::VIDEO_DONATION => parse_video_balloon_event(message),
_ => return Ok(None),
}?;
let _ = self.broadcast(Event::Donation(e));
Ok(None)
}
fn handle_sticker(
&self,
message: RawMessage,
) -> std::result::Result<Option<Vec<u8>>, ParseError> {
let e = match message.code {
message_codes::STICKER => parse_sticker_event(message),
message_codes::SUB_STICKER => parse_sticker_sub_event(message),
_ => return Ok(None),
}?;
let _ = self.broadcast(Event::Sticker(e));
Ok(None)
}
fn handle_gift(&self, message: RawMessage) -> std::result::Result<Option<Vec<u8>>, ParseError> {
let e = match message.code {
message_codes::OGQ_GIFTED => parse_ogq_gift_event(message),
message_codes::SUBSCRIPTION_GIFTED => parse_subscribe_gift_event(message),
message_codes::QUICKVIEW_GIFTED => parse_quickview_gift_event(message),
_ => return Ok(None),
}?;
let _ = self.broadcast(Event::Gift(e));
Ok(None)
}
fn handle_mission(
&self,
message: RawMessage,
) -> std::result::Result<Option<Vec<u8>>, ParseError> {
match parse_mission_event(message)? {
ParsedMissionEvent::Mission(e) => {
let _ = self.broadcast(Event::MissionDonation(e));
}
ParsedMissionEvent::MissionTotal(e) => {
let _ = self.broadcast(Event::MissionTotal(e));
}
ParsedMissionEvent::BattleNotice(e) => {
let _ = self.broadcast(Event::BattleMissionResult(e));
}
ParsedMissionEvent::ChallengeNotice(e) => {
let _ = self.broadcast(Event::ChallengeMissionResult(e));
}
}
Ok(None)
}
fn handle_subscribe_renew(
&self,
message: RawMessage,
) -> std::result::Result<Option<Vec<u8>>, ParseError> {
let _ = self.broadcast(Event::Subscribe(parse_subscribe_renew_event(message)?));
Ok(None)
}
fn handle_subscribe(
&self,
message: RawMessage,
) -> std::result::Result<Option<Vec<u8>>, ParseError> {
let _ = self.broadcast(Event::Subscribe(parse_subscribe_event(message)?));
Ok(None)
}
fn handle_slow(&self, message: RawMessage) -> std::result::Result<Option<Vec<u8>>, ParseError> {
let _ = self.broadcast(Event::Slow(parse_slow_event(message)?));
Ok(None)
}
fn handle_bj_state_change(
&self,
_: RawMessage,
) -> std::result::Result<Option<Vec<u8>>, ParseError> {
let _ = self.broadcast(Event::BJStateChange);
Ok(None)
}
fn handle_emoticon_message(
&self,
message: RawMessage,
) -> std::result::Result<Option<Vec<u8>>, ParseError> {
let _ = self.broadcast(Event::Chat(parse_emoticon_event(message)?));
Ok(None)
}
fn handle_notification(
&self,
message: RawMessage,
) -> std::result::Result<Option<Vec<u8>>, ParseError> {
let _ = self.broadcast(Event::Notification(parse_notification_event(message)?));
Ok(None)
}
fn handle_manager_message(
&self,
message: RawMessage,
) -> std::result::Result<Option<Vec<u8>>, ParseError> {
let _ = self.broadcast(Event::Chat(parse_manager_chat_event(message)?));
Ok(None)
}
fn handle_kick_cancel(
&self,
message: RawMessage,
) -> std::result::Result<Option<Vec<u8>>, ParseError> {
if let Some(e) = parse_kick_cancel_event(message)? {
let _ = self.broadcast(Event::KickCancel(e));
}
Ok(None)
}
fn handle_participant_list_or_exit(
&self,
message: RawMessage,
) -> std::result::Result<Option<Vec<u8>>, ParseError> {
if let Some(chunk) = parse_participant_list_chunk(message.clone())? {
self.with_participant_list(|list| list.push_participant_chunk(chunk));
return Ok(None);
}
self.handle_exit(message)
}
fn handle_participant_list_complete(
&self,
message: RawMessage,
) -> std::result::Result<Option<Vec<u8>>, ParseError> {
let complete = parse_participant_list_complete_event(message)?;
let event = self.with_participant_list(|list| list.finish(complete));
let _ = self.broadcast(Event::ParticipantList(event));
Ok(None)
}
fn handle_mute(&self, message: RawMessage) -> std::result::Result<Option<Vec<u8>>, ParseError> {
let _ = self.broadcast(Event::Mute(parse_mute_event(message)?));
Ok(None)
}
fn handle_freeze(
&self,
message: RawMessage,
) -> std::result::Result<Option<Vec<u8>>, ParseError> {
let _ = self.broadcast(Event::Freeze(parse_freeze_event(message)?));
Ok(None)
}
fn handle_chat(&self, message: RawMessage) -> std::result::Result<Option<Vec<u8>>, ParseError> {
let _ = self.broadcast(Event::Chat(parse_chat_event(message)?));
Ok(None)
}
fn handle_exit(&self, message: RawMessage) -> std::result::Result<Option<Vec<u8>>, ParseError> {
if let Some((is_kick, e)) = parse_exit_event(message)? {
if is_kick {
let _ = self.broadcast(Event::Kick(e));
} else {
let _ = self.broadcast(Event::Exit(e));
};
}
Ok(None)
}
fn handle_connect(&self, _: RawMessage) -> std::result::Result<Option<Vec<u8>>, ParseError> {
let ret = self.formatter.format_message(MessageType::JOIN);
Ok(Some(ret))
}
fn broadcast_unknown(&self, message: &RawMessage) {
let _ = self.broadcast(Event::Unknown(message.code));
let _ = self.broadcast_diagnostic(DiagnosticEvent::UnknownCode(UnknownCodeDiagnostic {
meta: EventMeta {
received_time: message.received_time,
},
code: message.code,
ret_code: message.ret_code,
field_count: message.body.len(),
body: message.body.clone(),
raw: message.raw.clone(),
}));
}
fn broadcast_parse_failed(
&self,
error: ParseError,
raw: std::sync::Arc<[u8]>,
message: Option<&RawMessage>,
) -> Result<()> {
let (field_index, expected_type) = match &error {
ParseError::MissingField { index, .. } => (Some(*index), None),
ParseError::InvalidNumber { index, target, .. } => {
(Some(*index), Some((*target).to_string()))
}
ParseError::InvalidJson { index, .. } => (Some(*index), Some("json".to_string())),
_ => (None, None),
};
let diagnostic = ParseFailedDiagnostic {
meta: EventMeta {
received_time: chrono::Utc::now(),
},
code: message.map(|message| message.code),
body_field_count: message.map(|message| message.body.len()),
field_index,
expected_type,
reason: error.to_string(),
raw,
};
self.broadcast_diagnostic(DiagnosticEvent::ParseFailed(diagnostic))
}
fn with_participant_list<T>(&self, f: impl FnOnce(&mut ParticipantListAccumulator) -> T) -> T {
let mut participant_list = self
.participant_list
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner());
f(&mut participant_list)
}
}
#[derive(Default)]
struct ParticipantListAccumulator {
entries: Vec<User>,
entry_indexes: HashMap<String, usize>,
}
impl ParticipantListAccumulator {
fn push_participant_chunk(&mut self, chunk: ParticipantListChunk) {
for participant in chunk.entries {
if self.entry_indexes.contains_key(&participant.id) {
continue;
}
self.entry_indexes
.insert(participant.id.clone(), self.entries.len());
self.entries.push(participant);
}
}
fn finish(&mut self, complete: ParticipantListCompleteEvent) -> ParticipantListEvent {
let entries = std::mem::take(&mut self.entries);
self.entry_indexes.clear();
ParticipantListEvent {
meta: complete.meta,
result_code: complete.result_code,
count: entries.len(),
entries,
}
}
}
#[cfg(test)]
mod tests {
use tokio::sync::{broadcast, mpsc};
use super::*;
use crate::{chat::constants::message_codes, models::LiveDetail};
fn frame(code: u32, body: &[u8]) -> Vec<u8> {
let mut bytes = Vec::new();
bytes.extend_from_slice(&[27, 9]);
bytes.extend_from_slice(format!("{code:04}").as_bytes());
bytes.extend_from_slice(format!("{:06}", body.len()).as_bytes());
bytes.extend_from_slice(b"00");
bytes.extend_from_slice(body);
bytes
}
fn handler_with_diagnostics(diagnostics: bool) -> (MessageHandler, broadcast::Receiver<Event>) {
let (event_tx, event_rx) = broadcast::channel(16);
let (command_tx, _) = mpsc::channel(4);
let formatter = ChatFormatter::new(
LiveDetail {
is_live: true,
ch_domain: "chat.example".to_string(),
ch_pt: 443,
bno: 123,
ch_no: "123".to_string(),
streamer_nick: "streamer".to_string(),
title: "title".to_string(),
categories: vec![],
},
String::new(),
);
(
MessageHandler::new(&formatter, event_tx, command_tx, diagnostics),
event_rx,
)
}
fn handler() -> (MessageHandler, broadcast::Receiver<Event>) {
handler_with_diagnostics(true)
}
#[test]
fn emits_unknown_code_diagnostic() {
let (handler, mut events) = handler();
handler.handle(frame(9999, b"\x0cnew")).unwrap();
assert!(matches!(
events.try_recv().unwrap(),
Event::Diagnostic(DiagnosticEvent::RawFrame(_))
));
assert!(matches!(
events.try_recv().unwrap(),
Event::Diagnostic(DiagnosticEvent::DecodedFrame(_))
));
assert!(matches!(events.try_recv().unwrap(), Event::Unknown(9999)));
assert!(matches!(
events.try_recv().unwrap(),
Event::Diagnostic(DiagnosticEvent::UnknownCode(_))
));
}
#[test]
fn emits_parse_failed_diagnostic_for_malformed_chat() {
let (handler, mut events) = handler();
handler
.handle(frame(message_codes::CHAT, b"\x0conly-comment"))
.unwrap();
let mut saw_parse_failed = false;
while let Ok(event) = events.try_recv() {
if let Event::Diagnostic(DiagnosticEvent::ParseFailed(diagnostic)) = event {
saw_parse_failed = true;
assert_eq!(diagnostic.code, Some(message_codes::CHAT));
assert_eq!(diagnostic.field_index, Some(1));
}
}
assert!(saw_parse_failed);
}
#[test]
fn suppresses_diagnostic_events_when_disabled() {
let (handler, mut events) = handler_with_diagnostics(false);
handler.handle(frame(9999, b"\x0cnew")).unwrap();
assert!(matches!(events.try_recv().unwrap(), Event::Unknown(9999)));
assert!(events.try_recv().is_err());
}
#[test]
fn aggregates_participant_list_until_complete() {
let (handler, mut events) = handler_with_diagnostics(false);
handler
.handle(frame(
message_codes::EXIT,
b"\x0c1\x0cwoalsdl1234\x0c\xec\x9d\xb4\xed\x8c\x8c\xeb\xa6\xac\xec\x95\x84\x0c537477152|32768\x0cksch1028(3)\x0cuser\x0c268550176|557056\x0c",
))
.unwrap();
assert!(events.try_recv().is_err());
handler
.handle(frame(
message_codes::PARTICIPANT_FOLLOW_STATE_LIST,
b"\x0cwoalsdl1234\x0cfw=-1&afw=-1\x0cksch1028(3)\x0cfw=15&afw=15\x0c",
))
.unwrap();
assert!(events.try_recv().is_err());
handler
.handle(b"\x1b\t011000000300\x0c1\x0c".to_vec())
.unwrap();
match events.try_recv().unwrap() {
Event::ParticipantList(event) => {
assert_eq!(event.result_code, "1");
assert_eq!(event.count, 2);
assert_eq!(event.entries.len(), 2);
assert_eq!(event.entries[0].id, "woalsdl1234");
assert!(event.entries[0].subscribe.is_none());
assert_eq!(event.entries[1].id, "ksch1028");
assert!(event.entries[1].subscribe.is_none());
}
other => panic!("unexpected event: {other:?}"),
}
}
#[test]
fn keeps_existing_exit_handling_for_exit_shape() {
let (handler, mut events) = handler_with_diagnostics(false);
handler
.handle(frame(
message_codes::EXIT,
b"\x0c0\x0cuser1\x0cnick\x0c1\x0cunused\x0c65536|163840",
))
.unwrap();
assert!(matches!(events.try_recv().unwrap(), Event::Exit(_)));
}
#[test]
fn discards_participant_follow_state_for_code_127() {
let (handler, mut events) = handler_with_diagnostics(false);
handler
.handle(frame(
message_codes::PARTICIPANT_FOLLOW_STATE_LIST,
b"\x0cwoalsdl1234\x0cfw=-1&afw=-1\x0cksch1028(3)\x0cfw=15&afw=15\x0c",
))
.unwrap();
assert!(events.try_recv().is_err());
}
#[test]
fn discards_non_list_code_127() {
let (handler, mut events) = handler_with_diagnostics(false);
handler
.handle(frame(
message_codes::PARTICIPANT_FOLLOW_STATE_LIST,
b"\x0cuser1\x0cignored\x0cignored",
))
.unwrap();
assert!(events.try_recv().is_err());
}
#[test]
fn emits_empty_participant_list_on_complete_without_chunks() {
let (handler, mut events) = handler_with_diagnostics(false);
handler
.handle(b"\x1b\t011000000300\x0c1\x0c".to_vec())
.unwrap();
match events.try_recv().unwrap() {
Event::ParticipantList(event) => {
assert_eq!(event.result_code, "1");
assert_eq!(event.count, 0);
assert!(event.entries.is_empty());
}
other => panic!("unexpected event: {other:?}"),
}
}
}