Queue

Purpose

The Queue module provides the FIFO processing queue behind each pipeline stage, and the two specialised queues built on it: the seal queue (plaintext packets in, sealed frames out) and the open queue (frames in, plaintext packets out). Strict one-at-a-time processing is what lets a protocol assign a monotonically increasing counter to every frame.

Key Interfaces

Queue<T>

interface Queue<T extends object> {
  readonly addMessage: (message: T) => void // Enqueue; starts processing when autoStart is on
  readonly isRunning: () => boolean // Whether the queue is processing
  readonly stop: () => void // Pause processing (messages accumulate)
  readonly resume: () => void // Resume processing accumulated messages
  readonly size: () => number // Number of messages waiting
  readonly currentMessage: () => T | null // The message being processed
}

MessageHandler<T>

type MessageHandler<T extends object> = (message: T) => Promise<void> | void

QueueFailureHandler

Called with the rejected input, why it was rejected, and the error the operation threw when it threw one.

type QueueFailureHandler = (raw: unknown, reason: string, cause?: unknown) => void

QueueOperation, QueueCreatorArguments, QueueCreatorValidity

type QueueOperation = PacketSealer | PacketOpener

interface QueueCreatorArguments<T = any> {
  label: string
  operation: QueueOperation
  logger: Logger
  onSuccess: (packet: T) => void
  onFail: QueueFailureHandler
}

interface QueueCreatorValidity {
  label: boolean
  operation: boolean
  logger: boolean
  onSuccess: boolean
  onFail: boolean
}

SealQueueCreater and OpenQueueCreater

type SealQueueCreater = (
  label: string,
  seal: PacketSealer,
  logger: Logger,
  onSuccess: (packet: WirePacket) => void,
  onFail: QueueFailureHandler
) => Queue<UnencryptedPacket>

type OpenQueueCreater = (
  label: string,
  open: PacketOpener,
  logger: Logger,
  onSuccess: (packet: UnencryptedPacket) => void,
  onFail: QueueFailureHandler
) => Queue<WirePacket>

Queue Types

Factory Functions

createQueue<T>

Location: @hyperfrontend/network-protocol/queue

function createQueue<T extends Record<string, any>>(processMessage: MessageHandler<T>, autoStart = true): Queue<T>

Messages are processed strictly one at a time in arrival order; the next one starts only after the handler's promise settles. The backing store is an array with a moving head, so a pull is constant time; once 1024 pulled slots sit at the front the array is compacted.

import { createQueue } from '@hyperfrontend/network-protocol/queue'

const queue = createQueue<{ id: string }>(async (message) => {
  await handle(message)
})
queue.addMessage({ id: '1' }) // starts processing at once

Throws processMessage must be a function and autoStart must be a boolean at creation; addMessage throws a TypeError (Message must be a non-null object) for anything that is not an object.

createSealQueue

import { createSealQueue } from '@hyperfrontend/network-protocol/queue'

const sealing = createSealQueue(
  'comms sender',
  protocol.seal,
  logger,
  (frame) => transport.post(frame),
  (packet, reason, cause) => report(reason, cause)
)
sealing.addMessage(unencryptedPacket)

Each packet is checked with isValidUnencryptedPacket, sealed, and the result checked with isValidWirePacket before onSuccess.

createOpenQueue

import { createOpenQueue } from '@hyperfrontend/network-protocol/queue'

const opening = createOpenQueue(
  'comms receiver',
  protocol.open,
  logger,
  (packet) => deliver(packet),
  (frame, reason, cause) => report(reason, cause)
)
opening.addMessage(frame)

Each frame is checked with isValidWirePacket, opened, and the result checked with isValidUnencryptedPacket before onSuccess.

Lifecycle Management

queue.stop()
queue.isRunning() // false once the current message settles

queue.addMessage(a) // accumulates while stopped
queue.addMessage(b)
queue.size() // 2

queue.resume() // processes a, then b

currentMessage() returns the message in flight, or null between messages.

Error Handling

A rejected input never blocks the queue: the stage logs it, calls onFail, and moves on to the next message.

The sender and receiver translate these calls into PacketDrop reports for the channel's onDrop.

Validation Errors

The specialised creators validate their arguments and throw Cannot create seal queue without ... or Cannot create open queue without ... followed by a label, seal function / open function, a logger, a success callback function, or a failed callback function.

Relationship to Other Modules

• Depends on: packet/ (packet types and validations), @hyperfrontend/logging
• Used by: sender/, receiver/

See Also

• Library Index - All modules
• Architecture Guide - Queue architecture
• Queue Entry - The @hyperfrontend/network-protocol/queue entry

Related Modules