Function fromSocket

packages/experimental/src/EventLogRemote.ts:250–422

(options?: {
  readonly disablePing?: boolean
})

 * @category construtors
 */
export const fromSocket = (options?: {
  readonly disablePing?: boolean
}): Effect.Effect<
  void,
  never,
  Scope.Scope | EventLog | EventLogEncryption | Socket.Socket
> =>
  Effect.gen(function*() {
    const log = yield* EventLog
    const socket = yield* Socket.Socket
    const encryption = yield* EventLogEncryption
    const scope = yield* Effect.scope
    const writeRaw = yield* socket.writer

    function* writeGen(request: typeof ProtocolRequest.Type) {
      const data = encodeRequest(request)
      if (request._tag !== "WriteEntries" || data.byteLength <= constChunkSize) {
        return yield* writeRaw(data)
      }
      const id = request.id
      for (const part of ChunkedMessage.split(id, data)) {
        yield* writeRaw(encodeRequest(part))
      }
    }

    const write = (request: typeof ProtocolRequest.Type) => Effect.gen(() => writeGen(request))

    yield* Effect.gen(function*() {
      let pendingCounter = 0
      const pending = new Map<number, {
        readonly entries: ReadonlyArray<Entry>
        readonly deferred: Deferred.Deferred<void>
        readonly publicKey: string
      }>()
      const chunks = new Map<number, {
        readonly parts: Array<Uint8Array>
        count: number
        bytes: number
      }>()

      const subscriptions = yield* RcMap.make({
        lookup: (publicKey: string) =>
          Effect.acquireRelease(
            Mailbox.make<RemoteEntry>(),
            (mailbox) =>
              Effect.zipRight(
                mailbox.shutdown,
                Effect.ignoreLogged(write(new StopChanges({ publicKey })))
              )
          )
      })
      const identities = new WeakMap<any, typeof Identity.Service>()
      const badPing = yield* Deferred.make<never, Error>()

      let latestPing = 0
      let latestPong = 0

      if (options?.disablePing !== true) {