Skip to content

Commit

Permalink
refactor: relax Sync bounds
Browse files Browse the repository at this point in the history 
Signed-off-by: Roman Volosatovs <rvolosatovs@riseup.net>
  • Loading branch information
@rvolosatovs
rvolosatovs committed Jul 9, 2024
1 parent 7a86248 commit ed1e162
Show file tree
Hide file tree
Showing 10 changed files with 77 additions and 103 deletions.
8 changes: 4 additions & 4 deletions Cargo.lock
View file

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

8 changes: 4 additions & 4 deletions Cargo.toml
View file
Original file line number Diff line number Diff line change
Expand Up @@ -128,16 +128,16 @@ wasmtime = { version = "22", default-features = false }
wasmtime-wasi = { version = "22", default-features = false }
wit-bindgen = { version = "0.27", default-features = false }
wit-bindgen-core = { version = "0.27", default-features = false }
wit-bindgen-wrpc = { version = "0.4.4", default-features = false, path = "./crates/wit-bindgen" }
wit-bindgen-wrpc = { version = "0.4.5", default-features = false, path = "./crates/wit-bindgen" }
wit-bindgen-wrpc-go = { version = "0.2", default-features = false, path = "./crates/wit-bindgen-go" }
wit-bindgen-wrpc-rust = { version = "0.4.4", default-features = false, path = "./crates/wit-bindgen-rust" }
wit-bindgen-wrpc-rust-macro = { version = "0.4.4", default-features = false, path = "./crates/wit-bindgen-rust-macro" }
wit-bindgen-wrpc-rust = { version = "0.4.5", default-features = false, path = "./crates/wit-bindgen-rust" }
wit-bindgen-wrpc-rust-macro = { version = "0.4.5", default-features = false, path = "./crates/wit-bindgen-rust-macro" }
wit-component = { version = "0.212", default-features = false }
wit-parser = { version = "0.212", default-features = false }
wrpc-cli = { version = "0.2", path = "./crates/cli", default-features = false }
wrpc-introspect = { version = "0.1", default-features = false, path = "./crates/introspect" }
wrpc-runtime-wasmtime = { version = "0.19", path = "./crates/runtime-wasmtime", default-features = false }
wrpc-transport = { version = "0.26", path = "./crates/transport", default-features = false }
wrpc-transport = { version = "0.26.3", path = "./crates/transport", default-features = false }
wrpc-transport-nats = { version = "0.22", path = "./crates/transport-nats", default-features = false }
wrpc-transport-quic = { version = "0.1", path = "./crates/transport-quic", default-features = false }
wrpc-wasmtime-nats-cli = { version = "0.3", path = "./crates/wasmtime-nats-cli", default-features = false }
2 changes: 1 addition & 1 deletion crates/transport/Cargo.toml
View file
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
[package]
name = "wrpc-transport"
version = "0.26.2"
version = "0.26.3"
description = "wRPC core transport functionality"

