Skip to content

Commit

Permalink
Deprecate StreamMuxer::is_remote_acknowledged (#1616)
Browse files Browse the repository at this point in the history
  • Loading branch information
tomaka authored Jun 19, 2020
1 parent 563d337 commit 00fc223
Show file tree
Hide file tree
Showing 7 changed files with 7 additions and 67 deletions.
8 changes: 0 additions & 8 deletions core/src/connection.rs
Original file line number Diff line number Diff line change
Expand Up @@ -230,14 +230,6 @@ where
self.handler.inject_event(event);
}

/// Returns `true` if the remote has shown any sign of activity
/// since the connection has been established.
///
/// See also [`StreamMuxer::is_remote_acknowledged`].
pub fn is_remote_acknowledged(&self) -> bool {
self.muxing.is_remote_acknowledged()
}

/// Begins an orderly shutdown of the connection, returning a
/// `Future` that resolves when connection shutdown is complete.
pub fn close(self) -> Close<TMuxer> {
Expand Down
7 changes: 0 additions & 7 deletions core/src/connection/substream.rs
Original file line number Diff line number Diff line change
Expand Up @@ -123,13 +123,6 @@ where
self.outbound_substreams.push((user_data, raw));
}

/// Returns `true` if the remote has shown any sign of activity after the muxer has been open.
///
/// See `StreamMuxer::is_remote_acknowledged`.
pub fn is_remote_acknowledged(&self) -> bool {
self.inner.is_remote_acknowledged()
}

/// Destroys the node stream and returns all the pending outbound substreams, plus an object
/// that signals the remote that we shut down the connection.
#[must_use]
Expand Down
7 changes: 0 additions & 7 deletions core/src/either.rs
Original file line number Diff line number Diff line change
Expand Up @@ -297,13 +297,6 @@ where
}
}

fn is_remote_acknowledged(&self) -> bool {
match self {
EitherOutput::First(inner) => inner.is_remote_acknowledged(),
EitherOutput::Second(inner) => inner.is_remote_acknowledged()
}
}

fn close(&self, cx: &mut Context) -> Poll<Result<(), Self::Error>> {
match self {
EitherOutput::First(inner) => inner.close(cx).map_err(|e| e.into()),
Expand Down
18 changes: 5 additions & 13 deletions core/src/muxing.rs
Original file line number Diff line number Diff line change
Expand Up @@ -64,8 +64,7 @@ mod singleton;
///
/// The state of a muxer, as exposed by this API, is the following:
///
/// - A connection to the remote. The `is_remote_acknowledged`, `flush_all` and `close` methods
/// operate on this.
/// - A connection to the remote. The `flush_all` and `close` methods operate on this.
/// - A list of substreams that are open. The `poll_inbound`, `poll_outbound`, `read_substream`,
/// `write_substream`, `flush_substream`, `shutdown_substream` and `destroy_substream` methods
/// allow controlling these entries.
Expand Down Expand Up @@ -180,7 +179,10 @@ pub trait StreamMuxer {
/// allowed to assume that the handshake has succeeded when it didn't in fact succeed. This
/// method can be called in order to determine whether the remote has accepted our handshake or
/// has potentially not received it yet.
fn is_remote_acknowledged(&self) -> bool;
#[deprecated(note = "This method is unused and will be removed in the future")]
fn is_remote_acknowledged(&self) -> bool {
true
}

/// Closes this `StreamMuxer`.
///
Expand Down Expand Up @@ -525,11 +527,6 @@ impl StreamMuxer for StreamMuxerBox {
self.inner.close(cx)
}

#[inline]
fn is_remote_acknowledged(&self) -> bool {
self.inner.is_remote_acknowledged()
}

#[inline]
fn flush_all(&self, cx: &mut Context) -> Poll<Result<(), Self::Error>> {
self.inner.flush_all(cx)
Expand Down Expand Up @@ -631,11 +628,6 @@ where
self.inner.close(cx).map_err(|e| e.into())
}

#[inline]
fn is_remote_acknowledged(&self) -> bool {
self.inner.is_remote_acknowledged()
}

#[inline]
fn flush_all(&self, cx: &mut Context) -> Poll<Result<(), Self::Error>> {
self.inner.flush_all(cx).map_err(|e| e.into())
Expand Down
13 changes: 1 addition & 12 deletions core/src/muxing/singleton.rs
Original file line number Diff line number Diff line change
Expand Up @@ -35,8 +35,6 @@ pub struct SingletonMuxer<TSocket> {
substream_extracted: AtomicBool,
/// Our local endpoint. Always the same value as was passed to `new`.
endpoint: Endpoint,
/// If true, we have received data from the remote.
remote_acknowledged: AtomicBool,
}

impl<TSocket> SingletonMuxer<TSocket> {
Expand All @@ -49,7 +47,6 @@ impl<TSocket> SingletonMuxer<TSocket> {
inner: Mutex::new(inner),
substream_extracted: AtomicBool::new(false),
endpoint,
remote_acknowledged: AtomicBool::new(false),
}
}
}
Expand Down Expand Up @@ -101,11 +98,7 @@ where
}

fn read_substream(&self, cx: &mut Context, _: &mut Self::Substream, buf: &mut [u8]) -> Poll<Result<usize, io::Error>> {
let res = AsyncRead::poll_read(Pin::new(&mut *self.inner.lock()), cx, buf);
if let Poll::Ready(Ok(_)) = res {
self.remote_acknowledged.store(true, Ordering::Release);
}
res
AsyncRead::poll_read(Pin::new(&mut *self.inner.lock()), cx, buf)
}

fn write_substream(&self, cx: &mut Context, _: &mut Self::Substream, buf: &[u8]) -> Poll<Result<usize, io::Error>> {
Expand All @@ -123,10 +116,6 @@ where
fn destroy_substream(&self, _: Self::Substream) {
}

fn is_remote_acknowledged(&self) -> bool {
self.remote_acknowledged.load(Ordering::Acquire)
}

fn close(&self, cx: &mut Context) -> Poll<Result<(), io::Error>> {
// The `StreamMuxer` trait requires that `close()` implies `flush_all()`.
self.flush_all(cx)
Expand Down
8 changes: 0 additions & 8 deletions muxers/mplex/src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -110,7 +110,6 @@ impl MplexConfig {
to_wake: Mutex::new(Default::default()),
}),
is_shutdown: false,
is_acknowledged: false,
})
}
}
Expand Down Expand Up @@ -203,8 +202,6 @@ struct MultiplexInner<C> {
/// If true, the connection has been shut down. We need to be careful not to accidentally
/// call `Sink::poll_complete` or `Sink::start_send` after `Sink::close`.
is_shutdown: bool,
/// If true, the remote has sent data to us.
is_acknowledged: bool,
}

