Add headers to service to service communication (#60)

* Add headers to service to service communication

* Fix little issue in docs

Author: slinkydeveloper

src/context/mod.rs

@@ -219,8 +219,7 @@ impl<'ctx> WorkflowContext<'ctx> {
 ///
 /// ## Scheduling Async Tasks
 ///
-/// To schedule a handler to be called at a later time, have a look at the documentation on [delayed calls][crate::context::ContextClient#delayed-calls].
-///
+/// To schedule a handler to be called at a later time, have a look at the documentation on [delayed calls](Request::send_after).
 ///
 /// ## Durable sleep
 /// To sleep in a Restate application for ten seconds, do the following:

src/context/request.rs

@@ -75,6 +75,7 @@ pub struct Request<'a, Req, Res = ()> {
     ctx: &'a ContextInternal,
     request_target: RequestTarget,
     idempotency_key: Option<String>,
+    headers: Vec<(String, String)>,
     req: Req,
     res: PhantomData<Res>,
 }
@@ -85,11 +86,17 @@ impl<'a, Req, Res> Request<'a, Req, Res> {
             ctx,
             request_target,
             idempotency_key: None,
+            headers: vec![],
             req,
             res: PhantomData,
         }
     }
 
+    pub fn header(mut self, key: String, value: String) -> Self {
+        self.headers.push((key, value));
+        self
+    }
+
     /// Add idempotency key to the request
     pub fn idempotency_key(mut self, idempotency_key: impl Into<String>) -> Self {
         self.idempotency_key = Some(idempotency_key.into());
@@ -102,17 +109,26 @@ impl<'a, Req, Res> Request<'a, Req, Res> {
         Req: Serialize + 'static,
         Res: Deserialize + 'static,
     {
-        self.ctx
-            .call(self.request_target, self.idempotency_key, self.req)
+        self.ctx.call(
+            self.request_target,
+            self.idempotency_key,
+            self.headers,
+            self.req,
+        )
     }
 
     /// Send the request to the service, without waiting for the response.
     pub fn send(self) -> impl InvocationHandle
     where
         Req: Serialize + 'static,
     {
-        self.ctx
-            .send(self.request_target, self.idempotency_key, self.req, None)
+        self.ctx.send(
+            self.request_target,
+            self.idempotency_key,
+            self.headers,
+            self.req,
+            None,
+        )
     }
 
     /// Schedule the request to the service, without waiting for the response.
@@ -123,6 +139,7 @@ impl<'a, Req, Res> Request<'a, Req, Res> {
         self.ctx.send(
             self.request_target,
             self.idempotency_key,
+            self.headers,
             self.req,
             Some(delay),
         )

src/endpoint/context.rs

@@ -15,8 +15,8 @@ use futures::future::{BoxFuture, Either, Shared};
 use futures::{FutureExt, TryFutureExt};
 use pin_project_lite::pin_project;
 use restate_sdk_shared_core::{
-    CoreVM, DoProgressResponse, Error as CoreError, NonEmptyValue, NotificationHandle, RetryPolicy,
-    RunExitResult, TakeOutputResult, Target, TerminalFailure, Value, VM,
+    CoreVM, DoProgressResponse, Error as CoreError, Header, NonEmptyValue, NotificationHandle,
+    RetryPolicy, RunExitResult, TakeOutputResult, Target, TerminalFailure, Value, VM,
 };
 use std::borrow::Cow;
 use std::collections::HashMap;
@@ -371,12 +371,20 @@ impl ContextInternal {
         &self,
         request_target: RequestTarget,
         idempotency_key: Option<String>,
+        headers: Vec<(String, String)>,
         req: Req,
     ) -> impl CallFuture<Response = Res> + Send {
         let mut inner_lock = must_lock!(self.inner);
 
         let mut target: Target = request_target.into();
         target.idempotency_key = idempotency_key;
+        target.headers = headers
+            .into_iter()
+            .map(|(k, v)| Header {
+                key: k.into(),
+                value: v.into(),
+            })
+            .collect();
         let call_result = Req::serialize(&req)
             .map_err(|e| Error::serialization("call", e))
             .and_then(|input| inner_lock.vm.sys_call(target, input).map_err(Into::into));
@@ -443,14 +451,21 @@ impl ContextInternal {
         &self,
         request_target: RequestTarget,
         idempotency_key: Option<String>,
+        headers: Vec<(String, String)>,
         req: Req,
         delay: Option<Duration>,
     ) -> impl InvocationHandle {
         let mut inner_lock = must_lock!(self.inner);
 
         let mut target: Target = request_target.into();
         target.idempotency_key = idempotency_key;
-
+        target.headers = headers
+            .into_iter()
+            .map(|(k, v)| Header {
+                key: k.into(),
+                value: v.into(),
+            })
+            .collect();
         let input = match Req::serialize(&req) {
             Ok(b) => b,
             Err(e) => {