Function decodeMessages

packages/cluster/src/MessageStorage.ts:547–598

(
    envelopes: Array<{
      readonly envelope: Envelope.Envelope.Encoded
      readonly lastSentReply: Option.Option<Reply.ReplyEncoded<any>>
    }>
  )

  })

  const decodeMessages = (
    envelopes: Array<{
      readonly envelope: Envelope.Envelope.Encoded
      readonly lastSentReply: Option.Option<Reply.ReplyEncoded<any>>
    }>
  ) => {
    const messages: Array<Message.Incoming<any>> = []
    let index = 0

    // if we have a malformed message, we should not return it and update
    // the storage with a defect
    const decodeMessage = Effect.catchAll(
      Effect.suspend(() => {
        const envelope = envelopes[index]
        if (!envelope) return Effect.succeed(undefined)
        return decodeEnvelopeWithReply(envelope)
      }),
      (error) => {
        const envelope = envelopes[index]
        return storage.saveReply(Reply.ReplyWithContext.fromDefect({
          id: snowflakeGen.unsafeNext(),
          requestId: Snowflake.Snowflake(envelope.envelope.requestId),
          defect: error.toString()
        })).pipe(
          Effect.forkDaemon,
          Effect.asVoid
        )
      }
    )
    return Effect.as(
      Effect.whileLoop({
        while: () => index < envelopes.length,
        body: () => decodeMessage,
        step: (message) => {
          const envelope = envelopes[index++]
          if (!message) return
          messages.push(
            message.envelope._tag === "Request"
              ? new Message.IncomingRequest({
                envelope: message.envelope,
                lastSentReply: envelope.lastSentReply,
                respond: storage.saveReply
              })
              : new Message.IncomingEnvelope({
                envelope: message.envelope
              })
          )
        }
      }),
      messages
    )
  }

  const decodeReplies = (
    messages: Map<string, Message.OutgoingRequest<any>>,