multilink 1.0.1

IPC library for communicating with local or remote processes, over stdio or HTTP
Documentation 
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

mod protocol;

use std::{
    task::{Context, Poll},
    time::Duration,
};

use async_stream::stream;
use clap::{command, Parser, Subcommand};
use futures::StreamExt;
use multilink::{
    http::server::{HttpServer, HttpServerConfig},
    stdio::server::StdioServer,
    ServiceError, ServiceFuture, ServiceResponse,
};
use protocol::{GreetingResponse, GreetingStreamResponse, Request, Response};
use tokio::time::sleep;
use tower::Service;
use tracing_subscriber::{filter::LevelFilter, EnvFilter};

#[derive(Debug, Subcommand)]
enum Command {
    /// Run a HTTP server.
    HttpServer,
    /// Run a stdio/json-rpc server.
    StdioServer,
}

/// A server that sends greetings to clients.
#[derive(Parser, Debug)]
#[command(about)]
struct Cli {
    /// The type of server to run.
    #[command(subcommand)]
    server_type: Command,

    /// The port that the HTTP server should listen on.
    #[arg(long, default_value_t = 8080)]
    http_listen_port: u16,
}

#[derive(Clone)]
struct GreetingService;

impl Service<Request> for GreetingService {
    type Response = ServiceResponse<Response>;
    type Error = ServiceError;
    type Future = ServiceFuture<ServiceResponse<Response>>;

    fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
        Poll::Ready(Ok(()))
    }

    fn call(&mut self, req: Request) -> Self::Future {
        Box::pin(async move {
            Ok(match req {
                Request::SayHello(request) => {
                    ServiceResponse::Single(Response::SayHello(GreetingResponse {
                        result: format!("Hello, {}!", request.name),
                    }))
                }
                Request::SayCustomGreeting(request) => {
                    ServiceResponse::Single(Response::SayCustomGreeting(GreetingResponse {
                        result: format!("{}, {}!", request.greeting, request.name),
                    }))
                }
                Request::SayHelloStream(request) => ServiceResponse::Multiple(
                    stream! {
                        let result = format!("Hello, {}!", request.name);
                        for character in result.chars() {
                            yield Ok(Response::SayHelloStream(GreetingStreamResponse { character }));
                            sleep(Duration::from_millis(300)).await;
                        }
                    }
                    .boxed(),
                ),
            })
        })
    }
}

#[tokio::main]
async fn main() {
    tracing_subscriber::fmt()
        .with_env_filter(
            EnvFilter::builder()
                .with_default_directive(LevelFilter::INFO.into())
                .from_env()
                .unwrap(),
        )
        .init();

    let cli = Cli::parse();

    let service = GreetingService;

    match cli.server_type {
        Command::HttpServer => HttpServer::new(
            service,
            HttpServerConfig {
                port: cli.http_listen_port,
                ..Default::default()
            },
        )
        .run()
        .await
        .expect("http server should not fail"),
        Command::StdioServer => StdioServer::new(service, Default::default())
            .run()
            .await
            .expect("stdio server should not fail"),
    };
}