add jsonrpc serialization to rpc (#4623)

Author: tim-smart

.changeset/blue-files-stare.md

@@ -0,0 +1,5 @@
+---
+"@effect/rpc": patch
+---
+
+add RpcSerialization.jsonRpc & RpcSerialization.ndJsonRpc

.changeset/pink-games-peel.md

@@ -0,0 +1,5 @@
+---
+"@effect/rpc": patch
+---
+
+support primitives in rpc payloads

packages/cluster/src/EntityProxy.ts

@@ -63,7 +63,7 @@ export const toRpcGroup = <Rpcs extends Rpc.Any, const Prefix extends string = "
     payloadSchema.make = (input: any, options?: Schema.MakeOptions) => {
       return oldMake({
         entityId: input.entityId,
-        payload: parentRpc.payloadSchema.make(input.payload, options)
+        payload: parentRpc.payloadSchema.make ? parentRpc.payloadSchema.make(input.payload, options) : input.payload
       }, options)
     }
     const rpc = Rpc.make(`${prefix}${parentRpc._tag}`, {

packages/cluster/src/ShardManager.ts

@@ -382,7 +382,7 @@ export const makeClientRpc: Effect.Effect<
     unregister: (address) => client.Unregister({ address }),
     notifyUnhealthyRunner: (address) => client.NotifyUnhealthyRunner({ address }),
     getAssignments: client.GetAssignments(),
-    shardingEvents: client.ShardingEvents({}, { asMailbox: true }),
+    shardingEvents: client.ShardingEvents(void 0, { asMailbox: true }),
     getTime: client.GetTime()
   })
 })

packages/cluster/test/Sharding.test.ts

