1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135

//! This crate provides utilities for using protocols that follow certain common patterns on
//! top of [Tokio](https://tokio.rs) and [Tower](https://github.com/tower-rs/tower).
//!
//! # Protocols
//!
//! At a high level, a protocol is a mechanism that lets you take a bunch of requests and turn them
//! into responses. Tower provides the [`Service`](https://docs.rs/tower-service/) trait, which is
//! an interface for mapping requests into responses, but it does not deal with how those requests
//! are sent between clients and servers. Tokio, on the other hand, provides asynchronous
//! communication primitives, but it does not deal with high-level abstractions like services. This
//! crate attempts to bridge that gap.
//!
//! There are many types of protocols in the wild, but they generally come in two forms:
//! *pipelining* and *multiplexing*. A pipelining protocol sends requests and responses in-order
//! between the consumer and provider of a service, and processes requests one at a time. A
//! multiplexing protocol on the other hand constructs requests in such a way that they can be
//! handled and responded to in *any* order while still allowing the client to know which response
//! is for which request. Pipelining and multiplexing both have their advantages and disadvantages;
//! see the module-level documentation for [`pipeline`] and [`multiplex`] for details. There is
//! also good deal of discussion in [this StackOverflow
//! answer](https://softwareengineering.stackexchange.com/a/325888/79642).
//!
//! # Transports
//!
//! A key part of any protocol is its transport, which is the way that it transmits requests and
//! responses. In general, `tokio-tower` leaves the on-the-wire implementations of protocols to
//! other crates (like [`tokio-codec`](https://docs.rs/tokio-codec/) or
//! [`async-bincode`](https://docs.rs/async-bincode)) and instead operates at the level of
//! [`Sink`](https://docs.rs/futures/0.1/futures/sink/trait.Sink.html)s and
//! [`Stream`](https://docs.rs/futures/0.15/futures/stream/trait.Stream.html)s. In particular, it
//! assumes that there exists a `Sink + Stream` transport where it can send `Request`s and receive
//! `Response`s, or vice-versa for the server side.
//!
//! # Servers and clients
//!
//! This crate provides utilities that make writing both clients and servers easier. You'll find
//! the client helper as `Client` in the protocol module you're working with (e.g.,
//! [`pipeline::Client`]), and the server helper as `Server` in the same place.
#![deny(missing_docs)]

macro_rules! event {
    ($span:expr, $($rest:tt)*) => {
        #[cfg(feature = "tracing")]
        $span.in_scope(|| tracing::event!($($rest)*));
    };
}

mod error;
mod mediator;
pub(crate) mod wrappers;
pub use error::Error;

use futures_core::{
    future::Future,
    stream::TryStream,
    task::{Context, Poll},
};
use futures_sink::Sink;
use tower_service::Service;

/// Creates new `Transport` (i.e., `Sink + Stream`) instances.
///
/// Acts as a transport factory. This is useful for cases where new `Sink + Stream`
/// values must be produced.
///
/// This is essentially a trait alias for a `Service` of `Sink + Stream`s.
pub trait MakeTransport<Target, Request>: self::sealed::Sealed<Target, Request> {
    /// Items produced by the transport
    type Item;

    /// Errors produced when receiving from the transport
    type Error;

    /// Errors produced when sending to the transport
    type SinkError;

    /// The `Sink + Stream` implementation created by this factory
    type Transport: TryStream<Ok = Self::Item, Error = Self::Error>
        + Sink<Request, Error = Self::SinkError>;

    /// Errors produced while building a transport.
    type MakeError;

    /// The future of the `Service` instance.
    type Future: Future<Output = Result<Self::Transport, Self::MakeError>>;

    /// Returns `Ready` when the factory is able to create more transports.
    ///
    /// If the service is at capacity, then `NotReady` is returned and the task
    /// is notified when the service becomes ready again. This function is
    /// expected to be called while on a task.
    ///
    /// This is a **best effort** implementation. False positives are permitted.
    /// It is permitted for the service to return `Ready` from a `poll_ready`
    /// call and the next invocation of `make_transport` results in an error.
    fn poll_ready(&mut self, cx: &mut Context) -> Poll<Result<(), Self::MakeError>>;

    /// Create and return a new transport asynchronously.
    fn make_transport(&mut self, target: Target) -> Self::Future;
}

impl<M, T, Target, Request> self::sealed::Sealed<Target, Request> for M
where
    M: Service<Target, Response = T>,
    T: TryStream + Sink<Request>,
{
}

impl<M, T, Target, Request> MakeTransport<Target, Request> for M
where
    M: Service<Target, Response = T>,
    T: TryStream + Sink<Request>,
{
    type Item = <T as TryStream>::Ok;
    type Error = <T as TryStream>::Error;
    type SinkError = <T as Sink<Request>>::Error;
    type Transport = T;
    type MakeError = M::Error;
    type Future = M::Future;

    fn poll_ready(&mut self, cx: &mut Context) -> Poll<Result<(), Self::MakeError>> {
        Service::poll_ready(self, cx)
    }

    fn make_transport(&mut self, target: Target) -> Self::Future {
        Service::call(self, target)
    }
}

mod sealed {
    pub trait Sealed<A, B> {}
}

pub mod multiplex;
pub mod pipeline;