Revert "🧵 Service::handle now blocking"

This reverts commit 2491fea.

When implementing the async handling, I realized how limiting that is
w/o having the whole handler async. With the reverted change, a lot of
async code will be forced to use interior mutability and method handling
being non-async, would mean that either handlers will be limited to sync
locks (which isn't what you always want in async code) or having to spawn
tasks for all method handling. This especially sucks given that the whole
Server machinery is completely single-threaded itself.

Moreover, it looks very likely that the reason we can't just
`tokio::spawn` `Server::run` is not because the future is always `!Send`
but rather that the compiler is unable to determine that to be the case
because of known issue[1].

[1]: rust-lang/rust#100013

Author: zeenix

zlink-tokio/tests/lowlevel-ftl.rs

@@ -1,7 +1,7 @@
 use std::time::Duration;
 
 use serde::{Deserialize, Serialize};
-use tokio::{spawn, time::sleep};
+use tokio::{select, spawn, time::sleep};
 use zlink::connection::Reply;
 use zlink_tokio::{
     connection::Call,
@@ -36,8 +36,15 @@ async fn lowlevel_ftl() -> Result<(), Box<dyn std::error::Error>> {
     let listener = bind(SOCKET_PATH).unwrap();
     let service = Ftl::new(conditions[0]);
     let server = zlink_tokio::Server::new(listener, service);
-    spawn(server.run());
+    select! {
+        _ = server.run() => {},
+        _ = run_client(&conditions) => {}
+    }
+
+    Ok(())
+}
 
+async fn run_client(conditions: &[DriveCondition]) -> Result<(), Box<dyn std::error::Error>> {
     // Now create a client connection that monitor changes in the drive condition.
     let mut drive_monitor_conn = connect(SOCKET_PATH).await?;
     drive_monitor_conn
@@ -154,7 +161,7 @@ impl Service for Ftl {
     type ReplyStreamParams = Replies;
     type ReplyError<'ser> = Errors;
 
-    fn handle<'ser>(
+    async fn handle<'ser>(
         &'ser mut self,
         call: Call<Self::MethodCall<'_>>,
     ) -> MethodReply<Self::ReplyParams<'ser>, Self::ReplyStream, Self::ReplyError<'ser>> {

zlink/src/server/mod.rs

@@ -165,7 +165,7 @@ where
         writer: &mut WriteConnection<<Listener::Socket as Socket>::WriteHalf>,
     ) -> crate::Result<Option<Service::ReplyStream>> {
         let mut stream = None;
-        match self.service.handle(call) {
+        match self.service.handle(call).await {
             MethodReply::Single(params) => {
                 writer
                     .send_reply(Reply::new(params).set_continues(Some(false)))

zlink/src/server/service.rs

@@ -1,6 +1,6 @@
 //! Serice-related API.
 
-use core::fmt::Debug;
+use core::{fmt::Debug, future::Future};
 
 use futures_util::Stream;
 use serde::{Deserialize, Serialize};
@@ -40,14 +40,12 @@ pub trait Service {
         Self: 'ser;
 
     /// Handle a method call.
-    ///
-    /// While this methos is no async, it can return a reply stream that can be used to
-    /// ascynchronously send out replies. The main use case for allowing to return a stream is
-    /// multiple replies but it can be used for single reply as well.
     fn handle<'ser>(
         &'ser mut self,
         method: Call<Self::MethodCall<'_>>,
-    ) -> MethodReply<Self::ReplyParams<'ser>, Self::ReplyStream, Self::ReplyError<'ser>>;
+    ) -> impl Future<
+        Output = MethodReply<Self::ReplyParams<'ser>, Self::ReplyStream, Self::ReplyError<'ser>>,
+    >;
 }
 
 /// A service method call reply.