struct Notifier {
Expand Down Expand Up @@ -295,7 +292,6 @@ where C: AsyncRead + AsyncWrite + Unpin,
};

trace!("Received message: {:?}", elem);
inner.is_acknowledged = true;

// Handle substreams opening/closing.
match elem {
Expand Down Expand Up @@ -587,10 +583,6 @@ where C: AsyncRead + AsyncWrite + Unpin
})
}

fn is_remote_acknowledged(&self) -> bool {
self.inner.lock().is_acknowledged
}

fn close(&self, cx: &mut Context) -> Poll<Result<(), IoError>> {
let inner = &mut *self.inner.lock();
if inner.is_shutdown {
Expand Down
13 changes: 1 addition & 12 deletions muxers/yamux/src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -43,8 +43,6 @@ struct Inner<S> {
incoming: S,
/// Handle to control the connection.
control: yamux::Control,
/// True, once we have received an inbound substream.
acknowledged: bool
}

/// A token to poll for an outbound substream.
Expand All @@ -66,7 +64,6 @@ where
_marker: std::marker::PhantomData
},
control: ctrl,
acknowledged: false
};
Yamux(Mutex::new(inner))
}
Expand All @@ -87,7 +84,6 @@ where
_marker: std::marker::PhantomData
},
control: ctrl,
acknowledged: false
};
Yamux(Mutex::new(inner))
}
Expand All @@ -106,10 +102,7 @@ where
fn poll_inbound(&self, c: &mut Context) -> Poll<Self::Substream> {
let mut inner = self.0.lock();
match ready!(inner.incoming.poll_next_unpin(c)) {
Some(Ok(s)) => {
inner.acknowledged = true;
Poll::Ready(Ok(s))
}
Some(Ok(s)) => Poll::Ready(Ok(s)),
Some(Err(e)) => Poll::Ready(Err(e)),
None => Poll::Ready(Err(yamux::ConnectionError::Closed.into()))
}
Expand Down Expand Up @@ -146,10 +139,6 @@ where

fn destroy_substream(&self, _: Self::Substream) { }

fn is_remote_acknowledged(&self) -> bool {
self.0.lock().acknowledged
}

fn close(&self, c: &mut Context) -> Poll<()> {
let mut inner = self.0.lock();
if let std::task::Poll::Ready(x) = Pin::new(&mut inner.control).poll_close(c) {
Expand Down

0 comments on commit 00fc223

Please sign in to comment.