Update the event handler code to use the new annotations (#170)

Author: slinkydeveloper

buf.lock

@@ -4,5 +4,5 @@ deps:
   - remote: buf.build
     owner: restatedev
     repository: proto
-    commit: 4c536701ef5348ecbf3cd1ef6cf825fc
-    digest: shake256:0fdebe27d9653dc31f9951623e0c8dc68a161d2b55146cc22c0501d2bccb22d49ab9b2b80c8f4de8827c4ba168119296f14d323eed5162ef63f128803cc64f47
+    commit: 4de69fd4bcad4d829ebe7643a6f461d1
+    digest: shake256:d9e8a43e890ecbc53822abd7bff03367697f588f815b9beddc1e34675c886549ef262a1e91c8a678db31a7b18979f88ceb98d7017c79fab0ebb81062740f996e

buf.yaml

@@ -1,6 +1,6 @@
 version: v1
 deps:
-  - buf.build/restatedev/proto
+  - buf.build/restatedev/proto:0.4.0
 build:
   excludes:
     - node_modules

proto/dynrpc.proto

@@ -21,7 +21,13 @@ service RpcEndpoint {
 
 	rpc call(RpcRequest) returns (RpcResponse) {};
 
-	rpc handle(dev.restate.StringKeyedEvent) returns (google.protobuf.Empty) {};
+	rpc handle(KeyedEvent) returns (google.protobuf.Empty) {};
+}
+
+message KeyedEvent {
+	string key = 1 [(dev.restate.ext.field) = KEY];
+	bytes payload = 3 [(dev.restate.ext.field) = EVENT_PAYLOAD];
+	map<string, string> attributes = 15 [(dev.restate.ext.field) = EVENT_METADATA];
 }
 
 service UnkeyedRpcEndpoint {

src/server/base_restate_server.ts

@@ -23,7 +23,6 @@ import {
   ServiceDiscoveryResponse,
 } from "../generated/proto/discovery";
 import { Event } from "../types/types";
-import { StringKeyedEvent } from "../generated/dev/restate/events";
 import {
   FileDescriptorProto,
   UninterpretedOption,
@@ -43,6 +42,7 @@ import {
   RpcResponse,
   ProtoMetadata as RpcServiceProtoMetadata,
   protoMetadata as rpcServiceProtoMetadata,
+  KeyedEvent,
 } from "../generated/proto/dynrpc";
 import { RestateContext, useContext } from "../restate_context";
 import { RpcContextImpl } from "../restate_context_impl";
@@ -202,15 +202,15 @@ export abstract class BaseRestateServer {
       throw new TerminalError("Unkeyed Event handlers are not yet supported.");
     }
     const descriptor = createStringKeyedMethodDescriptor(route);
-    const localMethod = (instance: unknown, input: StringKeyedEvent) => {
+    const localMethod = (instance: unknown, input: KeyedEvent) => {
       const ctx = useContext(instance);
       return dispatchKeyedEventHandler(ctx, input, handler);
     };
 
-    const decoder = StringKeyedEvent.decode;
+    const decoder = KeyedEvent.decode;
     const encoder = (message: Empty) => Empty.encode(message).finish();
 
-    const method = new GrpcServiceMethod<StringKeyedEvent, Empty>(
+    const method = new GrpcServiceMethod<KeyedEvent, Empty>(
       route,
       route,
       localMethod,
@@ -447,19 +447,19 @@ async function dispatchUnkeyedRpcHandler(
 
 async function dispatchKeyedEventHandler(
   origCtx: RestateContext,
-  req: StringKeyedEvent,
+  req: KeyedEvent,
   handler: Function
 ): Promise<Empty> {
   const ctx = new RpcContextImpl(origCtx);
   const key = req.key;
-  if (typeof key !== "string" || key.length === 0) {
+  if (key === null || key === undefined || key.length === 0) {
     // we throw a terminal error here, because this cannot be patched by updating code:
     // if the request is wrong (missing a key), the request can never make it
     throw new TerminalError(
-      "Keyed handlers must recieve a non null or empty string key"
+      "Keyed handlers must receive a non null or empty string key"
     );
   }
-  const jsEvent = new Event(key, req.payload, req.source, req.attributes);
+  const jsEvent = new Event(key, req.payload, req.attributes);
   await handler(ctx, jsEvent);
   return Empty.create({});
 }

src/types/types.ts

@@ -125,7 +125,6 @@ export class Event {
   constructor(
     readonly key: string,
     readonly payload: Buffer,
-    readonly source: string,
     readonly attributes: Record<string, string>
   ) {}