Awakeables test for node services

contracts/src/main/proto/replier.proto

@@ -6,7 +6,7 @@ option java_outer_classname = "ReplierProto";
 import "google/protobuf/empty.proto";
 import "dev/restate/ext.proto";
 
-package restate.e2e.externalcall.replier;
+package replier;
 
 service Replier {
   option (dev.restate.ext.service_type) = UNKEYED;

contracts/src/main/proto/rng.proto

@@ -5,7 +5,7 @@ option java_outer_classname = "RandomNumberListGeneratorProto";
 
 import "dev/restate/ext.proto";
 
-package restate.e2e.externalcall.rnlg;
+package rnlg;
 
 service RandomNumberListGenerator {
   option (dev.restate.ext.service_type) = UNKEYED;

services/node-services/src/app.ts

@@ -10,6 +10,8 @@ import { protoMetadata as verifierProtoMetadata } from "./generated/verifier";
 import { protoMetadata as interpreterProtoMetadata } from "./generated/interpreter";
 import { protoMetadata as sideEffectProtoMetadata } from "./generated/side_effect";
 import { protoMetadata as proxyProtoMetadata } from "./generated/proxy";
+import { protoMetadata as rngProtoMetadata } from "./generated/rng";
+import { protoMetadata as replierProtoMetadata } from "./generated/replier";
 import { CounterService, CounterServiceFQN } from "./counter";
 import { ListService, ListServiceFQN } from "./collections";
 import { FailingService, FailingServiceFQN } from "./errors";
@@ -30,6 +32,11 @@ import {
   ProxyService,
   ProxyServiceFQN,
 } from "./proxy";
+import {
+  RandomNumberListGeneratorService,
+  RandomNumberListGeneratorServiceFQN,
+} from "./random_number_generator";
+import { ReplierService, ReplierServiceFQN } from "./replier";
 
 let serverBuilder = restate.createServer();
 
@@ -122,6 +129,22 @@ const services = new Map<string, restate.ServiceOpts>([
       instance: new ProxyService(),
     },
   ],
+  [
+    RandomNumberListGeneratorServiceFQN,
+    {
+      descriptor: rngProtoMetadata,
+      service: "RandomNumberListGenerator",
+      instance: new RandomNumberListGeneratorService(),
+    },
+  ],
+  [
+    ReplierServiceFQN,
+    {
+      descriptor: replierProtoMetadata,
+      service: "Replier",
+      instance: new ReplierService(),
+    },
+  ],
 ]);
 
 console.log(services.keys());

services/node-services/src/number_sort_utils.ts

@@ -0,0 +1,22 @@
+export class NumberSortHttpServerUtils {
+  public static async sendSortNumbersRequest(
+    replyId: string,
+    numbers: number[]
+  ): Promise<void> {
+    const url = process.env.HTTP_SERVER_ADDRESS;
+    if (url == undefined) {
+      throw new Error("Supply the HTTP_SERVER_ADDRESS env variable");
+    }
+    const response = await fetch(url, {
+      method: "POST",
+      headers: {
+        "x-reply-id": Buffer.from(replyId).toString("base64"),
+      },
+      body: JSON.stringify(numbers),
+    });
+
+    if (!response.ok) {
+      throw new Error("Response is not ok: " + response);
+    }
+  }
+}

services/node-services/src/random_number_generator.ts

@@ -0,0 +1,36 @@
+import * as restate from "@restatedev/restate-sdk";
+
+import {
+  RandomNumberListGenerator as IRandomNumberListGeneratorService,
+  protobufPackage,
+  GenerateNumbersRequest,
+  GenerateNumbersResponse,
+} from "./generated/rng";
+import { NumberSortHttpServerUtils } from "./number_sort_utils";
+
+export const RandomNumberListGeneratorServiceFQN =
+  protobufPackage + ".RandomNumberListGenerator";
+
+export class RandomNumberListGeneratorService
+  implements IRandomNumberListGeneratorService
+{
+  async generateNumbers(
+    request: GenerateNumbersRequest
+  ): Promise<GenerateNumbersResponse> {
+    const ctx = restate.useContext(this);
+
+    const numbers = Array(request.itemsNumber)
+      .fill(undefined)
+      .map(() => request.itemsNumber * Math.random());
+
+    const { id, promise } = ctx.awakeable<number[]>();
+
+    await ctx.sideEffect(async () => {
+      await NumberSortHttpServerUtils.sendSortNumbersRequest(id, numbers);
+    });
+
+    const sortedNumbers: number[] = await promise;
+
+    return GenerateNumbersResponse.create({ numbers: sortedNumbers });
+  }
+}