authors.workspace = true
Expand Down
24 changes: 6 additions & 18 deletions crates/transport/src/lib.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -282,21 +282,16 @@ pub trait Serve: Sync {
Item = anyhow::Result<(
Self::Context,
Params,
Option<
impl Future<Output = std::io::Result<()>> + Sync + Send + Unpin + 'static,
>,
impl FnOnce(
Results,
)
-> Pin<Box<dyn Future<Output = anyhow::Result<()>> + Sync + Send>>,
Option<impl Future<Output = std::io::Result<()>> + Send + Unpin + 'static>,
impl FnOnce(Results) -> Pin<Box<dyn Future<Output = anyhow::Result<()>> + Send>>,
)>,
> + Send
+ 'static,
>,
> + Send
where
Params: TupleDecode<Self::Incoming> + Send + Sync + 'static,
Results: TupleEncode<Self::Outgoing> + Send + Sync + 'static,
Params: TupleDecode<Self::Incoming> + Send + 'static,
Results: TupleEncode<Self::Outgoing> + Send + 'static,
<Params::Decoder as tokio_util::codec::Decoder>::Error:
std::error::Error + Send + Sync + 'static,
<Results::Encoder as tokio_util::codec::Encoder<Results>>::Error:
Expand Down Expand Up @@ -368,13 +363,7 @@ mod tests {
#[allow(clippy::manual_async_fn)]
fn invoke_values_send<T>() -> impl Future<
Output = anyhow::Result<(
Pin<
Box<
dyn Stream<Item = Vec<Pin<Box<dyn Future<Output = String> + Send + Sync>>>>
+ Send
+ Sync,
>,
>,
Pin<Box<dyn Stream<Item = Vec<Pin<Box<dyn Future<Output = String> + Send>>>> + Send>>,
)>,
> + Send
where
Expand Down Expand Up @@ -404,8 +393,7 @@ mod tests {
i.invoke(cx, "foo", "bar", Bytes::default(), &paths).await
}

async fn call_invoke_async<T>(
) -> anyhow::Result<(Pin<Box<dyn Stream<Item = Bytes> + Send + Sync>>,)>
async fn call_invoke_async<T>() -> anyhow::Result<(Pin<Box<dyn Stream<Item = Bytes> + Send>>,)>
where
T: Invoke<Context = ()> + Default,
{
Expand Down
82 changes: 37 additions & 45 deletions crates/transport/src/value.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -156,12 +156,8 @@ impl<T: ?Sized> ResourceOwn<T> {
}

pub type DeferredFn<T> = Box<
dyn FnOnce(
Arc<T>,
Vec<usize>,
) -> Pin<Box<dyn Future<Output = std::io::Result<()>> + Send + Sync>>
+ Send
+ Sync,
dyn FnOnce(Arc<T>, Vec<usize>) -> Pin<Box<dyn Future<Output = std::io::Result<()>> + Send>>
+ Send,
>;

pub trait Deferred<T> {
Expand Down Expand Up @@ -325,7 +321,7 @@ where
}

pub trait Encode<T>: Sized {
type Encoder: tokio_util::codec::Encoder<Self> + Deferred<T> + Default + Send + Sync;
type Encoder: tokio_util::codec::Encoder<Self> + Deferred<T> + Default + Send;

#[instrument(level = "trace", skip(self, enc))]
fn encode(
Expand Down Expand Up @@ -752,9 +748,9 @@ where

impl<T, R> Decode<R> for Vec<T>
where
T: Decode<R> + Send + Sync,
T: Decode<R> + Send,
T::ListDecoder: Deferred<R> + Send,
R: crate::Index<R> + Send + Sync + 'static,
R: crate::Index<R> + Send + 'static,
{
type Decoder = T::ListDecoder;
type ListDecoder = ListDecoder<Self::Decoder, R>;
Expand Down Expand Up @@ -1119,7 +1115,7 @@ impl<R, T: ?Sized> Deferred<R> for CoreVecDecoder<ResourceBorrowDecoder<T>> {
}
}

impl<R, T: ?Sized + Send + Sync + 'static> Decode<R> for ResourceBorrow<T> {
impl<R, T: ?Sized + Send + 'static> Decode<R> for ResourceBorrow<T> {
type Decoder = ResourceBorrowDecoder<T>;
type ListDecoder = CoreVecDecoder<Self::Decoder>;
}
Expand Down Expand Up @@ -1163,7 +1159,7 @@ impl<R, T: ?Sized> Deferred<R> for CoreVecDecoder<ResourceOwnDecoder<T>> {
}
}

impl<R, T: ?Sized + Send + Sync + 'static> Decode<R> for ResourceOwn<T> {
impl<R, T: ?Sized + Send + 'static> Decode<R> for ResourceOwn<T> {
type Decoder = ResourceOwnDecoder<T>;
type ListDecoder = CoreVecDecoder<Self::Decoder>;
}
Expand Down Expand Up @@ -1304,8 +1300,8 @@ macro_rules! impl_tuple_codec {
R: crate::Index<R> + Send + Sync + 'static,
E: From<std::io::Error>,
$(
$vt: Decode<R> + Send + Sync + 'static,
$vt::Decoder: tokio_util::codec::Decoder<Error = E> + Send + Sync + 'static,
$vt: Decode<R> + Send + 'static,
$vt::Decoder: tokio_util::codec::Decoder<Error = E> + Send + 'static,
)+
{
type Decoder = TupleDecoder::<($($vt::Decoder),+,), ($(Option<$vt>),+,)>;
Expand All @@ -1317,8 +1313,8 @@ macro_rules! impl_tuple_codec {
R: crate::Index<R> + Send + Sync + 'static,
E: From<std::io::Error>,
$(
$vt: Decode<R> + Send + Sync + 'static,
$vt::Decoder: tokio_util::codec::Decoder<Error = E> + Send + Sync + 'static,
$vt: Decode<R> + Send + 'static,
$vt::Decoder: tokio_util::codec::Decoder<Error = E> + Send + 'static,
)+
{
}
Expand Down Expand Up @@ -1457,7 +1453,7 @@ impl<T, W, Fut> tokio_util::codec::Encoder<Fut> for FutureEncoder<W>
where
T: Encode<W>,
W: AsyncWrite + crate::Index<W> + Send + Sync + Unpin + 'static,
Fut: Future<Output = T> + Send + Sync + 'static,
Fut: Future<Output = T> + Send + 'static,
std::io::Error: From<<T::Encoder as tokio_util::codec::Encoder<T>>::Error>,
{
type Error = std::io::Error;
Expand Down Expand Up @@ -1489,7 +1485,7 @@ where
}
}

impl<T, W> Encode<W> for Pin<Box<dyn Future<Output = T> + Send + Sync>>
impl<T, W> Encode<W> for Pin<Box<dyn Future<Output = T> + Send>>
where
T: Encode<W> + 'static,
W: AsyncWrite + crate::Index<W> + Send + Sync + Unpin + 'static,
Expand Down Expand Up @@ -1531,12 +1527,11 @@ where

impl<T, R> tokio_util::codec::Decoder for FutureDecoder<T, R>
where
T: Decode<R> + Send + Sync + 'static,
T::Decoder: Sync,
T: Decode<R> + Send + 'static,
R: AsyncRead + crate::Index<R> + Send + Sync + Unpin + 'static,
std::io::Error: From<<T::Decoder as tokio_util::codec::Decoder>::Error>,
{
type Item = Pin<Box<dyn Future<Output = T> + Send + Sync>>;
type Item = Pin<Box<dyn Future<Output = T> + Send>>;
type Error = <T::Decoder as tokio_util::codec::Decoder>::Error;

#[instrument(level = "trace", skip(self), fields(ty = "future"))]
Expand Down Expand Up @@ -1588,10 +1583,9 @@ where
}
}

impl<T, R> Decode<R> for Pin<Box<dyn Future<Output = T> + Send + Sync>>
impl<T, R> Decode<R> for Pin<Box<dyn Future<Output = T> + Send>>
where
T: Decode<R> + Send + Sync + 'static,
T::Decoder: Sync,
T: Decode<R> + Send + 'static,
R: AsyncRead + crate::Index<R> + Send + Sync + Unpin + 'static,
std::io::Error: From<<T::Decoder as tokio_util::codec::Decoder>::Error>,
{
Expand All @@ -1617,9 +1611,9 @@ impl<W> Deferred<W> for StreamEncoder<W> {

impl<T, W, S> tokio_util::codec::Encoder<S> for StreamEncoder<W>
where
T: Encode<W> + Send + Sync + 'static,
T: Encode<W> + Send + 'static,
W: AsyncWrite + crate::Index<W> + Send + Sync + Unpin + 'static,
S: Stream<Item = Vec<T>> + Send + Sync + Unpin + 'static,
S: Stream<Item = Vec<T>> + Send + Unpin + 'static,
std::io::Error: From<<T::Encoder as tokio_util::codec::Encoder<T>>::Error>,
{
type Error = std::io::Error;
Expand Down Expand Up @@ -1689,9 +1683,9 @@ where
}
}

impl<T, W> Encode<W> for Pin<Box<dyn Stream<Item = Vec<T>> + Send + Sync>>
impl<T, W> Encode<W> for Pin<Box<dyn Stream<Item = Vec<T>> + Send>>
where
T: Encode<W> + Send + Sync + 'static,
T: Encode<W> + Send + 'static,
W: AsyncWrite + crate::Index<W> + Send + Sync + Unpin + 'static,
std::io::Error: From<<T::Encoder as tokio_util::codec::Encoder<T>>::Error>,
{
Expand All @@ -1717,7 +1711,7 @@ impl<W> Deferred<W> for StreamEncoderBytes<W> {
impl<W, S> tokio_util::codec::Encoder<S> for StreamEncoderBytes<W>
where
W: AsyncWrite + crate::Index<W> + Send + Sync + Unpin + 'static,
S: Stream<Item = Bytes> + Send + Sync + Unpin + 'static,
S: Stream<Item = Bytes> + Send + Unpin + 'static,
{
type Error = std::io::Error;

Expand Down Expand Up @@ -1758,7 +1752,7 @@ where
}
}

impl<W> Encode<W> for Pin<Box<dyn Stream<Item = Bytes> + Send + Sync>>
impl<W> Encode<W> for Pin<Box<dyn Stream<Item = Bytes> + Send>>
where
W: AsyncWrite + crate::Index<W> + Send + Sync + Unpin + 'static,
{
Expand All @@ -1784,7 +1778,7 @@ impl<W> Deferred<W> for StreamEncoderRead<W> {
impl<W, S> tokio_util::codec::Encoder<S> for StreamEncoderRead<W>
where
W: AsyncWrite + crate::Index<W> + Send + Sync + Unpin + 'static,
S: AsyncRead + Send + Sync + Unpin + 'static,
S: AsyncRead + Send + Unpin + 'static,
{
type Error = std::io::Error;

Expand Down Expand Up @@ -1828,7 +1822,7 @@ where
}
}

impl<W> Encode<W> for Pin<Box<dyn AsyncRead + Send + Sync>>
impl<W> Encode<W> for Pin<Box<dyn AsyncRead + Send>>
where
W: AsyncWrite + crate::Index<W> + Send + Sync + Unpin + 'static,
{
Expand All @@ -1837,7 +1831,7 @@ where

impl<T, W> Encode<W> for std::io::Cursor<T>
where
T: AsRef<[u8]> + Send + Sync + Unpin + 'static,
T: AsRef<[u8]> + Send + Unpin + 'static,
W: AsyncWrite + crate::Index<W> + Send + Sync + Unpin + 'static,
{
type Encoder = StreamEncoderRead<W>;
Expand Down Expand Up @@ -1930,7 +1924,7 @@ async fn handle_deferred_stream<C, T, R>(
) -> std::io::Result<()>
where
C: tokio_util::codec::Decoder<Item = T> + Deferred<R>,
R: AsyncRead + crate::Index<R> + Send + Sync + Unpin + 'static,
R: AsyncRead + crate::Index<R> + Send + Unpin + 'static,
std::io::Error: From<C::Error>,
{
let dec = ListDecoder::new(dec);
Expand Down Expand Up @@ -1982,14 +1976,13 @@ where

impl<T, R> tokio_util::codec::Decoder for StreamDecoder<T, R>
where
T: Decode<R> + Send + Sync + 'static,
T::Decoder: Sync,
T: Decode<R> + Send + 'static,
T::ListDecoder: Deferred<R>,
R: AsyncRead + crate::Index<R> + Send + Sync + Unpin + 'static,
<T::Decoder as tokio_util::codec::Decoder>::Error: Send + Sync,
<T::Decoder as tokio_util::codec::Decoder>::Error: Send,
std::io::Error: From<<T::Decoder as tokio_util::codec::Decoder>::Error>,
{
type Item = Pin<Box<dyn Stream<Item = Vec<T>> + Send + Sync>>;
type Item = Pin<Box<dyn Stream<Item = Vec<T>> + Send>>;
type Error = <<T as Decode<R>>::ListDecoder as tokio_util::codec::Decoder>::Error;

#[instrument(level = "trace", skip(self), fields(ty = "stream"))]
Expand All @@ -2013,13 +2006,12 @@ where
}
}

impl<T, R> Decode<R> for Pin<Box<dyn Stream<Item = Vec<T>> + Send + Sync>>
impl<T, R> Decode<R> for Pin<Box<dyn Stream<Item = Vec<T>> + Send>>
where
T: Decode<R> + Send + Sync + 'static,
T::Decoder: Sync,
T: Decode<R> + Send + 'static,
T::ListDecoder: Deferred<R> + Send,
R: AsyncRead + crate::Index<R> + Send + Sync + Unpin + 'static,
<T::Decoder as tokio_util::codec::Decoder>::Error: Send + Sync,
<T::Decoder as tokio_util::codec::Decoder>::Error: Send,
std::io::Error: From<<T::Decoder as tokio_util::codec::Decoder>::Error>,
{
type Decoder = StreamDecoder<T, R>;
Expand Down Expand Up @@ -2050,7 +2042,7 @@ impl<R> tokio_util::codec::Decoder for StreamDecoderBytes<R>
where
R: AsyncRead + crate::Index<R> + Send + Sync + Unpin + 'static,
{
type Item = Pin<Box<dyn Stream<Item = Bytes> + Send + Sync>>;
type Item = Pin<Box<dyn Stream<Item = Bytes> + Send>>;
type Error = std::io::Error;

#[instrument(level = "trace", skip(self), fields(ty = "stream<u8>"))]
Expand Down Expand Up @@ -2091,7 +2083,7 @@ where
}
}

impl<R> Decode<R> for Pin<Box<dyn Stream<Item = Bytes> + Send + Sync>>
impl<R> Decode<R> for Pin<Box<dyn Stream<Item = Bytes> + Send>>
where
R: AsyncRead + crate::Index<R> + Send + Sync + Unpin + 'static,
{
Expand Down Expand Up @@ -2123,7 +2115,7 @@ impl<R> tokio_util::codec::Decoder for StreamDecoderRead<R>
where
R: AsyncRead + crate::Index<R> + Send + Sync + Unpin + 'static,
{
type Item = Pin<Box<dyn AsyncRead + Send + Sync>>;
type Item = Pin<Box<dyn AsyncRead + Send>>;
type Error = std::io::Error;

#[instrument(level = "trace", skip(self), fields(ty = "stream<u8>"))]
Expand Down Expand Up @@ -2164,7 +2156,7 @@ where
}
}

impl<R> Decode<R> for Pin<Box<dyn AsyncRead + Send + Sync>>
impl<R> Decode<R> for Pin<Box<dyn AsyncRead + Send>>
where
R: AsyncRead + crate::Index<R> + Send + Sync + Unpin + 'static,
{
Expand Down
2 changes: 1 addition & 1 deletion crates/wit-bindgen-rust-macro/Cargo.toml
View file
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
[package]
name = "wit-bindgen-wrpc-rust-macro"
version = "0.4.4"
version = "0.4.5"
description = """
Procedural macro paired with the `wit-bindgen-wrpc` crate.
"""
Expand Down
Loading

0 comments on commit ed1e162

Please sign in to comment.