Revert "fix: rpc race condition (#702)" (#712)

This reverts commit e9c92ad.

Author: menduz

package-lock.json

@@ -66,7 +66,7 @@
     "@dcl/kernel-interface": "^2.0.0-20210922153939.commit-017905d",
     "@dcl/legacy-ecs": "^6.11.8",
     "@dcl/protocol": "^1.0.0-3373991894.commit-8aa3a49",
-    "@dcl/rpc": "^1.1.1-20221115000939.commit-9a51ad0",
+    "@dcl/rpc": "^1.0.4-20221113192916.commit-3ef5187",
     "@dcl/scene-runtime": "^1.0.0-20221102005705.commit-05d463c",
     "@dcl/schemas": "^5.21.0",
     "@dcl/urn-resolver": "^2.0.2",

package.json

@@ -4,6 +4,6 @@ globalThis.process = {
   browser: true,
   env: {},
   nextTick(fn, ...args) {
-    queueMicrotask(() => fn(...args))
+    require('queue-microtask')(() => fn(...args))
   }
 }

packages/entryPoints/inject.js

@@ -1,8 +1,19 @@
-import { createRpcClient, RpcClientPort, Transport } from '@dcl/rpc'
+import { createRpcClient, Transport } from '@dcl/rpc'
+import future, { IFuture } from 'fp-future'
+import { registerCRDTService } from './services/crdtService'
+import { RendererProtocol } from './types'
 
-export async function createRendererRpcClient(transport: Transport): Promise<RpcClientPort> {
+export const rendererProtocol: IFuture<RendererProtocol> = future()
+
+export async function createRendererRpcClient(transport: Transport): Promise<RendererProtocol> {
   const rpcClient = await createRpcClient(transport)
   const clientPort = await rpcClient.createPort('renderer-protocol')
 
-  return clientPort
+  const crdtService = registerCRDTService(clientPort)
+
+  rendererProtocol.resolve({
+    crdtService
+  })
+
+  return rendererProtocol
 }

packages/renderer-protocol/rpcClient.ts

@@ -90,9 +90,6 @@ export function webSocketTransportAdapter(url: string, options: CommonRendererOp
 
   const transport: Transport = {
     ...events,
-    get isConnected(): boolean {
-      return (socket && socket.readyState === socket.OPEN) || false
-    },
     sendMessage(message: any) {
       send(message)
     },

packages/renderer-protocol/transports/webSocketTransportAdapter.ts

@@ -5,7 +5,7 @@ export type WebTransportOptions = {
   wasmModule: any
 }
 
-export function webTransport(options: WebTransportOptions, unityDclInstance: any) {
+export function webTransport(options: WebTransportOptions): Transport {
   const events = mitt<TransportEvents>()
   const ALLOC_SIZE = 8388608
   let heapPtr: number
@@ -17,25 +17,15 @@ export function webTransport(options: WebTransportOptions, unityDclInstance: any
   }
 
   let isClosed = false
-  let didConnect = false
 
-  unityDclInstance.BinaryMessageFromEngine = function (data: Uint8Array) {
-    if (!didConnect) {
-      throw new Error('Received data from unity before connection was established')
-    }
+  ;(globalThis as any).DCL.BinaryMessageFromEngine = function (data: Uint8Array) {
     const copiedData = new Uint8Array(data)
     events.emit('message', copiedData)
   }
 
   const transport: Transport = {
     ...events,
-    get isConnected() {
-      return didConnect
-    },
     sendMessage(message) {
-      if (!didConnect) {
-        throw new Error('Tried to send a message before connection was established')
-      }
       if (!!sendMessageToRenderer && !isClosed) {
         options.wasmModule.HEAPU8.set(message, heapPtr)
         sendMessageToRenderer(heapPtr, message.length)
@@ -49,12 +39,7 @@ export function webTransport(options: WebTransportOptions, unityDclInstance: any
     }
   }
 
-  events.on('connect', () => {
-    didConnect = true
-  })
-
-  // connect the transport
-  events.emit('connect', {})
+  queueMicrotask(() => events.emit('connect', {}))
 
   return transport
 }

packages/renderer-protocol/transports/webTransport.ts

@@ -11,7 +11,8 @@ import {
 
 import { PortContext } from './context'
 import { EntityAction, EntityActionType } from 'shared/types'
-import { registerCRDTService } from 'renderer-protocol/services/crdtService'
+
+import { rendererProtocol } from './../../../renderer-protocol/rpcClient'
 
 function getPayload(payloadType: EAType, payload: Payload): any {
   switch (payloadType) {
@@ -59,62 +60,60 @@ function getPayload(payloadType: EAType, payload: Payload): any {
 }
 
 export function registerEngineApiServiceServerImplementation(port: RpcServerPort<PortContext>) {
-  codegen.registerService(port, EngineApiServiceDefinition, async (port, ctx) => {
-    const crdtService = registerCRDTService(ctx.rendererPort)
-
-    return {
-      async sendBatch(req: ManyEntityAction, ctx) {
-        const actions: EntityAction[] = []
+  codegen.registerService(port, EngineApiServiceDefinition, async () => ({
+    async sendBatch(req: ManyEntityAction, ctx) {
+      const actions: EntityAction[] = []
 
-        for (const action of req.actions) {
-          const actionType = eaTypeToStr(action.type)
-          if (actionType && action.payload) {
-            actions.push({
-              type: actionType,
-              tag: action.tag,
-              payload: getPayload(action.type, action.payload as any)
-            })
-          }
+      for (const action of req.actions) {
+        const actionType = eaTypeToStr(action.type)
+        if (actionType && action.payload) {
+          actions.push({
+            type: actionType,
+            tag: action.tag,
+            payload: getPayload(action.type, action.payload as any)
+          })
         }
+      }
 
-        if (actions.length) {
-          ctx.sendBatch(actions)
-        }
+      if (actions.length) {
+        ctx.sendBatch(actions)
+      }
 
-        const events: EventData[] = ctx.events
+      const events: EventData[] = ctx.events
 
-        if (events.length) {
-          ctx.events = []
-        }
+      if (events.length) {
+        ctx.events = []
+      }
 
-        return { events }
-      },
+      return { events }
+    },
 
-      async subscribe(req, ctx) {
-        ctx.subscribedEvents.add(req.eventId)
-        return {}
-      },
-      async unsubscribe(req, ctx) {
-        ctx.subscribedEvents.delete(req.eventId)
-        return {}
-      },
-      async crdtSendToRenderer(req, ctx) {
-        return crdtService.sendCrdt({
-          sceneId: ctx.sceneData.id,
-          payload: req.data,
-          sceneNumber: ctx.sceneData.sceneNumber
-        })
-      },
+    async subscribe(req, ctx) {
+      ctx.subscribedEvents.add(req.eventId)
+      return {}
+    },
+    async unsubscribe(req, ctx) {
+      ctx.subscribedEvents.delete(req.eventId)
+      return {}
+    },
+    async crdtSendToRenderer(req, ctx) {
+      const protocol = await rendererProtocol
+      return protocol.crdtService.sendCrdt({
+        sceneId: ctx.sceneData.id,
+        payload: req.data,
+        sceneNumber: ctx.sceneData.sceneNumber
+      })
+    },
 
-      async crdtGetMessageFromRenderer(_, ctx) {
-        const response = await crdtService.pullCrdt({
-          sceneId: ctx.sceneData.id,
-          sceneNumber: ctx.sceneData.sceneNumber
-        })
-        return { data: [response.payload] }
-      }
+    async crdtGetMessageFromRenderer(_, ctx) {
+      const protocol = await rendererProtocol
+      const response = await protocol.crdtService.pullCrdt({
+        sceneId: ctx.sceneData.id,
+        sceneNumber: ctx.sceneData.sceneNumber
+      })
+      return { data: [response.payload] }
     }
-  })
+  }))
 }
 function eaTypeToStr(type: EAType): EntityActionType | null {
   switch (type) {

packages/shared/apis/host/EngineAPI.ts

@@ -2,7 +2,6 @@ import { ILogger } from './../../logger'
 import { EntityAction, LoadableScene } from './../../types'
 import { PermissionItem } from '@dcl/protocol/out-ts/decentraland/kernel/apis/permissions.gen'
 import { EventData } from '@dcl/protocol/out-ts/decentraland/kernel/apis/engine_api.gen'
-import { RpcClientPort } from '@dcl/rpc'
 
 type WithRequired<T, K extends keyof T> = T & { [P in K]-?: T[P] }
 
@@ -23,7 +22,4 @@ export type PortContext = {
   sendSceneEvent<K extends keyof IEvents>(id: K, event: IEvents[K]): void
   sendProtoSceneEvent(event: EventData): void
   logger: ILogger
-
-  // port used for this specific scene in the renderer
-  rendererPort: RpcClientPort
 }

packages/shared/apis/host/context.ts

@@ -10,7 +10,7 @@ export class OfflineAdapter implements MinimumCommunicationsAdapter {
   async getVoiceHandler(): Promise<VoiceHandler> {
     return createOpusVoiceHandler()
   }
-  async disconnect(_error?: Error | undefined): Promise<void> {}
-  send(_data: Uint8Array, _hints: SendHints): void {}
+  async disconnect(error?: Error | undefined): Promise<void> {}
+  send(data: Uint8Array, hints: SendHints): void {}
   async connect(): Promise<void> {}
 }

packages/shared/comms/adapters/OfflineAdapter.ts

@@ -46,9 +46,9 @@ export class SimulationRoom implements RoomConnection {
     this.tick = setInterval(this.update.bind(this), 60)
     this.roomConnection = new Rfc4RoomConnection({
       events: mitt<CommsAdapterEvents>(),
-      send(_data: Uint8Array, _hints: SendHints): void {},
+      send(data: Uint8Array, hints: SendHints): void {},
       async connect(): Promise<void> {},
-      async disconnect(_error?: Error): Promise<void> {},
+      async disconnect(error?: Error): Promise<void> {},
       async getVoiceHandler() {
         throw new Error('not implemented')
       }
@@ -190,11 +190,11 @@ export class SimulationRoom implements RoomConnection {
     }
   }
 
-  async disconnect(_error?: Error | undefined): Promise<void> {
+  async disconnect(error?: Error | undefined): Promise<void> {
     clearInterval(this.tick)
   }
 
-  send(_data: Uint8Array, _hints: SendHints): void {}
+  send(data: Uint8Array, hints: SendHints): void {}
 
   async connect(): Promise<void> {
     await Promise.all(new Array(100).fill(0).map(() => this.spawnPeer()))