Add Promise support for http callout

move promise and dispatch http request to callout folder

Signed-off-by: jizhuozhi.george <jizhuozhi.george@bytedance.com>

Author: jizhuozhi.george

examples/http_parallel_call/Cargo.toml

@@ -2,7 +2,7 @@
 publish = false
 name = "proxy-wasm-example-http-parallel-call"
 version = "0.0.1"
-authors = ["Zhuozhi Ji <jizhuozhi.george@google.com>"]
+authors = ["Zhuozhi Ji <jizhuozhi.george@bytedance.com>"]
 description = "Proxy-Wasm plugin example: HTTP parallel call"
 license = "Apache-2.0"
 edition = "2018"

examples/http_parallel_call/src/lib.rs

@@ -12,83 +12,64 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use proxy_wasm::callout::http::HttpClient;
 use proxy_wasm::hostcalls;
-use proxy_wasm::promise::Promise;
+use proxy_wasm::callout::promise::Promise;
 use proxy_wasm::traits::*;
 use proxy_wasm::types::*;
-use std::collections::HashMap;
-use std::rc::Rc;
 use std::time::Duration;
 
 proxy_wasm::main! {{
     proxy_wasm::set_log_level(LogLevel::Trace);
     proxy_wasm::set_http_context(|_, _| -> Box<dyn HttpContext> { Box::new(HttpParallelCall::default()) });
 }}
 
-type OnHttpResponseArgs = (u32, usize, usize, usize);
-
 #[derive(Default)]
 struct HttpParallelCall {
-    m: HashMap<u32, Rc<Promise<OnHttpResponseArgs>>>,
+    client: HttpClient,
 }
 
 impl HttpContext for HttpParallelCall {
     fn on_http_request_headers(&mut self, _: usize, _: bool) -> Action {
         // "Hello, "
-        let token1 = self
-            .dispatch_http_call(
-                "httpbin",
-                vec![
-                    (":method", "GET"),
-                    (":path", "/base64/SGVsbG8sIAo="),
-                    (":authority", "httpbin.org"),
-                ],
-                None,
-                vec![],
-                Duration::from_secs(1),
-            )
-            .unwrap();
+        let promise1 = self.client.dispatch(
+            "httpbin",
+            vec![
+                (":method", "GET"),
+                (":path", "/base64/SGVsbG8sIA=="),
+                (":authority", "httpbin.org"),
+            ],
+            None,
+            vec![],
+            Duration::from_secs(1),
+        );
 
         // "World!"
-        let token2 = self
-            .dispatch_http_call(
-                "httpbin",
-                vec![
-                    (":method", "GET"),
-                    (":path", "/base64/V29ybGQhCg=="),
-                    (":authority", "httpbin.org"),
-                ],
-                None,
-                vec![],
-                Duration::from_secs(1),
-            )
-            .unwrap();
-
-        let promise1 = Promise::new();
-        let promise2 = Promise::new();
-        self.m.insert(token1, promise1.clone());
-        self.m.insert(token2, promise2.clone());
+        let promise2 = self.client.dispatch(
+            "httpbin",
+            vec![
+                (":method", "GET"),
+                (":path", "/base64/V29ybGQh"),
+                (":authority", "httpbin.org"),
+            ],
+            None,
+            vec![],
+            Duration::from_secs(1),
+        );
 
         Promise::all_of(vec![
             promise1
-                .then(|(_, _, _body_size, _)| get_http_call_response_body_string(0, _body_size))
+                .then(|(_, _, body_size, _)| get_http_call_response_body_string(0, body_size))
                 .then(|body| body.unwrap_or_default()),
             promise2
-                .then(|(_, _, _body_size, _)| get_http_call_response_body_string(0, _body_size))
+                .then(|(_, _, body_size, _)| get_http_call_response_body_string(0, body_size))
                 .then(|body| body.unwrap_or_default()),
         ])
         .then(|results| {
             send_http_response(
                 200,
                 vec![],
-                Some(
-                    format!(
-                        "{}{}\n",
-                        results[0].strip_suffix("\n").unwrap(),
-                        results[1].strip_suffix("\n").unwrap()
-                    )
-                    .as_bytes(),
-                ),
+                Some(format!("{}{}\n", results[0], results[1]).as_bytes()),
             );
         });
 
@@ -104,15 +85,13 @@ impl HttpContext for HttpParallelCall {
 impl Context for HttpParallelCall {
     fn on_http_call_response(
         &mut self,
-        _token_id: u32,
-        _num_headers: usize,
-        _body_size: usize,
-        _num_trailers: usize,
+        token_id: u32,
+        num_headers: usize,
+        body_size: usize,
+        num_trailers: usize,
     ) {
-        let promise = self.m.remove(&_token_id);
-        promise
-            .unwrap()
-            .fulfill((_token_id, _num_headers, _body_size, _num_trailers));
+        self.client
+            .callback(token_id, num_headers, body_size, num_trailers)
     }
 }
 

src/callout/http.rs

@@ -0,0 +1,42 @@
+use crate::callout::promise::Promise;
+use crate::hostcalls;
+use std::collections::HashMap;
+use std::rc::Rc;
+use std::time::Duration;
+
+type OnHttpResponseArgs = (u32, usize, usize, usize);
+
+#[derive(Default)]
+pub struct HttpClient {
+    m: HashMap<u32, Rc<Promise<OnHttpResponseArgs>>>,
+}
+
+impl HttpClient {
+    pub fn dispatch(
+        &mut self,
+        upstream: &str,
+        headers: Vec<(&str, &str)>,
+        body: Option<&[u8]>,
+        trailers: Vec<(&str, &str)>,
+        timeout: Duration,
+    ) -> Rc<Promise<(u32, usize, usize, usize)>> {
+        let token =
+            hostcalls::dispatch_http_call(upstream, headers, body, trailers, timeout).unwrap();
+        let promise = Promise::new();
+        self.m.insert(token, promise.clone());
+        promise
+    }
+
+    pub fn callback(
+        &mut self,
+        token_id: u32,
+        num_headers: usize,
+        body_size: usize,
+        num_trailers: usize,
+    ) {
+        let promise = self.m.remove(&token_id);
+        promise
+            .unwrap()
+            .fulfill((token_id, num_headers, body_size, num_trailers))
+    }
+}

src/callout/mod.rs

@@ -0,0 +1,2 @@
+pub mod http;
+pub mod promise;

src/promise.rs renamed to src/callout/promise.rs

@@ -12,14 +12,13 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-pub mod hostcalls;
-pub mod promise;
-pub mod traits;
-pub mod types;
-
 mod allocator;
+pub mod callout;
 mod dispatcher;
+pub mod hostcalls;
 mod logger;
+pub mod traits;
+pub mod types;
 
 // For crate-type="cdylib".
 #[cfg(not(wasi_exec_model_reactor))]