OMGeeky/tarpc
1
0
Fork 0
mirror of https://github.com/OMGeeky/tarpc.git synced 2025-12-29 07:40:14 +01:00
Code Issues Packages Projects Releases Wiki Activity

Address Clippy lints

Browse Source
This commit is contained in:
Tim Kuehn
2018-11-06 17:00:15 -08:00
parent 2c7c64841f
commit 4d2d3f24c6
7 changed files with 31 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files

23
rpc/src/client/channel.rs
View File

@@ -6,6 +6,7 @@
use crate::{
context,
PollIo,
util::{deadline_compat, AsDuration, Compact},
ClientMessage, ClientMessageKind, Request, Response, Transport,
};
@@ -62,12 +63,12 @@ impl<Req, Resp> Clone for Channel<Req, Resp> {
#[derive(Debug)]
#[must_use = "futures do nothing unless polled"]
struct Send<'a, Req, Resp> {
fut: MapOkDispatchResponse<
MapErrConnectionReset<futures::sink::Send<'a, mpsc::Sender<DispatchRequest<Req, Resp>>>>,
Resp,
>,
fut: MapOkDispatchResponse<SendMapErrConnectionReset<'a, Req, Resp>, Resp>,
}
type SendMapErrConnectionReset<'a, Req, Resp> =
MapErrConnectionReset<futures::sink::Send<'a, mpsc::Sender<DispatchRequest<Req, Resp>>>>;
impl<'a, Req, Resp> Send<'a, Req, Resp> {
unsafe_pinned!(
fut: MapOkDispatchResponse<
@@ -109,7 +110,7 @@ impl<'a, Req, Resp> Future for Call<'a, Req, Resp> {
impl<Req, Resp> Channel<Req, Resp> {
/// Sends a request to the dispatch task to forward to the server, returning a [`Future`] that
/// resolves when the request is sent (not when the response is received).
fn send<'a>(&'a mut self, mut ctx: context::Context, request: Req) -> Send<'a, Req, Resp> {
fn send(&mut self, mut ctx: context::Context, request: Req) -> Send<Req, Resp> {
// Convert the context to the call context.
ctx.trace_context.parent_id = Some(ctx.trace_context.span_id);
ctx.trace_context.span_id = SpanId::random(&mut rand::thread_rng());
@@ -150,7 +151,7 @@ impl<Req, Resp> Channel<Req, Resp> {
/// Sends a request to the dispatch task to forward to the server, returning a [`Future`] that
/// resolves to the response.
pub fn call<'a>(&'a mut self, context: context::Context, request: Req) -> Call<'a, Req, Resp> {
pub fn call(&mut self, context: context::Context, request: Req) -> Call<Req, Resp> {
Call {
fut: AndThenIdent::new(self.send(context, request)),
}
@@ -317,7 +318,7 @@ where
unsafe_pinned!(pending_requests: Fuse<mpsc::Receiver<DispatchRequest<Req, Resp>>>);
unsafe_pinned!(transport: Fuse<C>);
fn pump_read(self: &mut Pin<&mut Self>, waker: &LocalWaker) -> Poll<Option<io::Result<()>>> {
fn pump_read(self: &mut Pin<&mut Self>, waker: &LocalWaker) -> PollIo<()> {
Poll::Ready(match ready!(self.transport().poll_next(waker)?) {
Some(response) => {
self.complete(response);
@@ -330,7 +331,7 @@ where
})
}
fn pump_write(self: &mut Pin<&mut Self>, waker: &LocalWaker) -> Poll<Option<io::Result<()>>> {
fn pump_write(self: &mut Pin<&mut Self>, waker: &LocalWaker) -> PollIo<()> {
enum ReceiverStatus {
NotReady,
Closed,
@@ -374,7 +375,7 @@ where
fn poll_next_request(
self: &mut Pin<&mut Self>,
waker: &LocalWaker,
) -> Poll<Option<io::Result<DispatchRequest<Req, Resp>>>> {
) -> PollIo<DispatchRequest<Req, Resp>> {
if self.in_flight_requests().len() >= self.config.max_in_flight_requests {
info!(
"At in-flight request capacity ({}/{}).",
@@ -417,7 +418,7 @@ where
fn poll_next_cancellation(
self: &mut Pin<&mut Self>,
waker: &LocalWaker,
) -> Poll<Option<io::Result<(context::Context, u64)>>> {
) -> PollIo<(context::Context, u64)> {
while let Poll::Pending = self.transport().poll_ready(waker)? {
ready!(self.transport().poll_flush(waker)?);
}
@@ -481,7 +482,7 @@ where
};
self.transport().start_send(cancel)?;
trace!("[{}/{}] Cancel message sent.", trace_id, self.server_addr());
return Ok(());
Ok(())
}
/// Sends a server response to the client task that initiated the associated request.

4
rpc/src/lib.rs
View File

@@ -43,7 +43,7 @@ pub(crate) mod util;
pub use crate::{client::Client, server::Server, transport::Transport};
use futures::{
task::{Spawn, SpawnError, SpawnExt},
task::{Poll, Spawn, SpawnError, SpawnExt},
Future,
};
use std::{cell::RefCell, io, sync::Once, time::SystemTime};
@@ -162,6 +162,8 @@ impl<T> Request<T> {
}
}
pub(crate) type PollIo<T> = Poll<Option<io::Result<T>>>;
static INIT: Once = Once::new();
static mut SEED_SPAWN: Option<Box<dyn CloneSpawn>> = None;
thread_local! {

5
rpc/src/server/filter.rs
View File

@@ -5,6 +5,7 @@
// https://opensource.org/licenses/MIT.
use crate::{
PollIo,
server::{Channel, Config},
util::Compact,
ClientMessage, Response, Transport,
@@ -200,7 +201,7 @@ impl<S, Req, Resp> ConnectionFilter<S, Req, Resp> {
fn poll_listener<C>(
self: &mut Pin<&mut Self>,
cx: &LocalWaker,
) -> Poll<Option<io::Result<NewConnection<Req, Resp, C>>>>
) -> PollIo<NewConnection<Req, Resp, C>>
where
S: Stream<Item = Result<C, io::Error>>,
C: Transport<Item = ClientMessage<Req>, SinkItem = Response<Resp>> + Send,
@@ -232,7 +233,7 @@ where
fn poll_next(
mut self: Pin<&mut Self>,
cx: &LocalWaker,
) -> Poll<Option<io::Result<Channel<Req, Resp, T>>>> {
) -> PollIo<Channel<Req, Resp, T>> {
loop {
match (self.poll_listener(cx)?, self.poll_closed_connections(cx)?) {
(Poll::Ready(Some(NewConnection::Accepted(channel))), _) => {

18
rpc/src/server/mod.rs
View File

@@ -8,7 +8,7 @@
use crate::{
context::Context, util::deadline_compat, util::AsDuration, util::Compact, ClientMessage,
ClientMessageKind, Request, Response, ServerError, Transport,
ClientMessageKind, PollIo, Request, Response, ServerError, Transport,
};
use fnv::FnvHashMap;
use futures::{
@@ -150,7 +150,7 @@ where
}
}
info!("Server shutting down.");
return Poll::Ready(());
Poll::Ready(())
}
}
@@ -247,7 +247,7 @@ where
pub(crate) fn poll_next(
self: &mut Pin<&mut Self>,
cx: &LocalWaker,
) -> Poll<Option<io::Result<ClientMessage<Req>>>> {
) -> PollIo<ClientMessage<Req>> {
self.transport().poll_next(cx)
}
@@ -336,7 +336,7 @@ where
Poll::Ready(Ok(()))
}
fn pump_read(self: &mut Pin<&mut Self>, cx: &LocalWaker) -> Poll<Option<io::Result<()>>> {
fn pump_read(self: &mut Pin<&mut Self>, cx: &LocalWaker) -> PollIo<()> {
ready!(self.poll_ready_if_throttling(cx)?);
Poll::Ready(match ready!(self.channel().poll_next(cx)?) {
@@ -362,7 +362,7 @@ where
self: &mut Pin<&mut Self>,
cx: &LocalWaker,
read_half_closed: bool,
) -> Poll<Option<io::Result<()>>> {
) -> PollIo<()> {
match self.poll_next_response(cx)? {
Poll::Ready(Some((_, response))) => {
self.channel().start_send(response)?;
@@ -392,7 +392,7 @@ where
fn poll_next_response(
self: &mut Pin<&mut Self>,
cx: &LocalWaker,
) -> Poll<Option<io::Result<(Context, Response<Resp>)>>> {
) -> PollIo<(Context, Response<Resp>)> {
// Ensure there's room to write a response.
while let Poll::Pending = self.channel().poll_ready(cx)? {
ready!(self.channel().poll_flush(cx)?);
@@ -402,7 +402,7 @@ where
match ready!(self.pending_responses().poll_next(cx)) {
Some((ctx, response)) => {
if let Some(_) = self.in_flight_requests().remove(&response.request_id) {
if self.in_flight_requests().remove(&response.request_id).is_some() {
self.in_flight_requests().compact(0.1);
}
trace!(
@@ -411,7 +411,7 @@ where
peer,
self.in_flight_requests().len(),
);
return Poll::Ready(Some(Ok((ctx, response))));
Poll::Ready(Some(Ok((ctx, response))))
}
None => {
// This branch likely won't happen, since the ClientHandler is holding a Sender.
@@ -467,7 +467,7 @@ where
let mut response_tx = self.responses_tx().clone();
let trace_id = *ctx.trace_id();
let response = self.f().clone()(ctx.clone(), request);
let response = self.f().clone()(ctx, request);
let response = deadline_compat::Deadline::new(response, Instant::now() + timeout).then(
async move |result| {
let response = Response {

4
rpc/src/transport/channel.rs
View File

@@ -6,7 +6,7 @@
//! Transports backed by in-memory channels.
use crate::Transport;
use crate::{PollIo, Transport};
use futures::{channel::mpsc, task::LocalWaker, Poll, Sink, Stream};
use pin_utils::unsafe_pinned;
use std::pin::Pin;
@@ -45,7 +45,7 @@ impl<Item, SinkItem> UnboundedChannel<Item, SinkItem> {
impl<Item, SinkItem> Stream for UnboundedChannel<Item, SinkItem> {
type Item = Result<Item, io::Error>;
fn poll_next(mut self: Pin<&mut Self>, cx: &LocalWaker) -> Poll<Option<io::Result<Item>>> {
fn poll_next(mut self: Pin<&mut Self>, cx: &LocalWaker) -> PollIo<Item> {
self.rx().poll_next(cx).map(|option| option.map(Ok))
}
}

1
rpc/src/util/serde.rs
View File

@@ -31,6 +31,7 @@ where
}
/// Serializes [`io::ErrorKind`] as a `u32`.
#[allow(clippy::trivially_copy_pass_by_ref)] // Exact fn signature required by serde derive
pub fn serialize_io_error_kind_as_u32<S>(
kind: &io::ErrorKind,
serializer: S,

2
trace/src/lib.rs
View File

@@ -80,7 +80,7 @@ impl TraceId {
/// Returns a random trace ID that can be assumed to be globally unique if `rng` generates
/// actually-random numbers.
pub fn random<R: Rng>(rng: &mut R) -> Self {
TraceId((rng.next_u64() as u128) << mem::size_of::<u64>() | rng.next_u64() as u128)
TraceId(u128::from(rng.next_u64()) << mem::size_of::<u64>() | u128::from(rng.next_u64()))
}
}