Refactor APIs to be a bit more explicit. (#174)

Refactor example.

Author: slinkydeveloper

examples/embedded_example.ts

@@ -14,27 +14,22 @@
 import express, { Request, Response } from "express";
 import * as restate from "../src/public_api";
 
-const rs = restate.connection({ ingress: "http://127.0.0.1:9090" });
+const rs = restate.connection("http://127.0.0.1:8080");
 
 const app = express();
 app.use(express.json());
 
 app.post("/workflow", async (req: Request, res: Response) => {
-  const { id, name } = req.body;
-
-  const response = await rs.invoke({
-    id,
-    input: name,
-    handler: async (ctx, name) => {
-      const p1 = ctx.sideEffect(async () => `Hello ${name}!`);
-      const p2 = ctx.sideEffect(async () => `Bonjour ${name}`);
-      const p3 = ctx.sideEffect(async () => `Hi ${name}`);
-      //const p4 =  ctx
-      // .rpc<{ greet: (name: string) => Promise<string> }>({ path: "greeter" })
-      //   .greet(name);
-
-      return (await p1) + (await p2) + (await p3);
-    },
+  const { id } = req.body;
+
+  const response = await rs.invoke(id, {}, async (ctx) => {
+    // You can use all RestateContext features here (except sleep)
+
+    const response = await ctx.sideEffect(async () => {
+      return (await fetch("https://dummyjson.com/products/1")).json();
+    });
+
+    return response.title;
   });
 
   res.send(response);

src/embedded/api.ts

@@ -15,31 +15,43 @@ import { wrapHandler } from "./handler";
 import crypto from "crypto";
 import { RemoteContext } from "../generated/proto/services";
 import { bufConnectRemoteContext } from "./http2_remote";
+import { OutgoingHttpHeaders } from "http";
 
 export type RestateConnectionOptions = {
-  ingress: string;
+  /**
+   * Additional headers attached to the requests sent to Restate.
+   */
+  headers: OutgoingHttpHeaders;
 };
 
-export type RestateInvocationOptions<I, O> = {
-  id: string;
-  handler: (ctx: RpcContext, input: I) => Promise<O>;
-  input: I;
+export type RestateInvocationOptions = {
+  /**
+   * Retention period for the response in seconds.
+   * After the invocation completes, the response will be persisted for the given duration.
+   * Afterward, the system will clean up the response and treats any subsequent invocation with same operation_id as new.
+   *
+   * If not set, 30 minutes will be used as retention period.
+   */
   retain?: number;
 };
 
-export const connection = (opts: RestateConnectionOptions): RestateConnection =>
-  new RestateConnection(opts);
+export const connection = (
+  address: string,
+  opt?: RestateConnectionOptions
+): RestateConnection =>
+  new RestateConnection(bufConnectRemoteContext(address, opt));
 
 export class RestateConnection {
-  private remote: RemoteContext;
+  constructor(private readonly remote: RemoteContext) {}
 
-  constructor(readonly opts: RestateConnectionOptions) {
-    this.remote = bufConnectRemoteContext(opts.ingress);
-  }
-
-  public invoke<I, O>(opt: RestateInvocationOptions<I, O>): Promise<O> {
-    const method = wrapHandler(opt.handler);
+  public invoke<I, O>(
+    id: string,
+    input: I,
+    handler: (ctx: RpcContext, input: I) => Promise<O>,
+    opt?: RestateInvocationOptions
+  ): Promise<O> {
+    const method = wrapHandler(handler);
     const streamId = crypto.randomUUID();
-    return doInvoke<I, O>(this.remote, opt.id, streamId, method, opt.input);
+    return doInvoke<I, O>(this.remote, id, streamId, input, method, opt);
   }
 }

src/embedded/http2_remote.ts

@@ -15,6 +15,8 @@ import {
   RemoteContext,
   RemoteContextClientImpl,
 } from "../generated/proto/services";
+import { RestateConnectionOptions } from "./api";
+import { OutgoingHttpHeaders } from "http";
 
 export class RequestError extends Error {
   constructor(
@@ -30,8 +32,11 @@ export class RequestError extends Error {
   }
 }
 
-export const bufConnectRemoteContext = (url: string): RemoteContext => {
-  const httpClient = new ProtobufHttp2Client(url);
+export const bufConnectRemoteContext = (
+  url: string,
+  opt?: RestateConnectionOptions
+): RemoteContext => {
+  const httpClient = new ProtobufHttp2Client(url, opt);
 
   return new RemoteContextClientImpl({
     request: (service: string, method: string, data: Uint8Array) =>
@@ -41,8 +46,14 @@ export const bufConnectRemoteContext = (url: string): RemoteContext => {
 
 class ProtobufHttp2Client {
   private session?: http2.ClientHttp2Session;
+  private additionalHeaders: OutgoingHttpHeaders;
 
-  public constructor(private readonly ingress: string) {}
+  public constructor(
+    private readonly ingress: string,
+    opt?: RestateConnectionOptions
+  ) {
+    this.additionalHeaders = opt?.headers == undefined ? {} : opt?.headers;
+  }
 
   private async client(): Promise<http2.ClientHttp2Session> {
     if (this.session !== undefined) {
@@ -65,11 +76,16 @@ class ProtobufHttp2Client {
     const client = await this.client();
 
     const req = client.request({
-      [http2.constants.HTTP2_HEADER_SCHEME]: "http",
-      [http2.constants.HTTP2_HEADER_METHOD]: http2.constants.HTTP2_METHOD_POST,
-      [http2.constants.HTTP2_HEADER_PATH]: path,
-      [http2.constants.HTTP2_HEADER_CONTENT_TYPE]: "application/proto",
-      [http2.constants.HTTP2_HEADER_CONTENT_LENGTH]: body.length,
+      ...{
+        [http2.constants.HTTP2_HEADER_SCHEME]: "http",
+        [http2.constants.HTTP2_HEADER_METHOD]:
+          http2.constants.HTTP2_METHOD_POST,
+        [http2.constants.HTTP2_HEADER_PATH]: path,
+        [http2.constants.HTTP2_HEADER_CONTENT_TYPE]: "application/proto",
+        [http2.constants.HTTP2_HEADER_ACCEPT]: "application/proto",
+        [http2.constants.HTTP2_HEADER_CONTENT_LENGTH]: body.length,
+      },
+      ...this.additionalHeaders,
     });
     req.end(body);
 

src/embedded/invocation.ts

@@ -11,7 +11,11 @@
 
 import { decodeMessagesBuffer } from "../io/decoder";
 import { Message } from "../types/types";
-import { GetResultResponse, RemoteContext } from "../generated/proto/services";
+import {
+  GetResultResponse,
+  RemoteContext,
+  StartRequest,
+} from "../generated/proto/services";
 import { InvocationBuilder } from "../invocation";
 import { HostedGrpcServiceMethod } from "../types/grpc";
 import { StateMachine } from "../state_machine";
@@ -20,24 +24,29 @@ import {
   EmbeddedConnection,
   FencedOffError,
 } from "../connection/embedded_connection";
+import { RestateInvocationOptions } from "./api";
 
 export const doInvoke = async <I, O>(
   remote: RemoteContext,
   operationId: string,
   streamId: string,
+  input: I,
   method: HostedGrpcServiceMethod<I, O>,
-  input: I
+  opt?: RestateInvocationOptions
 ): Promise<O> => {
   //
   // 1. ask to Start this execution.
   //
 
-  const res = await remote.start({
+  const startRequest = StartRequest.fromPartial({
     operationId,
     streamId,
-    retentionPeriodSec: 60,
     argument: Buffer.from(JSON.stringify(input)),
   });
+  if (opt != undefined && opt.retain != undefined) {
+    startRequest.retentionPeriodSec = opt.retain;
+  }
+  const res = await remote.start(startRequest);
 
   if (res.completed !== undefined) {
     return unwrap(res.completed);