Skip to content
This repository was archived by the owner on Oct 24, 2022. It is now read-only.

protocol: Rework to allow using the same socket for several peerTypes #76

Merged
merged 4 commits into from
Aug 30, 2022
Merged

protocol: Rework to allow using the same socket for several peerTypes #76

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
5c31269
protocol: Add an "Unregister" message type
thiblahute Jul 12, 2022
0c73b9c
Refactor signalling protocol around a Session ID
thiblahute Jul 14, 2022
50eeefa
protocol: Add a welcoming message, removing registering as consumer r…
thiblahute Aug 18, 2022
c886607
protocol: Rework the way peers set their status
thiblahute Aug 18, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
47 changes: 26 additions & 21 deletions plugins/src/signaller/imp.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -105,10 +105,11 @@ impl Signaller {
} else {
None
};

websocket_sender
.send(p::IncomingMessage::Register(p::RegisterMessage::Producer {
.send(p::IncomingMessage::SetPeerStatus(p::PeerStatus {
roles: vec![p::PeerRole::Producer],
meta,
peer_id: None,
}))
.await?;

Expand All @@ -122,34 +123,38 @@ impl Signaller {

if let Ok(msg) = serde_json::from_str::<p::OutgoingMessage>(&msg) {
match msg {
p::OutgoingMessage::Registered(
p::RegisteredMessage::Producer { peer_id, .. },
) => {
p::OutgoingMessage::Welcome { peer_id } => {
gst::info!(
CAT,
obj: &element,
"We are registered with the server, our peer id is {}",
peer_id
);
}
p::OutgoingMessage::Registered(_) => unreachable!(),
p::OutgoingMessage::StartSession { peer_id } => {
if let Err(err) = element.add_consumer(&peer_id) {
p::OutgoingMessage::StartSession {
session_id,
peer_id,
} => {
if let Err(err) =
element.start_session(&session_id, &peer_id)
{
gst::warning!(CAT, obj: &element, "{}", err);
}
}
p::OutgoingMessage::EndSession { peer_id } => {
if let Err(err) = element.remove_consumer(&peer_id) {
p::OutgoingMessage::EndSession(session_info) => {
if let Err(err) =
element.end_session(&session_info.session_id)
{
gst::warning!(CAT, obj: &element, "{}", err);
}
}
p::OutgoingMessage::Peer(p::PeerMessage {
peer_id,
session_id,
peer_message,
}) => match peer_message {
p::PeerMessageInner::Sdp(p::SdpMessage::Answer { sdp }) => {
if let Err(err) = element.handle_sdp(
&peer_id,
&session_id,
&gst_webrtc::WebRTCSessionDescription::new(
gst_webrtc::WebRTCSDPType::Answer,
gst_sdp::SDPMessage::parse_buffer(
Expand All @@ -175,7 +180,7 @@ impl Signaller {
sdp_m_line_index,
} => {
if let Err(err) = element.handle_ice(
&peer_id,
&session_id,
Some(sdp_m_line_index),
None,
&candidate,
Expand Down Expand Up @@ -254,13 +259,13 @@ impl Signaller {
pub fn handle_sdp(
&self,
element: &WebRTCSink,
peer_id: &str,
session_id: &str,
sdp: &gst_webrtc::WebRTCSessionDescription,
) {
let state = self.state.lock().unwrap();

let msg = p::IncomingMessage::Peer(p::PeerMessage {
peer_id: peer_id.to_string(),
session_id: session_id.to_string(),
peer_message: p::PeerMessageInner::Sdp(p::SdpMessage::Offer {
sdp: sdp.sdp().as_text().unwrap(),
}),
Expand All @@ -281,15 +286,15 @@ impl Signaller {
pub fn handle_ice(
&self,
element: &WebRTCSink,
peer_id: &str,
session_id: &str,
candidate: &str,
sdp_m_line_index: Option<u32>,
_sdp_mid: Option<String>,
) {
let state = self.state.lock().unwrap();

let msg = p::IncomingMessage::Peer(p::PeerMessage {
peer_id: peer_id.to_string(),
session_id: session_id.to_string(),
peer_message: p::PeerMessageInner::Ice {
candidate: candidate.to_string(),
sdp_m_line_index: sdp_m_line_index.unwrap(),
Expand Down Expand Up @@ -331,17 +336,17 @@ impl Signaller {
}
}

pub fn consumer_removed(&self, element: &WebRTCSink, peer_id: &str) {
gst::debug!(CAT, obj: element, "Signalling consumer {} removed", peer_id);
pub fn end_session(&self, element: &WebRTCSink, session_id: &str) {
gst::debug!(CAT, obj: element, "Signalling session {} ended", session_id);

let state = self.state.lock().unwrap();
let peer_id = peer_id.to_string();
let session_id = session_id.to_string();
let element = element.downgrade();
if let Some(mut sender) = state.websocket_sender.clone() {
task::spawn(async move {
if let Err(err) = sender
.send(p::IncomingMessage::EndSession(p::EndSessionMessage {
peer_id: peer_id.to_string(),
session_id: session_id.to_string(),
}))
.await
{
Expand Down
8 changes: 4 additions & 4 deletions plugins/src/signaller/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,13 +34,13 @@ impl Signallable for Signaller {
fn handle_ice(
&mut self,
element: &WebRTCSink,
peer_id: &str,
session_id: &str,
candidate: &str,
sdp_mline_index: Option<u32>,
sdp_mid: Option<String>,
) -> Result<(), Box<dyn Error>> {
let signaller = imp::Signaller::from_instance(self);
signaller.handle_ice(element, peer_id, candidate, sdp_mline_index, sdp_mid);
signaller.handle_ice(element, session_id, candidate, sdp_mline_index, sdp_mid);
Ok(())
}

Expand All @@ -49,9 +49,9 @@ impl Signallable for Signaller {
signaller.stop(element);
}

fn consumer_removed(&mut self, element: &WebRTCSink, peer_id: &str) {
fn session_ended(&mut self, element: &WebRTCSink, session_id: &str) {
let signaller = imp::Signaller::from_instance(self);
signaller.consumer_removed(element, peer_id);
signaller.end_session(element, session_id);
}
}

Expand Down
Loading