How to introduce trait indirection with futures?

Question

I have a tokio tcp server which should hand individual incoming connections to a service. How do I correctly handle indirection so that the server can work with different service implementations? I have resorted to calling tokio::spawn() inside the service because I couldn't find a way to return a future with indirection.

Here's a minimal example of what I'm doing:

extern crate tokio;

use tokio::prelude::future::FutureResult;
use tokio::prelude::*;

struct Subject {
    name: String,
}    
struct MySvc {
    name: String,
}    
trait Svc6 {
    fn handle6(&self, subject: Subject);
}
impl Svc6 for MySvc {
    fn handle6(&self, subject: Subject) {
        let task = future::ok((self.name.to_string(), subject))
            .and_then(|(n, s)| Ok(println!("#6. Hi {}! My name is {}.", s.name, n)));
        tokio::spawn(task);
    }
}   

#[test]
fn svc6_works() {
    let svc = MySvc {
        name: "Zorg".into(),
    };
    let subj = Subject {
        name: "Gandalf".into(),
    };
    tokio::run(future::ok(svc).and_then(|s| Ok(s.handle6(subj))));
}

While this works with indirection, I'm concerned if I'm using tokio correctly. Each Svc6 impl has to call tokio::spawn() rather than just returning a task. I'd also prefer if the server handles spawning as it may need to deal with prioritization and queueing. It is also hard to test a method that doesn't return anything.

Here is a playground link of the other things I've been trying. To see the full context, go to Samotop source and the accept fn.

It would be great if trait method implementation could return impl Trait!

trait Svc1 {
    fn handle1(&self, subject: Subject) -> Future<Item = (), Error = ()>;
}
impl Svc1 for MySvc {
    // error[E0562]: `impl Trait` not allowed outside of function and inherent method return types
    fn handle1(&self, subject: Subject) -> impl Future<Item = (), Error = ()> {
        future::ok(println!(
            "#1. Hi {}! My name is {}.",
            subject.name, self.name
        ))
    }
}

Answer 1

There's nothing special about futures or Tokio here, this is just Rust. I'd highly encourage you to learn how to use basic Rust functionality before diving into the complicated world of asynchronous programming. Start with The Rust Programming Language, specicially the section on trait objects:

trait Svc {
    fn handle(&self, subject: Subject) -> Box<Future<Item = (), Error = ()> + Send>;
}

impl Svc for MySvc {
    fn handle(&self, subject: Subject) -> Box<Future<Item = (), Error = ()> + Send> {
        Box::new(future::ok(println!(
            "#1. Hi {}! My name is {}.",
            subject.name, self.name
        )))
    }
}

#[test]
fn svc_works() {
    let svc = MySvc {
        name: "Zorg".into(),
    };
    let subj = Subject {
        name: "Gandalf".into(),
    };
    tokio::run(svc.handle(subj))
}

This is explicitly called out as the number one suggestion of the Tokio documentation on how to return a Future.

if trait method implementation could return impl Trait!

As far as I'm aware, this is impossible. Every function that returns an impl Trait returns a concrete type of a potentially different size. A specific caller would not know how much stack space to allocate for an arbitrary trait implementation.

See also:

• What is the correct way to return an Iterator (or any other trait)?
• Sending trait objects between threads in Rust
• Sharing a struct with trait objects as properties across threads

Answer 2

Returning impl Trait from trait implementations is a no go as I had to learn (again?) that we need a concrete size. Boxing things up might work but I keep pondering. So I turned the thing around a bit and made the Service return a Sink instead, that will receive the item. Then I forward the stream into the sink. It seems this can be wrapped around with tokio::spawn():

use futures::StartSend;
use tokio;
use tokio::io;
use tokio::prelude::*;

struct Subject {
    name: String,
}

trait Svc {
    type Receiver;
    type Error;
    fn start(&self) -> Self::Receiver;
}

struct MySvc {
    name: String,
}

impl Svc for MySvc {
    type Receiver = MyReceiver;
    type Error = io::Error;
    fn start(&self) -> Self::Receiver {
        MyReceiver::new(&self.name)
    }
}

struct MyReceiver {
    name: String,
    pending: Box<Future<Item = (), Error = ()> + Send>,
}

impl MyReceiver {
    fn say_hi(&self, subject: Subject) {
        println!("Hi {}! It's {}.", subject.name, self.name)
    }
    fn new(name: impl ToString) -> Self {
        Self {
            name: name.to_string(),
            pending: Box::new(future::ok(())),
        }
    }
}

impl Future for MyReceiver {
    type Item = Self;
    type Error = Self;
    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
        Ok(Async::Ready(MyReceiver::new(&self.name)))
    }
}

impl Sink for MyReceiver {
    type SinkItem = Subject;
    type SinkError = ();
    fn start_send(&mut self, item: Self::SinkItem) -> StartSend<Self::SinkItem, Self::SinkError> {
        self.say_hi(item);
        Ok(AsyncSink::Ready)
    }
    fn poll_complete(&mut self) -> Poll<(), Self::SinkError> {
        Ok(Async::Ready(()))
    }
}

#[test]
fn try() {
    let svc = MySvc { name: "jOy".into() };

    let task = future::ok(svc)
        .and_then(|s| {
            s.start().and_then(|r| {
                let subject = Subject {
                    name: "Miou".into(),
                };
                let task = stream::once(Ok::<Subject, ()>(subject))
                    .forward(r)
                    .map_err(|_| ())
                    .and_then(|_| Ok(()));
                tokio::spawn(task);
                Ok(())
            })
        })
        .and_then(|_| Ok(()))
        .map_err(|_| ());

    tokio::run(task);
}