use crate::one_to_many::callbacks::{
set_data_channel_on_error, set_data_channel_on_message, set_data_channel_on_open,
set_peer_connection_on_data_channel, set_peer_connection_on_ice_candidate,
set_peer_connection_on_ice_connection_state_change,
set_peer_connection_on_ice_gathering_state_change, set_peer_connection_on_negotiation_needed,
};
use crate::one_to_many::{Connection, NetworkManager};
use crate::utils::{create_peer_connection, create_sdp_answer, create_sdp_offer, IceCandidate};
use log::{debug, error, info};
use wasm_peers_protocol::one_to_many::SignalMessage;
use wasm_peers_protocol::UserId;
use wasm_bindgen::JsValue;
use wasm_bindgen_futures::JsFuture;
use web_sys::{
RtcIceCandidate, RtcIceCandidateInit, RtcSdpType, RtcSessionDescriptionInit, WebSocket,
};
pub(crate) async fn handle_websocket_message(
network_manager: NetworkManager,
message: SignalMessage,
websocket: WebSocket,
on_open_callback: impl FnMut(UserId) + Clone + 'static,
on_message_callback: impl FnMut(UserId, String) + Clone + 'static,
is_host: bool,
) -> Result<(), JsValue> {
match message {
SignalMessage::SessionJoin(_session_id, _user_id) => {
error!("error, SessionStartOrJoin should only be sent by peers to signaling server");
}
SignalMessage::SessionReady(session_id, peer_id) => {
info!(
"peer received info that session with {:?} is ready {:?}",
peer_id, session_id
);
let peer_connection =
create_peer_connection(&network_manager.inner.borrow().connection_type).unwrap();
set_peer_connection_on_data_channel(
&peer_connection,
peer_id,
network_manager.clone(),
on_open_callback.clone(),
on_message_callback.clone(),
);
set_peer_connection_on_ice_candidate(
&peer_connection,
peer_id,
websocket.clone(),
session_id.clone(),
);
set_peer_connection_on_ice_connection_state_change(&peer_connection);
set_peer_connection_on_ice_gathering_state_change(&peer_connection);
set_peer_connection_on_negotiation_needed(&peer_connection);
let data_channel =
peer_connection.create_data_channel(&format!("{}-{}", session_id, peer_id));
set_data_channel_on_open(&data_channel, peer_id, on_open_callback.clone());
set_data_channel_on_error(&data_channel);
set_data_channel_on_message(&data_channel, peer_id, on_message_callback.clone());
let offer = create_sdp_offer(&peer_connection).await?;
let signal_message = SignalMessage::SdpOffer(session_id, peer_id, offer);
let signal_message = serde_json_wasm::to_string(&signal_message)
.expect("failed to serialize SignalMessage");
websocket.send_with_str(&signal_message)?;
network_manager.inner.borrow_mut().connections.insert(
peer_id,
Connection::new(peer_connection.clone(), Some(data_channel.clone())),
);
debug!(
"(is_host: {}) sent an offer to {:?} successfully",
is_host, peer_id
);
}
SignalMessage::SdpOffer(session_id, user_id, offer) => {
let peer_connection =
create_peer_connection(&network_manager.inner.borrow().connection_type).unwrap();
set_peer_connection_on_data_channel(
&peer_connection,
user_id,
network_manager.clone(),
on_open_callback.clone(),
on_message_callback.clone(),
);
set_peer_connection_on_ice_candidate(
&peer_connection,
user_id,
websocket.clone(),
session_id.clone(),
);
set_peer_connection_on_ice_connection_state_change(&peer_connection);
set_peer_connection_on_ice_gathering_state_change(&peer_connection);
set_peer_connection_on_negotiation_needed(&peer_connection);
network_manager
.inner
.borrow_mut()
.connections
.insert(user_id, Connection::new(peer_connection.clone(), None));
debug!(
"(is_host: {}) added connection for {:?} successfully",
is_host, user_id
);
let answer = create_sdp_answer(&peer_connection, offer)
.await
.expect("failed to create SDP answer");
debug!(
"received an offer from {:?} and created an answer: {}",
user_id, answer
);
let signal_message = SignalMessage::SdpAnswer(session_id, user_id, answer);
let signal_message = serde_json_wasm::to_string(&signal_message)
.expect("failed to serialize SignalMessage");
websocket
.send_with_str(&signal_message)
.expect("failed to send SPD answer to signaling server");
}
SignalMessage::SdpAnswer(session_id, user_id, answer) => {
let peer_connection = network_manager
.inner
.borrow()
.connections
.get(&user_id)
.unwrap_or_else(|| {
panic!(
"(is_host: {}) no connection to send answer for given user_id: {:?}",
is_host, &user_id
)
})
.peer_connection
.clone();
let mut remote_session_description = RtcSessionDescriptionInit::new(RtcSdpType::Answer);
remote_session_description.sdp(&answer);
JsFuture::from(peer_connection.set_remote_description(&remote_session_description))
.await
.expect("failed to set remote description");
debug!(
"received answer from peer and set remote description: {}, {:?}",
answer, session_id
);
}
SignalMessage::IceCandidate(_session_id, user_id, ice_candidate) => {
let peer_connection = network_manager
.inner
.borrow()
.connections
.get(&user_id)
.unwrap_or_else(|| {
panic!(
"no connection to send ice candidate to for given user_id: {:?}",
&user_id
)
})
.peer_connection
.clone();
debug!("peer received ice candidate: {}", &ice_candidate);
let ice_candidate = serde_json_wasm::from_str::<IceCandidate>(&ice_candidate)
.expect("failed to deserialize IceCandidate");
let mut rtc_candidate = RtcIceCandidateInit::new("");
rtc_candidate.candidate(&ice_candidate.candidate);
rtc_candidate.sdp_m_line_index(ice_candidate.sdp_m_line_index);
rtc_candidate.sdp_mid(ice_candidate.sdp_mid.as_deref());
let rtc_candidate =
RtcIceCandidate::new(&rtc_candidate).expect("failed to create new RtcIceCandidate");
JsFuture::from(
peer_connection.add_ice_candidate_with_opt_rtc_ice_candidate(Some(&rtc_candidate)),
)
.await
.expect("failed to add ICE candidate");
debug!("added ice candidate {:?}", ice_candidate);
}
SignalMessage::Error(session_id, error) => {
error!(
"signaling server returned error: session id: {:?}, error: {}",
session_id, error
);
}
}
Ok(())
}