[client] Add 'raw' opt/sendOpt to skip JSON serde (#404)

This commit adds a `raw` optional property to the client
To allow skipping JSON serde. This will pass the raw input buffer
to the underlying fetch

Author: igalshilman

packages/restate-sdk-clients/src/api.ts

@@ -88,6 +88,13 @@ export interface IngresCallOptions {
    * Headers to attach to the request.
    */
   headers?: Record<string, string>;
+
+  /**
+   * Do not auto serialize the input and output of the handler.
+   * Setting this to true, would allow you to send binary data to the handler,
+   * and not assuming that the payload is JSON.
+   */
+  raw?: boolean;
 }
 
 export interface IngresSendOptions extends IngresCallOptions {

packages/restate-sdk-clients/src/ingress.ts

@@ -138,10 +138,14 @@ const doComponentInvocation = async <I, O>(
       fragments.push("send");
     }
   }
+  const raw = params.opts?.opts.raw ?? false;
   //
   // request body
   //
-  const { body, contentType } = serializeBodyWithContentType(params.parameter);
+  const { body, contentType } = serializeBodyWithContentType(
+    params.parameter,
+    raw
+  );
   //
   // headers
   //
@@ -178,7 +182,7 @@ const doComponentInvocation = async <I, O>(
     );
   }
   const responseBuf = await httpResponse.arrayBuffer();
-  const json = deserializeJson(new Uint8Array(responseBuf)) as O;
+  const json = deserializeJson(new Uint8Array(responseBuf), raw) as O;
   if (!params.send) {
     return json;
   }
@@ -210,7 +214,7 @@ const doWorkflowHandleCall = async <O>(
   });
   if (httpResponse.ok) {
     const responseBuf = await httpResponse.arrayBuffer();
-    return deserializeJson(new Uint8Array(responseBuf)) as O;
+    return deserializeJson(new Uint8Array(responseBuf), false) as O;
   }
   const body = await httpResponse.text();
   throw new HttpCallError(
@@ -361,7 +365,7 @@ class HttpIngress implements Ingress {
     payload?: T | undefined
   ): Promise<void> {
     const url = `${this.opts.url}/restate/a/${id}/resolve`;
-    const { body, contentType } = serializeBodyWithContentType(payload);
+    const { body, contentType } = serializeBodyWithContentType(payload, false);
     const headers = {
       ...(this.opts.headers ?? {}),
     };
@@ -427,7 +431,7 @@ class HttpIngress implements Ingress {
     });
     if (httpResponse.ok) {
       const responseBuf = await httpResponse.arrayBuffer();
-      return deserializeJson(new Uint8Array(responseBuf)) as T;
+      return deserializeJson(new Uint8Array(responseBuf), false) as T;
     }
     const body = await httpResponse.text();
     throw new HttpCallError(
@@ -447,21 +451,33 @@ function computeDelayAsIso(opts: SendOpts): string {
 }
 
 // eslint-disable-next-line @typescript-eslint/no-explicit-any
-function deserializeJson(what: Uint8Array): any {
-  if (what === undefined || what.length == 0) {
+function deserializeJson(what: Uint8Array, raw: boolean): any {
+  if (what === undefined) {
+    return undefined;
+  }
+  if (raw) {
+    return what;
+  }
+  if (what.length == 0) {
     return undefined;
   }
   const json = new TextDecoder().decode(what);
   return JSON.parse(json);
 }
 
-function serializeBodyWithContentType(body: unknown): {
+function serializeBodyWithContentType(
+  body: unknown,
+  raw: boolean
+): {
   body?: Uint8Array;
   contentType?: string;
 } {
   if (body === undefined) {
     return {};
   }
+  if (raw) {
+    return { body: body as Uint8Array };
+  }
   const json = JSON.stringify(body);
   const buffer = new TextEncoder().encode(json);
 

packages/restate-sdk-examples/src/ingress_client.ts

@@ -146,6 +146,24 @@ const workflow = async (name: string) => {
   console.log(await client.workflowAttach());
 };
 
+const binaryRawCall = async (name: string) => {
+  const counter = ingress.objectClient(Counter, name);
+
+  const buffer = new TextEncoder().encode("hello!");
+
+  const outBuffer = await counter.binary(
+    buffer,
+    restate.Opts.from({
+      raw: true, // <-- tell the client to avoid JSON encoding/decoding
+      headers: { "content-type": "application/octet-stream" },
+    })
+  );
+
+  const str = new TextDecoder().decode(outBuffer);
+
+  console.log(`We got a buffer for ${name} : ${str}`);
+};
+
 // Before running this example, make sure
 // to run and register `greeter`, `counter` and `workflow` services.
 //
@@ -156,15 +174,16 @@ const workflow = async (name: string) => {
 // 3. restate deployment add localhost:9080
 
 Promise.resolve()
-  .then(() => sendAndCollectResultLater("boby"))
-  .then(() => simpleCall("bob"))
   .then(() => objectCall("bob"))
   .then(() => objectCall("mop"))
+  .then(() => simpleCall("bob"))
+  .then(() => sendAndCollectResultLater("boby"))
   .then(() => workflow("boby"))
   .then(() => idempotentCall("joe", "idemp-1"))
   .then(() => idempotentCall("joe", "idemp-1"))
   .then(() => customHeadersCall("bob"))
   .then(() => globalCustomHeaders("bob"))
   .then(() => customInterface("bob"))
   .then(() => delayedCall("bob"))
+  .then(() => binaryRawCall("bob"))
   .catch((e) => console.error(e));

packages/restate-sdk-examples/src/object.ts

@@ -36,6 +36,24 @@ const counter = restate.object({
         return (await ctx.get("count")) ?? 0;
       }
     ),
+
+    /**
+     * Handlers (shared or exclusive) can be configured to bypass JSON serialization,
+     * by specifying the input (accept) and output (contentType) content types.
+     *
+     * to call that handler with binary data, you can use the following curl command:
+     * curl -X POST -H "Content-Type: application/octet-stream" --data-binary 'hello' ${RESTATE_INGRESS_URL}/counter/mykey/binary
+     */
+    binary: restate.handlers.object.exclusive(
+      {
+        accept: "application/octet-stream",
+        contentType: "application/octet-stream",
+      },
+      async (ctx: restate.ObjectContext, data: Uint8Array) => {
+        // console.log("Received binary data", data);
+        return data;
+      }
+    ),
   },
 });