grpc 0.6.2

Rust implementation of gRPC
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
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158

///! Convert HTTP response stream to gRPC stream
use std::collections::VecDeque;

use futures::future::Future;
use futures::stream;
use futures::stream::Stream;
use futures::Async;
use futures::Poll;

use grpc::GrpcStatus;
use grpc::HEADER_GRPC_MESSAGE;
use grpc::HEADER_GRPC_STATUS;

use httpbis;
use httpbis::Headers;

use grpc_frame::*;

use bytes::Bytes;

use result;

use error::Error;
use error::GrpcMessageError;

use httpbis::DataOrTrailers;
use httpbis::HttpStreamAfterHeaders;
use metadata::*;
use resp::*;
use stream_item::*;

fn init_headers_to_metadata(headers: Headers) -> result::Result<Metadata> {
    if headers.get_opt(":status") != Some("200") {
        return Err(Error::Other("not 200"));
    }

    // Check gRPC status code and message
    // TODO: a more detailed error message.
    if let Some(grpc_status) = headers.get_opt_parse(HEADER_GRPC_STATUS) {
        if grpc_status != GrpcStatus::Ok as i32 {
            let message = headers
                .get_opt(HEADER_GRPC_MESSAGE)
                .unwrap_or("unknown error");
            return Err(Error::GrpcMessage(GrpcMessageError {
                grpc_status: grpc_status,
                grpc_message: message.to_owned(),
            }));
        }
    }

    Ok(Metadata::from_headers(headers)?)
}

pub fn http_response_to_grpc_frames(response: httpbis::Response) -> StreamingResponse<Bytes> {
    StreamingResponse::new(
        response
            .0
            .map_err(|e| Error::from(e))
            .and_then(|(headers, rem)| {
                let metadata = init_headers_to_metadata(headers)?;
                let frames: GrpcStreamWithTrailingMetadata<Bytes> =
                    GrpcStreamWithTrailingMetadata::new(
                        GrpcFrameFromHttpFramesStreamResponse::new(rem),
                    );
                Ok((metadata, frames))
            }),
    )
}

struct GrpcFrameFromHttpFramesStreamResponse {
    http_stream_stream: HttpStreamAfterHeaders,
    buf: Bytes,
    parsed_frames: VecDeque<Bytes>,
    error: Option<stream::Once<ItemOrMetadata<Bytes>, Error>>,
}

impl GrpcFrameFromHttpFramesStreamResponse {
    pub fn new(http_stream_stream: HttpStreamAfterHeaders) -> Self {
        GrpcFrameFromHttpFramesStreamResponse {
            http_stream_stream,
            buf: Bytes::new(),
            parsed_frames: VecDeque::new(),
            error: None,
        }
    }
}

impl Stream for GrpcFrameFromHttpFramesStreamResponse {
    type Item = ItemOrMetadata<Bytes>;
    type Error = Error;

    fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
        loop {
            if let Some(ref mut error) = self.error {
                return error.poll();
            }

            self.parsed_frames
                .extend(match parse_grpc_frames_from_bytes(&mut self.buf) {
                    Ok(r) => r,
                    Err(e) => {
                        self.error = Some(stream::once(Err(e)));
                        continue;
                    }
                });

            if let Some(frame) = self.parsed_frames.pop_front() {
                return Ok(Async::Ready(Some(ItemOrMetadata::Item(frame))));
            }

            let part_opt = match self.http_stream_stream.poll()? {
                Async::NotReady => return Ok(Async::NotReady),
                Async::Ready(part_opt) => part_opt,
            };
            let part = match part_opt {
                None => {
                    if self.buf.is_empty() {
                        return Ok(Async::Ready(None));
                    } else {
                        self.error = Some(stream::once(Err(Error::Other("partial frame"))));
                        continue;
                    }
                }
                Some(part) => part,
            };

            match part {
                DataOrTrailers::Trailers(headers) => {
                    if !self.buf.is_empty() {
                        self.error = Some(stream::once(Err(Error::Other("partial frame"))));
                    } else {
                        let grpc_status = headers.get_opt_parse(HEADER_GRPC_STATUS);
                        if grpc_status == Some(GrpcStatus::Ok as i32) {
                            return Ok(Async::Ready(Some(ItemOrMetadata::TrailingMetadata(
                                Metadata::from_headers(headers)?,
                            ))));
                        } else {
                            self.error = Some(stream::once(Err(if let Some(message) =
                                headers.get_opt(HEADER_GRPC_MESSAGE)
                            {
                                Error::GrpcMessage(GrpcMessageError {
                                    grpc_status: grpc_status.unwrap_or(GrpcStatus::Unknown as i32),
                                    grpc_message: message.to_owned(),
                                })
                            } else {
                                Error::Other("not xxx")
                            })));
                        }
                    }
                    continue;
                }
                DataOrTrailers::Data(data, ..) => {
                    self.buf.extend_from_slice(&data);
                }
            }
        }
    }
}