@@ -479,7 +479,7 @@ describe.concurrent("Sharding", () => {
       const driver = yield* MessageStorage.MemoryDriver
       const makeClient = yield* TestEntity.client
       const client = makeClient("1")
-      const result = yield* client.Never({}, { discard: true })
+      const result = yield* client.Never(void 0, { discard: true })
       expect(result).toEqual(void 0)
       yield* TestClock.adjust(1)
       expect(driver.journal.length).toEqual(1)

packages/platform-node/test/RpcServer.test.ts

@@ -35,6 +35,13 @@ describe("RpcServer", () => {
       Layer.provide([NodeHttpServer.layerTest, RpcSerialization.layerMsgPack])
     )
   )
+  e2eSuite(
+    "e2e http jsonrpc",
+    HttpNdjsonClient.pipe(
+      Layer.provideMerge(HttpNdjsonServer),
+      Layer.provide([NodeHttpServer.layerTest, RpcSerialization.layerNdJsonRpc()])
+    )
+  )
 
   // websocket
   const HttpWsServer = HttpRouter.Default.serve().pipe(
@@ -72,6 +79,13 @@ describe("RpcServer", () => {
       Layer.provide([NodeHttpServer.layerTest, RpcSerialization.layerMsgPack])
     )
   )
+  e2eSuite(
+    "e2e ws jsonrpc",
+    HttpWsClient.pipe(
+      Layer.provideMerge(HttpWsServer),
+      Layer.provide([NodeHttpServer.layerTest, RpcSerialization.layerJsonRpc()])
+    )
+  )
 
   // tcp
   const TcpServer = RpcLive.pipe(
@@ -102,6 +116,13 @@ describe("RpcServer", () => {
       Layer.provide([NodeHttpServer.layerTest, RpcSerialization.layerMsgPack])
     )
   )
+  e2eSuite(
+    "e2e tcp jsonrpc",
+    TcpClient.pipe(
+      Layer.provideMerge(TcpServer),
+      Layer.provide([NodeHttpServer.layerTest, RpcSerialization.layerNdJsonRpc()])
+    )
+  )
 
   // worker
   const WorkerClient = UsersClient.layer.pipe(

packages/rpc/src/Rpc.ts

@@ -45,7 +45,7 @@ export const isRpc = (u: unknown): u is Rpc<any, any, any> => Predicate.hasPrope
  */
 export interface Rpc<
   in out Tag extends string,
-  out Payload extends AnyStructSchema = Schema.Struct<{}>,
+  out Payload extends AnySchema = typeof Schema.Void,
   out Success extends Schema.Schema.Any = typeof Schema.Void,
   out Error extends Schema.Schema.All = typeof Schema.Never,
   out Middleware extends RpcMiddleware.TagClassAny = never
@@ -154,7 +154,7 @@ export interface AnyWithProps {
   readonly [TypeId]: TypeId
   readonly _tag: string
   readonly key: string
-  readonly payloadSchema: AnyStructSchema
+  readonly payloadSchema: AnySchema
   readonly successSchema: Schema.Schema.Any
   readonly errorSchema: Schema.Schema.All
   readonly annotations: Context_.Context<never>
@@ -287,10 +287,8 @@ export type PayloadConstructor<R> = R extends Rpc<
   infer _Success,
   infer _Error,
   infer _Middleware
-> ?
-  Schema.Struct.Constructor<_Payload["fields"]> extends infer T ?
-    [keyof T] extends [never] ? void | {} : Schema.Simplify<T>
-  : never
+> ? _Payload extends { readonly make: (params: infer P, ...rest: infer _Rest) => infer _ } ? P
+  : _Payload["Type"]
   : never
 
 /**
@@ -548,7 +546,7 @@ const Proto = {
 
 const makeProto = <
   const Tag extends string,
-  Payload extends AnyStructSchema,
+  Payload extends Schema.Schema.Any,
   Success extends Schema.Schema.Any,
   Error extends Schema.Schema.All,
   Middleware extends RpcMiddleware.TagClassAny
@@ -567,15 +565,13 @@ const makeProto = <
   return Rpc as any
 }
 
-const constEmptyStruct = Schema.Struct({})
-
 /**
  * @since 1.0.0
  * @category constructors
  */
 export const make = <
   const Tag extends string,
-  Payload extends AnyStructSchema | Schema.Struct.Fields = Schema.Struct<{}>,
+  Payload extends Schema.Schema.Any | Schema.Struct.Fields = typeof Schema.Void,
   Success extends Schema.Schema.Any = typeof Schema.Void,
   Error extends Schema.Schema.All = typeof Schema.Never,
   const Stream extends boolean = false
@@ -607,7 +603,7 @@ export const make = <
       ? options?.payload as any
       : options?.payload
       ? Schema.Struct(options?.payload as any)
-      : constEmptyStruct
+      : Schema.Void
   }
   return makeProto({
     _tag: tag,
@@ -628,22 +624,21 @@ export const make = <
  * @since 1.0.0
  * @category constructors
  */
-export interface AnyStructSchema extends Pipeable {
+export interface AnySchema extends Pipeable {
   readonly [Schema.TypeId]: any
-  readonly make: any
   readonly Type: any
   readonly Encoded: any
   readonly Context: any
+  readonly make?: (params: any, ...rest: ReadonlyArray<any>) => any
   readonly ast: AST.AST
-  readonly fields: Schema.Struct.Fields
   readonly annotations: any
 }
 
 /**
  * @since 1.0.0
  * @category constructors
  */
-export interface AnyTaggedRequestSchema extends AnyStructSchema {
+export interface AnyTaggedRequestSchema extends AnySchema {
   readonly _tag: string
   readonly success: Schema.Schema.Any
   readonly failure: Schema.Schema.All

packages/rpc/src/RpcClient.ts

@@ -182,14 +182,15 @@ export const makeNoSerialization: <Rpcs extends Rpc.Any, E>(
   const onRequest = (rpc: Rpc.AnyWithProps) => {
     const isStream = RpcSchema.isStreamSchema(rpc.successSchema)
     const middleware = getRpcClientMiddleware(rpc)
-    return (payload: any, opts?: {
+    return (payload_: any, opts?: {
       readonly asMailbox?: boolean | undefined
       readonly streamBufferSize?: number | undefined
       readonly headers?: Headers.Input | undefined
       readonly context?: Context.Context<never> | undefined
       readonly discard?: boolean | undefined
     }) => {
       const headers = opts?.headers ? Headers.fromInput(opts.headers) : Headers.empty
+      const payload = payload_ ?? undefined
       if (!isStream) {
         const effect = Effect.useSpan(
           `${spanPrefix}.${rpc._tag}`,
@@ -199,7 +200,7 @@ export const makeNoSerialization: <Rpcs extends Rpc.Any, E>(
               rpc,
               middleware,
               span,
-              "make" in rpc.payloadSchema ? rpc.payloadSchema.make(payload ?? {}) : {},
+              rpc.payloadSchema.make ? rpc.payloadSchema.make(payload) : payload,
               headers,
               opts?.context ?? Context.empty(),
               opts?.discard ?? false
@@ -211,7 +212,7 @@ export const makeNoSerialization: <Rpcs extends Rpc.Any, E>(
         onStreamRequest(
           rpc,
           middleware,
-          payload ? rpc.payloadSchema.make(payload) : {},
+          rpc.payloadSchema.make ? rpc.payloadSchema.make(payload) : payload,
           headers,
           opts?.streamBufferSize ?? 16,
           opts?.context ?? Context.empty()
@@ -712,7 +713,7 @@ export const makeProtocolHttp = (client: HttpClient.HttpClient): Effect.Effect<
 
       const parser = serialization.unsafeMake()
 
-      const encoded = parser.encode(request)
+      const encoded = parser.encode(request)!
       const body = typeof encoded === "string" ?
         HttpBody.text(encoded, serialization.contentType) :
         HttpBody.uint8Array(encoded, serialization.contentType)
@@ -798,7 +799,7 @@ export const makeProtocolSocket = (options?: {
 
     let parser = serialization.unsafeMake()
 
-    const pinger = yield* makePinger(write(parser.encode(constPing)))
+    const pinger = yield* makePinger(write(parser.encode(constPing)!))
 
     yield* Effect.suspend(() => {
       parser = serialization.unsafeMake()
@@ -861,7 +862,9 @@ export const makeProtocolSocket = (options?: {
 
     return {
       send(request) {
-        return Effect.orDie(write(parser.encode(request)))
+        const encoded = parser.encode(request)
+        if (encoded === undefined) return Effect.void
+        return Effect.orDie(write(encoded))
       },
       supportsAck: true,
       supportsTransferables: false