upgrade deps retty to v0.27.0

Author: yngrtc

Cargo.toml

@@ -12,7 +12,7 @@ keywords = ["networking", "protocols", "webrtc", "sans-io", "sfu"]
 categories = ["network-programming", "asynchronous", "multimedia"]
 
 [dependencies]
-retty = "0.24"
+retty = "0.27.0"
 bytes = "1.5"
 log = "0.4"
 base64 = "0.22"
@@ -27,14 +27,14 @@ url = { version = "2", features = [] }
 hex = { version = "0.4", features = [] }
 
 # RTC protocols
-shared = { version = "0.1", package = "rtc-shared" }
-sdp = { version = "0.1", package = "rtc-sdp" }
-stun = { version = "0.1", package = "rtc-stun" }
+shared = { version = "0.1.1", package = "rtc-shared" }
+sdp = { version = "0.1.1", package = "rtc-sdp" }
+stun = { version = "0.1.1", package = "rtc-stun" }
 rtp = { version = "0.1", package = "rtc-rtp" }
 rtcp = { version = "0.1", package = "rtc-rtcp" }
-srtp = { version = "0.1", package = "rtc-srtp" }
-dtls = { version = "0.1", package = "rtc-dtls" }
-sctp = { version = "0.1", package = "rtc-sctp" }
+srtp = { version = "0.1.1", package = "rtc-srtp" }
+dtls = { version = "0.1.1", package = "rtc-dtls" }
+sctp = { version = "0.1.1", package = "rtc-sctp" }
 datachannel = { version = "0.1", package = "rtc-datachannel" }
 
 [dev-dependencies]

examples/async_chat.rs

@@ -15,7 +15,7 @@ use log::{error, info};
 use retty::bootstrap::BootstrapUdpServer;
 use retty::channel::Pipeline;
 use retty::executor::LocalExecutorBuilder;
