Only send Close message over the websocket connection if not shutting down as result of Close (21dc680f) · Commits · Orm Finnendahl / WASP

server/src/publisher.rs

+8 −6

Original line number	Diff line number	Diff line
		@@ -356,13 +356,13 @@ impl Publisher {
		self.remote_addr, err
		);
		ctx.notify(ErrorMessage(String::from("Internal processing error")));
		self.shutdown(ctx);
		self.shutdown(ctx, false);
		}
		}
		}

		/// Shut down this publisher and delete its room.
		fn shutdown(&mut self, ctx: &mut ws::WebsocketContext<Self>) {
		fn shutdown(&mut self, ctx: &mut ws::WebsocketContext<Self>, from_close: bool) {
		debug!("Shutting down publisher {}", self.remote_addr);

		let _ = self.pipeline.set_state(gst::State::Null);
		@@ -384,7 +384,9 @@ impl Publisher {
		self.app_sink
		.set_callbacks(gst_app::AppSinkCallbacks::builder().build());

		if !from_close {
		ctx.close(None);
		}
		ctx.stop();
		}

		@@ -580,7 +582,7 @@ impl Actor for Publisher {
		/// Called when the publisher is fully stopped.
		fn stopped(&mut self, ctx: &mut Self::Context) {
		trace!("Publisher {} stopped", self.remote_addr);
		self.shutdown(ctx);
		self.shutdown(ctx, false);
		}
		}

		@@ -597,7 +599,7 @@ impl StreamHandler<Result<ws::Message, ws::ProtocolError>> for Publisher {
		"Publisher {} websocket connection closed: {:?}",
		self.remote_addr, reason
		);
		self.shutdown(ctx);
		self.shutdown(ctx, true);
		}
		Ok(ws::Message::Binary(_binary)) => {
		error!("Unsupported binary message, ignoring");
		@@ -613,7 +615,7 @@ impl StreamHandler<Result<ws::Message, ws::ProtocolError>> for Publisher {
		"Publisher {} websocket connection error: {:?}",
		self.remote_addr, err
		);
		self.shutdown(ctx);
		self.shutdown(ctx, false);
		}
		}
		}
		@@ -708,7 +710,7 @@ impl Handler<ErrorMessage> for Publisher {
		serde_json::to_string(&ServerMessage::Error { message: msg.0 })
		.expect("Failed to serialize error message"),
		);
		self.shutdown(ctx);
		self.shutdown(ctx, false);
		}
		}

server/src/subscriber.rs

+11 −7

Original line number	Diff line number	Diff line
		@@ -241,6 +241,8 @@ impl Subscriber {
		.expect("Invalid argument")
		.unwrap();

		// TODO: SDP media should be sendonly

		debug!("Created offer {:#?}", offer.get_sdp());

		webrtcbin
		@@ -358,13 +360,13 @@ impl Subscriber {
		self.remote_addr, err
		);
		ctx.notify(ErrorMessage(String::from("Internal processing error")));
		self.shutdown(ctx);
		self.shutdown(ctx, false);
		}
		}
		}

		/// Shut down this subscriber and delete its room.
		fn shutdown(&mut self, ctx: &mut ws::WebsocketContext<Self>) {
		fn shutdown(&mut self, ctx: &mut ws::WebsocketContext<Self>, from_close: bool) {
		debug!("Shutting down subscriber {}", self.remote_addr);

		let _ = self.pipeline.set_state(gst::State::Null);
		@@ -378,7 +380,9 @@ impl Subscriber {
		}
		}

		if !from_close {
		ctx.close(None);
		}
		ctx.stop();
		}
		}
		@@ -433,7 +437,7 @@ impl Actor for Subscriber {
		/// Called when the subscriber is fully stopped.
		fn stopped(&mut self, ctx: &mut Self::Context) {
		trace!("Subscriber {} stopped", self.remote_addr);
		self.shutdown(ctx);
		self.shutdown(ctx, false);
		}
		}

		@@ -449,7 +453,7 @@ impl StreamHandler<Result<ws::Message, ws::ProtocolError>> for Subscriber {
		"Publisher {} websocket connection closed: {:?}",
		self.remote_addr, reason
		);
		self.shutdown(ctx);
		self.shutdown(ctx, true);
		}
		Ok(ws::Message::Binary(_binary)) => {
		error!("Unsupported binary message, ignoring");
		@@ -465,7 +469,7 @@ impl StreamHandler<Result<ws::Message, ws::ProtocolError>> for Subscriber {
		"Subscriber {} websocket connection error: {:?}",
		self.remote_addr, err
		);
		self.shutdown(ctx);
		self.shutdown(ctx, false);
		}
		}
		}
		@@ -488,7 +492,7 @@ impl Handler<RoomDeletedMessage> for Subscriber {
		ctx: &mut ws::WebsocketContext<Self>,
		) -> Self::Result {
		debug!("Room deleted for subscriber {}", self.remote_addr);
		self.shutdown(ctx);
		self.shutdown(ctx, false);
		}
		}

		@@ -513,7 +517,7 @@ impl Handler<ErrorMessage> for Subscriber {
		serde_json::to_string(&ServerMessage::Error { message: msg.0 })
		.expect("Failed to serialize error message"),
		);
		self.shutdown(ctx);
		self.shutdown(ctx, false);
		}
		}