1use bytes::{BufMut, BytesMut};
2use http::{Request, Response};
3use tokio_util::codec::{Decoder, Encoder};
4use yykv_types::DsError;
5
6pub struct S3HttpCodec;
7
8impl Decoder for S3HttpCodec {
9 type Item = Request<Vec<u8>>;
10 type Error = DsError;
11
12 fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
13 let header_end = src.windows(4).position(|w| w == b"\r\n\r\n");
15 if let Some(pos) = header_end {
16 let header_part = src.split_to(pos + 4);
17 let header_str = String::from_utf8_lossy(&header_part);
18
19 let mut content_len = 0;
21 for line in header_str.lines() {
22 if line.to_lowercase().starts_with("content-length:") {
23 content_len = line["content-length:".len()..].trim().parse().unwrap_or(0);
24 }
25 }
26
27 if src.len() < content_len {
28 return Ok(None);
29 }
30
31 let body = src.split_to(content_len).to_vec();
32
33 let req = Request::builder()
35 .method("GET")
36 .uri("/")
37 .body(body)
38 .map_err(|e| DsError::protocol(e.to_string()))?;
39
40 Ok(Some(req))
41 } else {
42 Ok(None)
43 }
44 }
45}
46
47impl Encoder<Response<Vec<u8>>> for S3HttpCodec {
48 type Error = DsError;
49
50 fn encode(&mut self, item: Response<Vec<u8>>, dst: &mut BytesMut) -> Result<(), Self::Error> {
51 let status = item.status();
52 let body = item.into_body();
53
54 let response_str = format!(
55 "HTTP/1.1 {} {}\r\nContent-Length: {}\r\n\r\n",
56 status.as_u16(),
57 status.canonical_reason().unwrap_or(""),
58 body.len()
59 );
60
61 dst.put_slice(response_str.as_bytes());
62 dst.put_slice(&body);
63 Ok(())
64 }
65}