services/node-services/src/replier.ts

@@ -0,0 +1,18 @@
+import * as restate from "@restatedev/restate-sdk";
+import { Replier, Reply } from "./generated/replier";
+import { Empty } from "./generated/google/protobuf/empty";
+import { protobufPackage } from "./generated/replier";
+
+export const ReplierServiceFQN = protobufPackage + ".Replier";
+
+export class ReplierService implements Replier {
+  async replyToRandomNumberListGenerator(request: Reply): Promise<Empty> {
+    const ctx = restate.useContext(this);
+
+    const numbers = JSON.parse(request.payload.toString()) as number[];
+
+    ctx.completeAwakeable(request.replyIdentifier.toString(), numbers);
+
+    return Empty.create({});
+  }
+}

tests/src/test/kotlin/dev/restate/e2e/java/AwakeableTest.kt → tests/src/test/kotlin/dev/restate/e2e/AwakeableTest.kt

@@ -1,6 +1,5 @@
-package dev.restate.e2e.java
+package dev.restate.e2e
 
-import dev.restate.e2e.Containers
 import dev.restate.e2e.services.externalcall.RandomNumberListGeneratorGrpc.RandomNumberListGeneratorBlockingStub
 import dev.restate.e2e.services.externalcall.RandomNumberListGeneratorProto.GenerateNumbersRequest
 import dev.restate.e2e.utils.InjectBlockingStub
@@ -10,14 +9,9 @@ import org.assertj.core.api.Assertions.assertThat
 import org.junit.jupiter.api.Tag
 import org.junit.jupiter.api.Test
 import org.junit.jupiter.api.extension.RegisterExtension
-import org.junit.jupiter.api.parallel.Execution
-import org.junit.jupiter.api.parallel.ExecutionMode
 
-// Need to implement the typescript e2e test:
-// https://github.com/restatedev/e2e/issues/108
 @Tag("always-suspending")
-class AwakeableTest {
-
+class JavaAwakeableTest : BaseAwakeableTest() {
   companion object {
     @RegisterExtension
     val deployerExt: RestateDeployerExtension =
@@ -28,9 +22,25 @@ class AwakeableTest {
                 .withContainer(Containers.EXTERNALCALL_HTTP_SERVER_CONTAINER_SPEC)
                 .build())
   }
+}
+
+@Tag("always-suspending")
+class NodeAwakeableTest : BaseAwakeableTest() {
+  companion object {
+    @RegisterExtension
+    val deployerExt: RestateDeployerExtension =
+        RestateDeployerExtension(
+            RestateDeployer.Builder()
+                .withEnv(Containers.getRestateEnvironment())
+                .withServiceEndpoint(Containers.NODE_EXTERNALCALL_SERVICE_SPEC)
+                .withContainer(Containers.EXTERNALCALL_HTTP_SERVER_CONTAINER_SPEC)
+                .build())
+  }
+}
+
+abstract class BaseAwakeableTest {
 
   @Test
-  @Execution(ExecutionMode.CONCURRENT)
   fun generate(
       @InjectBlockingStub randomNumberListGenerator: RandomNumberListGeneratorBlockingStub
   ) {

tests/src/test/kotlin/dev/restate/e2e/Containers.kt

@@ -94,6 +94,15 @@ object Containers {
   val NODE_COLLECTIONS_SERVICE_SPEC =
       nodeServicesContainer("node-collections", ListServiceGrpc.SERVICE_NAME).build()
 
+  val NODE_EXTERNALCALL_SERVICE_SPEC =
+      nodeServicesContainer(
+              "node-externalcall",
+              ReplierGrpc.SERVICE_NAME,
+              RandomNumberListGeneratorGrpc.SERVICE_NAME)
+          .withEnv(
+              "HTTP_SERVER_ADDRESS", "http://${EXTERNALCALL_HTTP_SERVER_CONTAINER_SPEC.first}:8080")
+          .build()
+
   val NODE_ERRORS_SERVICE_SPEC =
       nodeServicesContainer("node-errors", FailingServiceGrpc.SERVICE_NAME).build()