This works:
// for demonstrating the more general case, I'm using a non-Copy type here
let path = PathBuf::from("your path here");
let channel = Endpoint::try_from("http://[::]:50051")?
.connect_with_connector(service_fn(move |_: Uri| {
let path_shared = Arc::new(path.clone());
async move {
// Connect to a Uds socket
Ok::<_, std::io::Error>(TokioIo::new(
UnixStream::connect(path_shared.as_ref()).await?,
))
}
}))
.await?;
Note that depending on what exact behaviour you want, you may wish to move the construction of path_shared to outside of the closure.
Let's see how we got here. We'll start with a naive implementation, and let the compiler guide us.
This is what we want, written as simply as possible:
let path = PathBuf::from("your path here");
let channel = Endpoint::try_from("http://[::]:50051")?
.connect_with_connector(service_fn(|_: Uri| {
async {
// Connect to a Uds socket
Ok::<_, std::io::Error>(TokioIo::new(UnixStream::connect(path).await?))
}
}))
And this doesn't compile:
error[E0525]: expected a closure that implements the `FnMut` trait, but this closure only implements `FnOnce`
--> examples/src/uds/client.rs:26:44
|
26 | .connect_with_connector(service_fn(|_: Uri| {
| ---------------------- ^^^^^^^^ this closure implements `FnOnce`, not `FnMut`
| |
| the requirement to implement `FnMut` derives from here
...
29 | Ok::<_, std::io::Error>(TokioIo::new(UnixStream::connect(path).await?))
| ---- closure is `FnOnce` because it moves the variable `path` out of its environment
|
= note: required for `ServiceFn<{closure@examples/src/uds/client.rs:26:44: 26:52}>` to implement `Service<Uri>`
Okay this makes sense. We moved in an external value and consumed it, of course the closure is FnOnce. Since UnixStream::connect is generic, we will try passing it a &path instead.
Note that the move keyword is not necessary for a closure to capture by value; it only forces capturing by value in cases where it normally wouldn't (ref).
let path = PathBuf::from("your path here");
let channel = Endpoint::try_from("http://[::]:50051")?
.connect_with_connector(service_fn(|_: Uri| {
async {
// Connect to a Uds socket
Ok::<_, std::io::Error>(TokioIo::new(UnixStream::connect(&path).await?))
}
}))
.await?;
error[E0373]: closure may outlive the current function, but it borrows `path`, which is owned by the current function
--> examples/src/uds/client.rs:26:44
|
26 | .connect_with_connector(service_fn(|_: Uri| {
| ^^^^^^^^ may outlive borrowed value `path`
...
29 | Ok::<_, std::io::Error>(TokioIo::new(UnixStream::connect(&path).await?))
| ---- `path` is borrowed here
|
note: function requires argument type to outlive `'static`
--> examples/src/uds/client.rs:26:33
|
26 | .connect_with_connector(service_fn(|_: Uri| {
| _________________________________^
27 | | async {
28 | | // Connect to a Uds socket
29 | | Ok::<_, std::io::Error>(TokioIo::new(UnixStream::connect(&path).await?))
30 | | }
31 | | }))
| |__________^
help: to force the closure to take ownership of `path` (and any other referenced variables), use the `move` keyword
|
26 | .connect_with_connector(service_fn(move |_: Uri| {
| ++++
Okay, lifetime issue, makes sense. The closure could run on a thread long after this function has returned. Let's add move.
let path = PathBuf::from("your path here");
let channel = Endpoint::try_from("http://[::]:50051")?
.connect_with_connector(service_fn(move |_: Uri| {
async {
// Connect to a Uds socket
Ok::<_, std::io::Error>(TokioIo::new(UnixStream::connect(&path).await?))
}
}))
.await?;
error: lifetime may not live long enough
--> examples/src/uds/client.rs:27:13
|
26 | .connect_with_connector(service_fn(move |_: Uri| {
| -------------
| | |
| | return type of closure `{async block@examples/src/uds/client.rs:27:13: 27:18}` contains a lifetime `'2`
| lifetime `'1` represents this closure's body
27 | / async {
28 | | // Connect to a Uds socket
29 | | Ok::<_, std::io::Error>(TokioIo::new(UnixStream::connect(&path).await?))
30 | | }
| |_____________^ returning this value requires that `'1` must outlive `'2`
|
= note: closure implements `Fn`, so references to captured variables can't escape the closure
This error message may appear a bit cryptic, especially the annotations on '1 and '2. But the note at the bottom is telling: we just aren't allowed to keep a reference to data within the closure in our async block. Async blocks capture variables using the same rules as closures (ref), so let's make it async move.
let path = PathBuf::from("your path here");
let channel = Endpoint::try_from("http://[::]:50051")?
.connect_with_connector(service_fn(move |_: Uri| {
async move {
// Connect to a Uds socket
Ok::<_, std::io::Error>(TokioIo::new(UnixStream::connect(&path).await?))
}
}))
.await?;
error[E0525]: expected a closure that implements the `FnMut` trait, but this closure only implements `FnOnce`
--> examples/src/uds/client.rs:26:44
|
26 | .connect_with_connector(service_fn(move |_: Uri| {
| ---------------------- ^^^^^^^^^^^^^ this closure implements `FnOnce`, not `FnMut`
| |
| the requirement to implement `FnMut` derives from here
...
29 | Ok::<_, std::io::Error>(TokioIo::new(UnixStream::connect(&path).await?))
| ---- closure is `FnOnce` because it moves the variable `path` out of its environment
|
= note: required for `ServiceFn<{closure@examples/src/uds/client.rs:26:44: 26:57}>` to implement `Service<Uri>`
Okay, so our closure is FnOnce again. Why? Again the compiler tells us: it's because we moved path into our async block. We can't do that. For the closure to be FnMut (and in fact, Fn), path has to stay in the closure.
So here we hit a T-junction. We can either clone path and move it into the async block, or we can satisfy the ownership requirement using a reference counting smart pointers. We'll do smart pointers here.
let path = PathBuf::from("your path here");
let channel = Endpoint::try_from("http://[::]:50051")?
.connect_with_connector(service_fn(move |_: Uri| {
let path_shared = Rc::new(path);
async move {
// Connect to a Uds socket
Ok::<_, std::io::Error>(TokioIo::new(
UnixStream::connect(path_shared.as_ref()).await?,
))
}
}))
.await?;
error: future cannot be sent between threads safely
--> examples/src/uds/client.rs:28:33
|
28 | .connect_with_connector(service_fn(move |_: Uri| {
| _________________________________^
29 | | let path_shared = Rc::new(path);
30 | | async move {
31 | | // Connect to a Uds socket
... |
35 | | }
36 | | }))
| |__________^ future created by async block is not `Send`
|
= help: within `{async block@examples/src/uds/client.rs:30:13: 30:23}`, the trait `Send` is not implemented for `Rc<PathBuf>`, which is required by `{async block@examples/src/uds/client.rs:30:13: 30:23}: Send`
note: captured value is not `Send`
--> examples/src/uds/client.rs:33:41
|
33 | UnixStream::connect(path_shared.as_ref()).await?,
| ^^^^^^^^^^^ has type `Rc<PathBuf>` which is not `Send`
note: required by a bound in `Endpoint::connect_with_connector`
--> /home/cyq/Repos/Public/tonic/tonic/src/transport/channel/endpoint.rs:368:20
|
364 | pub async fn connect_with_connector<C>(&self, connector: C) -> Result<Channel, Error>
| ---------------------- required by a bound in this associated function
...
368 | C::Future: Send,
| ^^^^ required by this bound in `Endpoint::connect_with_connector`
So Endpoint::connect_with_connector wants a Send future. What does this mean? The documentation of Send tells us: it's a type that can be transferred across thread boundaries (ref). The specific reason it wants this is that tokio by default uses a multi-threaded executor. It is possible to configure it to run single-threaded, thereby alleviating the Send bound, but it's seldom done in practice.
So why isn't our future Send? Again the error tells us: path_shared has type Rc<PathBuf> which is not Send. Why isn't Rc Send? We can find it in the docs:
Rc uses non-atomic reference counting. This means that overhead is very low, but an Rc cannot be sent between threads, and consequently Rc does not implement Send. As a result, the Rust compiler will check at compile time that you are not sending Rcs between threads. If you need multi-threaded, atomic reference counting, use sync::Arc.
So let's use Arc instead.
let path = PathBuf::from("your path here");
let channel = Endpoint::try_from("http://[::]:50051")?
.connect_with_connector(service_fn(move |_: Uri| {
let path_shared = Arc::new(path);
async move {
// Connect to a Uds socket
Ok::<_, std::io::Error>(TokioIo::new(
UnixStream::connect(path_shared.as_ref()).await?,
))
}
}))
.await?;
error[E0525]: expected a closure that implements the `FnMut` trait, but this closure only implements `FnOnce`
--> examples/src/uds/client.rs:26:44
|
26 | .connect_with_connector(service_fn(move |_: Uri| {
| ---------------------- ^^^^^^^^^^^^^ this closure implements `FnOnce`, not `FnMut`
| |
| the requirement to implement `FnMut` derives from here
27 | let path_shared = Arc::new(path);
| ---- closure is `FnOnce` because it moves the variable `path` out of its environment
|
= note: required for `ServiceFn<{closure@examples/src/uds/client.rs:26:44: 26:57}>` to implement `Service<Uri>`
Once again our closure is FnOnce. Why? Recall that we concluded earlier that path has to stay in the closure. But here we have moved it into path_shared. So instead, let's clone path so that it stays put.
let path = PathBuf::from("your path here");
let channel = Endpoint::try_from("http://[::]:50051")?
.connect_with_connector(service_fn(move |_: Uri| {
let path_shared = Arc::new(path.clone());
async move {
// Connect to a Uds socket
Ok::<_, std::io::Error>(TokioIo::new(
UnixStream::connect(path_shared.as_ref()).await?,
))
}
}))
.await?;
And finally we have arrived at our final code.
The biggest characteristic about Rust is that it doesn't let you compile until everything is perfect, and getting to perfection is hard. Some people find it frustrating, but personally I like it because I can rest assured that there's not going to be an urgent call that wakes me up at 0300. But as you can see here, the Rust compiler can often do a tremendous job in guiding you to the right answer, if you follow its advice step by step.
