You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@dubbo.apache.org by li...@apache.org on 2023/09/01 07:00:17 UTC
[dubbo-rust] branch main updated: Fixed bugs related to protocols and encoding. (#154)
This is an automated email from the ASF dual-hosted git repository.
liujun pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/dubbo-rust.git
The following commit(s) were added to refs/heads/main by this push:
new d966a5e Fixed bugs related to protocols and encoding. (#154)
d966a5e is described below
commit d966a5e2c8758362eaa9bc232cf81045dcaf087f
Author: Urara <95...@users.noreply.github.com>
AuthorDate: Fri Sep 1 15:00:12 2023 +0800
Fixed bugs related to protocols and encoding. (#154)
---
dubbo/src/triple/client/triple.rs | 45 ++++++++++++++++---------------------
dubbo/src/triple/decode.rs | 23 ++++++++++---------
dubbo/src/triple/encode.rs | 34 ++++++++++++++--------------
dubbo/src/triple/server/triple.rs | 47 +++++++++++++++++++--------------------
4 files changed, 71 insertions(+), 78 deletions(-)
diff --git a/dubbo/src/triple/client/triple.rs b/dubbo/src/triple/client/triple.rs
index f8e1175..d4bc220 100644
--- a/dubbo/src/triple/client/triple.rs
+++ b/dubbo/src/triple/client/triple.rs
@@ -142,17 +142,16 @@ impl TripleClient {
M1: Message + Send + Sync + 'static + Serialize,
M2: Message + Send + Sync + 'static + for<'a> Deserialize<'a> + Default,
{
- let is_json = false;
let (decoder, encoder): (
Box<dyn Decoder<Item = M2, Error = Status> + Send + 'static>,
Box<dyn Encoder<Error = Status, Item = M1> + Send + 'static>,
- ) = get_codec(is_json);
+ ) = get_codec("application/grpc+proto");
let req = req.map(|m| stream::once(future::ready(m)));
let body_stream = encode(
encoder,
req.into_inner().map(Ok),
self.send_compression_encoding,
- is_json,
+ true,
)
.into_stream();
let body = hyper::Body::wrap_stream(body_stream);
@@ -179,9 +178,8 @@ impl TripleClient {
match response {
Ok(v) => {
- let resp = v.map(|body| {
- Decoding::new(body, decoder, self.send_compression_encoding, is_json)
- });
+ let resp = v
+ .map(|body| Decoding::new(body, decoder, self.send_compression_encoding, true));
let (mut parts, body) = Response::from_http(resp).into_parts();
futures_util::pin_mut!(body);
@@ -215,17 +213,16 @@ impl TripleClient {
M1: Message + Send + Sync + 'static + Serialize,
M2: Message + Send + Sync + 'static + for<'a> Deserialize<'a> + Default,
{
- let is_json = false;
let (decoder, encoder): (
Box<dyn Decoder<Item = M2, Error = Status> + Send + 'static>,
Box<dyn Encoder<Error = Status, Item = M1> + Send + 'static>,
- ) = get_codec(is_json);
+ ) = get_codec("application/grpc+proto");
let req = req.into_streaming_request();
let en = encode(
encoder,
req.into_inner().map(Ok),
self.send_compression_encoding,
- is_json,
+ true,
)
.into_stream();
let body = hyper::Body::wrap_stream(en);
@@ -251,9 +248,8 @@ impl TripleClient {
match response {
Ok(v) => {
- let resp = v.map(|body| {
- Decoding::new(body, decoder, self.send_compression_encoding, is_json)
- });
+ let resp = v
+ .map(|body| Decoding::new(body, decoder, self.send_compression_encoding, true));
Ok(Response::from_http(resp))
}
@@ -271,17 +267,16 @@ impl TripleClient {
M1: Message + Send + Sync + 'static + Serialize,
M2: Message + Send + Sync + 'static + for<'a> Deserialize<'a> + Default,
{
- let is_json = false;
let (decoder, encoder): (
Box<dyn Decoder<Item = M2, Error = Status> + Send + 'static>,
Box<dyn Encoder<Error = Status, Item = M1> + Send + 'static>,
- ) = get_codec(is_json);
+ ) = get_codec("application/grpc+proto");
let req = req.into_streaming_request();
let en = encode(
encoder,
req.into_inner().map(Ok),
self.send_compression_encoding,
- is_json,
+ true,
)
.into_stream();
let body = hyper::Body::wrap_stream(en);
@@ -308,9 +303,8 @@ impl TripleClient {
match response {
Ok(v) => {
- let resp = v.map(|body| {
- Decoding::new(body, decoder, self.send_compression_encoding, is_json)
- });
+ let resp = v
+ .map(|body| Decoding::new(body, decoder, self.send_compression_encoding, true));
let (mut parts, body) = Response::from_http(resp).into_parts();
futures_util::pin_mut!(body);
@@ -344,17 +338,16 @@ impl TripleClient {
M1: Message + Send + Sync + 'static + Serialize,
M2: Message + Send + Sync + 'static + for<'a> Deserialize<'a> + Default,
{
- let is_json = false;
let (decoder, encoder): (
Box<dyn Decoder<Item = M2, Error = Status> + Send + 'static>,
Box<dyn Encoder<Error = Status, Item = M1> + Send + 'static>,
- ) = get_codec(is_json);
+ ) = get_codec("application/grpc+proto");
let req = req.map(|m| stream::once(future::ready(m)));
let en = encode(
encoder,
req.into_inner().map(Ok),
self.send_compression_encoding,
- is_json,
+ true,
)
.into_stream();
let body = hyper::Body::wrap_stream(en);
@@ -379,9 +372,8 @@ impl TripleClient {
match response {
Ok(v) => {
- let resp = v.map(|body| {
- Decoding::new(body, decoder, self.send_compression_encoding, is_json)
- });
+ let resp = v
+ .map(|body| Decoding::new(body, decoder, self.send_compression_encoding, true));
Ok(Response::from_http(resp))
}
@@ -391,7 +383,7 @@ impl TripleClient {
}
pub fn get_codec<M1, M2>(
- is_json: bool,
+ content_type: &str,
) -> (
Box<dyn Decoder<Item = M2, Error = Status> + Send + 'static>,
Box<dyn Encoder<Error = Status, Item = M1> + Send + 'static>,
@@ -400,7 +392,8 @@ where
M1: Message + Send + Sync + 'static + Serialize,
M2: Message + Send + Sync + 'static + for<'a> Deserialize<'a> + Default,
{
- match is_json {
+ //Determine whether to use JSON as the serialization method.
+ match content_type.ends_with("json") {
true => {
let mut codec = SerdeCodec::<M1, M2>::default();
(Box::new(codec.decoder()), Box::new(codec.encoder()))
diff --git a/dubbo/src/triple/decode.rs b/dubbo/src/triple/decode.rs
index 875bbea..5b5df84 100644
--- a/dubbo/src/triple/decode.rs
+++ b/dubbo/src/triple/decode.rs
@@ -38,13 +38,13 @@ pub struct Decoding<T> {
trailers: Option<Metadata>,
compress: Option<CompressionEncoding>,
decompress_buf: BytesMut,
- is_json: bool,
+ decode_as_grpc: bool,
}
#[derive(PartialEq)]
enum State {
ReadHeader,
- ReadJSON,
+ ReadHttpBody,
ReadBody { len: usize, is_compressed: bool },
Error,
}
@@ -54,12 +54,13 @@ impl<T> Decoding<T> {
body: B,
decoder: Box<dyn Decoder<Item = T, Error = crate::status::Status> + Send + 'static>,
compress: Option<CompressionEncoding>,
- is_json: bool,
+ decode_as_grpc: bool,
) -> Self
where
B: Body + Send + 'static,
B::Error: Into<crate::Error>,
{
+ //Determine whether to use the gRPC mode to handle request data
Self {
state: State::ReadHeader,
body: body
@@ -76,7 +77,7 @@ impl<T> Decoding<T> {
trailers: None,
compress,
decompress_buf: BytesMut::new(),
- is_json,
+ decode_as_grpc,
}
}
@@ -98,12 +99,12 @@ impl<T> Decoding<T> {
trailer.map(|data| data.map(Metadata::from_headers))
}
- pub fn decode_json(&mut self) -> Result<Option<T>, crate::status::Status> {
+ pub fn decode_http(&mut self) -> Result<Option<T>, crate::status::Status> {
if self.state == State::ReadHeader {
- self.state = State::ReadJSON;
+ self.state = State::ReadHttpBody;
return Ok(None);
}
- if let State::ReadJSON = self.state {
+ if let State::ReadHttpBody = self.state {
if self.buf.is_empty() {
return Ok(None);
}
@@ -138,7 +139,7 @@ impl<T> Decoding<T> {
Ok(None)
}
- pub fn decode_proto(&mut self) -> Result<Option<T>, crate::status::Status> {
+ pub fn decode_grpc(&mut self) -> Result<Option<T>, crate::status::Status> {
if self.state == State::ReadHeader {
// buffer is full
if self.buf.remaining() < super::consts::HEADER_SIZE {
@@ -215,10 +216,10 @@ impl<T> Decoding<T> {
}
pub fn decode_chunk(&mut self) -> Result<Option<T>, crate::status::Status> {
- if self.is_json {
- self.decode_json()
+ if self.decode_as_grpc {
+ self.decode_grpc()
} else {
- self.decode_proto()
+ self.decode_http()
}
}
}
diff --git a/dubbo/src/triple/encode.rs b/dubbo/src/triple/encode.rs
index 1c971d6..baff6be 100644
--- a/dubbo/src/triple/encode.rs
+++ b/dubbo/src/triple/encode.rs
@@ -31,7 +31,7 @@ pub fn encode<E, B>(
mut encoder: Box<dyn Encoder<Error = Status, Item = E> + Send + 'static>,
resp_body: B,
compression_encoding: Option<CompressionEncoding>,
- is_json: bool,
+ encode_as_grpc: bool,
) -> impl TryStream<Ok = Bytes, Error = Status>
where
B: Stream<Item = Result<E, Status>>,
@@ -48,10 +48,10 @@ where
loop {
match resp_body.next().await {
Some(Ok(item)) => {
- if !is_json {
+ if encode_as_grpc {
buf.reserve(super::consts::HEADER_SIZE);
- unsafe {
- buf.advance_mut(super::consts::HEADER_SIZE);
+ unsafe {
+ buf.advance_mut(super::consts::HEADER_SIZE);
}
}
// 编码数据到缓冲中
@@ -67,18 +67,18 @@ where
} else {
encoder.encode(item, &mut EncodeBuf::new(&mut buf)).map_err(|_e| crate::status::Status::new(crate::status::Code::Internal, "encode error".to_string()));
}
- let result=match is_json{
+ let result=match encode_as_grpc{
true=>{
- buf.clone()
- }
- false=>{
let len = buf.len() - super::consts::HEADER_SIZE;
- {
- let mut buf = &mut buf[..super::consts::HEADER_SIZE];
- buf.put_u8(enable_compress as u8);
- buf.put_u32(len as u32);
+ {
+ let mut buf = &mut buf[..super::consts::HEADER_SIZE];
+ buf.put_u8(enable_compress as u8);
+ buf.put_u32(len as u32);
+ }
+ buf.split_to(len + super::consts::HEADER_SIZE)
}
- buf.split_to(len + super::consts::HEADER_SIZE)
+ false=>{
+ buf.clone()
}
};
yield Ok(result.freeze());
@@ -94,12 +94,12 @@ pub fn encode_server<E, B>(
encoder: Box<dyn Encoder<Error = Status, Item = E> + Send + 'static>,
body: B,
compression_encoding: Option<CompressionEncoding>,
- is_json: bool,
+ encode_as_grpc: bool,
) -> EncodeBody<impl Stream<Item = Result<Bytes, Status>>>
where
B: Stream<Item = Result<E, Status>>,
{
- let s = encode(encoder, body, compression_encoding, is_json).into_stream();
+ let s = encode(encoder, body, compression_encoding, encode_as_grpc).into_stream();
EncodeBody::new_server(s)
}
@@ -107,12 +107,12 @@ pub fn encode_client<E, B>(
encoder: Box<dyn Encoder<Error = Status, Item = E> + Send + 'static>,
body: B,
compression_encoding: Option<CompressionEncoding>,
- is_json: bool,
+ is_grpc: bool,
) -> EncodeBody<impl Stream<Item = Result<Bytes, Status>>>
where
B: Stream<Item = E>,
{
- let s = encode(encoder, body.map(Ok), compression_encoding, is_json).into_stream();
+ let s = encode(encoder, body.map(Ok), compression_encoding, is_grpc).into_stream();
EncodeBody::new_client(s)
}
diff --git a/dubbo/src/triple/server/triple.rs b/dubbo/src/triple/server/triple.rs
index 2586e64..2567b49 100644
--- a/dubbo/src/triple/server/triple.rs
+++ b/dubbo/src/triple/server/triple.rs
@@ -73,12 +73,12 @@ where
.headers()
.get("content-type")
.cloned()
- .unwrap_or(HeaderValue::from_str("application/json").unwrap());
- let is_json = content_type == "application/json" || content_type == "application/grpc+json";
+ .unwrap_or(HeaderValue::from_str("application/grpc+proto").unwrap());
+ let content_type_str = content_type.to_str().unwrap();
let (decoder, encoder): (
Box<dyn Decoder<Item = M1, Error = Status> + Send + 'static>,
Box<dyn Encoder<Error = Status, Item = M2> + Send + 'static>,
- ) = get_codec(is_json);
+ ) = get_codec(content_type_str);
let mut accept_encoding = CompressionEncoding::from_accept_encoding(req.headers());
if self.compression.is_none() || accept_encoding.is_none() {
accept_encoding = None;
@@ -90,7 +90,7 @@ where
Err(status) => return status.to_http(),
};
- let req_stream = req.map(|body| Decoding::new(body, decoder, compression, is_json));
+ let req_stream = req.map(|body| Decoding::new(body, decoder, compression, true));
let resp = service.call(Request::from_http(req_stream)).await;
@@ -103,7 +103,7 @@ where
encoder,
stream::once(future::ready(resp_body)).map(Ok).into_stream(),
accept_encoding,
- is_json,
+ true,
);
parts
@@ -133,13 +133,12 @@ where
.headers()
.get("content-type")
.cloned()
- .unwrap_or(HeaderValue::from_str("application/json").unwrap());
- let is_json = content_type == "application/json" || content_type == "application/grpc+json";
-
+ .unwrap_or(HeaderValue::from_str("application/grpc+proto").unwrap());
+ let content_type_str = content_type.to_str().unwrap();
let (decoder, encoder): (
Box<dyn Decoder<Item = M1, Error = Status> + Send + 'static>,
Box<dyn Encoder<Error = Status, Item = M2> + Send + 'static>,
- ) = get_codec(is_json);
+ ) = get_codec(content_type_str);
// Firstly, get grpc_accept_encoding from http_header, get compression
// Secondly, if server enable compression and compression is valid, this method should compress response
let mut accept_encoding = CompressionEncoding::from_accept_encoding(req.headers());
@@ -153,7 +152,7 @@ where
Err(status) => return status.to_http(),
};
- let req_stream = req.map(|body| Decoding::new(body, decoder, compression, is_json));
+ let req_stream = req.map(|body| Decoding::new(body, decoder, compression, true));
let resp = service.call(Request::from_http(req_stream)).await;
@@ -161,7 +160,7 @@ where
Ok(v) => v.into_http().into_parts(),
Err(err) => return err.to_http(),
};
- let resp_body = encode_server(encoder, resp_body, compression, is_json);
+ let resp_body = encode_server(encoder, resp_body, compression, true);
parts
.headers
@@ -190,13 +189,12 @@ where
.headers()
.get("content-type")
.cloned()
- .unwrap_or(HeaderValue::from_str("application/json").unwrap());
- let is_json = content_type == "application/json" || content_type == "application/grpc+json";
-
+ .unwrap_or(HeaderValue::from_str("application/grpc+proto").unwrap());
+ let content_type_str = content_type.to_str().unwrap();
let (decoder, encoder): (
Box<dyn Decoder<Item = M1, Error = Status> + Send + 'static>,
Box<dyn Encoder<Error = Status, Item = M2> + Send + 'static>,
- ) = get_codec(is_json);
+ ) = get_codec(content_type_str);
// Firstly, get grpc_accept_encoding from http_header, get compression
// Secondly, if server enable compression and compression is valid, this method should compress response
let mut accept_encoding = CompressionEncoding::from_accept_encoding(req.headers());
@@ -209,7 +207,7 @@ where
Ok(val) => val,
Err(status) => return status.to_http(),
};
- let req_stream = req.map(|body| Decoding::new(body, decoder, compression, is_json));
+ let req_stream = req.map(|body| Decoding::new(body, decoder, compression, true));
let (parts, mut body) = Request::from_http(req_stream).into_parts();
let msg = body.try_next().await.unwrap().ok_or_else(|| {
crate::status::Status::new(crate::status::Code::Unknown, "request wrong".to_string())
@@ -225,7 +223,7 @@ where
Ok(v) => v.into_http().into_parts(),
Err(err) => return err.to_http(),
};
- let resp_body = encode_server(encoder, resp_body, compression, is_json);
+ let resp_body = encode_server(encoder, resp_body, compression, true);
parts
.headers
@@ -253,7 +251,6 @@ where
if self.compression.is_none() || accept_encoding.is_none() {
accept_encoding = None;
}
-
let compression = match self.get_encoding_from_req(req.headers()) {
Ok(val) => val,
Err(status) => return status.to_http(),
@@ -262,14 +259,16 @@ where
.headers()
.get("content-type")
.cloned()
- .unwrap_or(HeaderValue::from_str("application/json").unwrap());
- let is_json = content_type == "application/json" || content_type == "application/grpc+json";
-
+ .unwrap_or(HeaderValue::from_str("application/grpc+proto").unwrap());
+ let content_type_str = content_type.to_str().unwrap();
+ //Determine whether to use the gRPC mode to handle request data
+ let handle_request_as_grpc = content_type_str.contains("grpc");
let (decoder, encoder): (
Box<dyn Decoder<Item = M1, Error = Status> + Send + 'static>,
Box<dyn Encoder<Error = Status, Item = M2> + Send + 'static>,
- ) = get_codec(is_json);
- let req_stream = req.map(|body| Decoding::new(body, decoder, compression, is_json));
+ ) = get_codec(content_type_str);
+ let req_stream =
+ req.map(|body| Decoding::new(body, decoder, compression, handle_request_as_grpc));
let (parts, mut body) = Request::from_http(req_stream).into_parts();
let msg = body.try_next().await.unwrap().ok_or_else(|| {
crate::status::Status::new(crate::status::Code::Unknown, "request wrong".to_string())
@@ -289,7 +288,7 @@ where
encoder,
stream::once(future::ready(resp_body)).map(Ok).into_stream(),
accept_encoding,
- is_json,
+ handle_request_as_grpc,
);
parts