-use retty::transport::{AsyncTransport, AsyncTransportWrite, TaggedBytesMut};
+use retty::transport::TaggedBytesMut;
 use waitgroup::WaitGroup;
 
 use sfu::{
@@ -148,13 +148,10 @@ fn main() -> anyhow::Result<()> {
                 let server_states_moved = server_states.clone();
                 let mut bootstrap = BootstrapUdpServer::new();
                 bootstrap.pipeline(Box::new(
-                    move |writer: AsyncTransportWrite<TaggedBytesMut>| {
+                    move || {
                         let pipeline: Pipeline<TaggedBytesMut, TaggedBytesMut> = Pipeline::new();
 
-                        let local_addr = writer.get_local_addr();
-                        let async_transport_handler = AsyncTransport::new(writer);
                         let demuxer_handler = DemuxerHandler::new();
-                        let write_exception_handler = ExceptionHandler::new();
                         let stun_handler = StunHandler::new();
                         // DTLS
                         let dtls_handler = DtlsHandler::new(local_addr, Rc::clone(&server_states_moved));
@@ -165,11 +162,9 @@ fn main() -> anyhow::Result<()> {
                         let interceptor_handler = InterceptorHandler::new(Rc::clone(&server_states_moved));
                         // Gateway
                         let gateway_handler = GatewayHandler::new(Rc::clone(&server_states_moved));
-                        let read_exception_handler = ExceptionHandler::new();
+                        let exception_handler = ExceptionHandler::new();
 
-                        pipeline.add_back(async_transport_handler);
                         pipeline.add_back(demuxer_handler);
-                        pipeline.add_back(write_exception_handler);
                         pipeline.add_back(stun_handler);
                         // DTLS
                         pipeline.add_back(dtls_handler);
@@ -180,7 +175,7 @@ fn main() -> anyhow::Result<()> {
                         pipeline.add_back(interceptor_handler);
                         // Gateway
                         pipeline.add_back(gateway_handler);
-                        pipeline.add_back(read_exception_handler);
+                        pipeline.add_back(exception_handler);
 
                         pipeline.finalize()
                     },

examples/sync_signal/mod.rs

@@ -1,8 +1,5 @@
 #![allow(dead_code)]
 
-mod sync_transport;
-use sync_transport::SyncTransport;
-
 use bytes::{Bytes, BytesMut};
 use log::error;
 use retty::channel::{InboundPipeline, Pipeline};
@@ -14,7 +11,7 @@ use sfu::{
     SrtpHandler, StunHandler,
 };
 use std::cell::RefCell;
-use std::collections::{HashMap, VecDeque};
+use std::collections::HashMap;
 use std::io::{Error, ErrorKind, Read};
 use std::net::{SocketAddr, UdpSocket};
 use std::rc::Rc;
@@ -106,13 +103,7 @@ pub fn sync_run(
 
     println!("listening {}...", socket.local_addr()?);
 
-    let outgoing_queue = Rc::new(RefCell::new(VecDeque::new()));
-
-    let pipeline = build_pipeline(
-        socket.local_addr()?,
-        outgoing_queue.clone(),
-        server_states.clone(),
-    );
+    let pipeline = build_pipeline(socket.local_addr()?, server_states.clone());
 
     let mut buf = vec![0; 2000];
 
@@ -126,7 +117,7 @@ pub fn sync_run(
             }
         };
 
-        write_socket_output(&socket, &outgoing_queue);
+        write_socket_output(&socket, &pipeline);
 
         // Spawn new incoming signal message from the signaling server thread.
         if let Ok(signal_message) = rx.try_recv() {
@@ -167,9 +158,11 @@ pub fn sync_run(
     Ok(())
 }
 
-fn write_socket_output(socket: &UdpSocket, outgoing_queue: &Rc<RefCell<VecDeque<TaggedBytesMut>>>) {
-    let mut queue = outgoing_queue.borrow_mut();
-    while let Some(transmit) = queue.pop_front() {
+fn write_socket_output(
+    socket: &UdpSocket,
+    pipeline: &Rc<Pipeline<TaggedBytesMut, TaggedBytesMut>>,
+) {
+    while let Some(transmit) = pipeline.poll_transmit() {
         socket
             .send_to(&transmit.message, transmit.transport.peer_addr)
             .expect("sending UDP data");
@@ -200,14 +193,11 @@ fn read_socket_input(socket: &UdpSocket, buf: &mut [u8]) -> Option<TaggedBytesMu
 
 fn build_pipeline(
     local_addr: SocketAddr,
-    writer: Rc<RefCell<VecDeque<TaggedBytesMut>>>,
     server_states: Rc<RefCell<ServerStates>>,
 ) -> Rc<Pipeline<TaggedBytesMut, TaggedBytesMut>> {
     let pipeline: Pipeline<TaggedBytesMut, TaggedBytesMut> = Pipeline::new();
 
-    let sync_transport_handler = SyncTransport::new(writer);
     let demuxer_handler = DemuxerHandler::new();
-    let write_exception_handler = ExceptionHandler::new();
     let stun_handler = StunHandler::new();
     // DTLS
     let dtls_handler = DtlsHandler::new(local_addr, Rc::clone(&server_states));
@@ -218,11 +208,9 @@ fn build_pipeline(
     let interceptor_handler = InterceptorHandler::new(Rc::clone(&server_states));
     // Gateway
     let gateway_handler = GatewayHandler::new(Rc::clone(&server_states));
-    let read_exception_handler = ExceptionHandler::new();
+    let exception_handler = ExceptionHandler::new();
 
-    pipeline.add_back(sync_transport_handler);
     pipeline.add_back(demuxer_handler);
-    pipeline.add_back(write_exception_handler);
     pipeline.add_back(stun_handler);
     // DTLS
     pipeline.add_back(dtls_handler);
@@ -233,7 +221,7 @@ fn build_pipeline(
     pipeline.add_back(interceptor_handler);
     // Gateway
     pipeline.add_back(gateway_handler);
-    pipeline.add_back(read_exception_handler);
+    pipeline.add_back(exception_handler);
 
     pipeline.finalize()
 }