Support Stream

[akiradeveloper]

#[norpc::service]
trait Pub {
  fn message_stream(st: Stream<Elem>) -> Stream<Elem>;
// or
  #[stream]
  fn message_stream() -> Elem; // output only?
}

yields fn message_stream(self) -> Pin<Box<dyn futures::stream::Stream<Item = Elem>;?

I don't know. Need some experiments.

[akiradeveloper]

Related: hyperium/tonic#804

This PR in tonic remove Sync constraint from Stream which annoyed the users (including me) for a long time.

type RouteChatStream = Pin<Box<dyn Stream<Item = Result<RouteNote, Status>> + Send + Sync + 'static>>; to type RouteChatStream = Pin<Box<dyn Stream<Item = Result<RouteNote, Status>> + Send + 'static>>;

Why not add Send constraint to Item?

I know that the the type should be Send because it may move around thread pool on await. (There is impl Send for Pin and impl Send for Box of course) But how can we impl Send for Stream without add impl Send for Item?

=> I understood.

[akiradeveloper]

So the generated function from fn f() -> stream Item should be fn f() -> Pin<Box<dyn futures::stream::Stream<Item = Elem> + Send + 'static>> because we can't return impl Trait for now.

[akiradeveloper]

Code from Tonic. I think rpc like (stream X, A) -> (stream Y, B) isn't permitted.

        let method = match (method.client_streaming(), method.server_streaming()) {
            (false, false) => {
                quote! {
                    #method_doc
                    async fn #name(&self, request: tonic::Request<#req_message>)
                        -> Result<tonic::Response<#res_message>, tonic::Status>;
                }
            }
            (true, false) => {
                quote! {
                    #method_doc
                    async fn #name(&self, request: tonic::Request<tonic::Streaming<#req_message>>)
                        -> Result<tonic::Response<#res_message>, tonic::Status>;
                }
            }
            (false, true) => {
                let stream = quote::format_ident!("{}Stream", method.identifier());
                let stream_doc = generate_doc_comment(&format!(
                    "Server streaming response type for the {} method.",
                    method.identifier()
                ));

                quote! {
                    #stream_doc
                    type #stream: futures_core::Stream<Item = Result<#res_message, tonic::Status>> + Send + 'static;

                    #method_doc
                    async fn #name(&self, request: tonic::Request<#req_message>)
                        -> Result<tonic::Response<Self::#stream>, tonic::Status>;
                }
            }
            (true, true) => {
                let stream = quote::format_ident!("{}Stream", method.identifier());
                let stream_doc = generate_doc_comment(&format!(
                    "Server streaming response type for the {} method.",
                    method.identifier()
                ));

                quote! {
                    #stream_doc
                    type #stream: futures_core::Stream<Item = Result<#res_message, tonic::Status>> + Send + 'static;

                    #method_doc
                    async fn #name(&self, request: tonic::Request<tonic::Streaming<#req_message>>)
                        -> Result<tonic::Response<Self::#stream>, tonic::Status>;
                }
            }

Only these four pairs are allowed.

1. X -> Y
2. X -> stream Y
3. stream X -> Y
4. stream X -> stream Y

[akiradeveloper]

So the possible syntax would be

#[stream(client, server)]
fn f(x: X) -> Y;

fn f(x: Stream<X>) -> Stream<Y>;

fn f(#[stream] x: X) -> #[stream] Y;

[akiradeveloper]

done.