Skip to main content
Logo for Custom Channel Runtime

Custom Channel Runtime

info

This is still a work in progress as websockets have a few different aspects in order to work in a distributed fashion.

There are two different ways we can deal with channels, both local and serverless. The implementation is slightly different, since running them locally (like with uWebSockets or ws) means we can skip a lot of serverless state management and achieve better performance, but it doesn’t scale the same way or offer the benefits of serverless deployments.

Interfaces Used​

Creating a websocket handler requires us to understand a few concepts:

  1. EventHubStore
  2. ChannelStore

Local Channels​

For local channels, the lifecycle events mirror the typical WebSocket flow:

1) onUpgrade​

When a connection is upgraded, we obtain a channel handler which is our interface to the Pikku framework:

const request = new PikkuUWSRequest(req, res)
const response = new PikkuUWSResponse(res)

const channelHandler = await runLocalChannel({
  channelId: crypto.randomUUID().toString(),
  request,
  response,
  singletonServices: singletonServicesWithEventHub,
  createSessionServices,
  route: req.getUrl() as string,
})

2) onOpen​

Upon opening the connection, we:

  • Register the send method on the WebSocket.
  • Notify the EventHub that the channel has been opened.
  • Invoke the channel handler’s open logic.
info

We perform both these actions so that Pikku remains minimal and does not need to wrap the underlying WebSocket objects.

open: (ws) => {
  const { channelHandler } = ws.getUserData()
  channelHandler.registerOnSend((data) => {
    if (isSerializable(data)) {
      ws.send(JSON.stringify(data))
    } else {
      ws.send(data as any)
    }
  })
  eventHub.onChannelOpened(channelHandler.channelId, ws)
  channelHandler.open()
}

3) onMessage​

Incoming messages are decoded and passed to the channel handler. If a response is generated (supporting direct responses similar to AWS Lambda WebSocket functions), it is sent back to the client:

message: async (ws, message, isBinary) => {
  const { channelHandler } = ws.getUserData()
  const data = isBinary ? message : decoder.decode(message)
  const result = await channelHandler.message(data)
  if (result) {
    // TODO: Handle binary responses as needed
    ws.send(JSON.stringify(result))
  }
}

4) onClose​

When the connection closes, we clean up by notifying the EventHub and closing the channel handler:

close: (ws) => {
  const { channelHandler } = ws.getUserData()
  eventHub.onChannelClosed(channelHandler.channelId)
  channelHandler.close()
}

Serverless Integration​

Serverless WebSocket setups typically invoke functions on-demand without retaining state on the same machine that handles the WebSocket connection. Imagine a load balancer accepting WebSocket connections and distributing events across multiple processes. To emulate a local-like experience, we must store both EventHub connections and channel state in a distributed memory.

1) onOpen – Establishing the Connection​

When a new WebSocket connection is initiated via AWS Lambda, the connection details are processed through a dedicated function. This function sets up the channel by storing its state in distributed memory and wiring it up to the framework.

websocket-disconnect.ts
loading...

2) onMessage – Handling Incoming Messages​

For serverless environments, incoming messages are processed by a separate function. This function decodes the incoming data and dispatches it to the appropriate channel handler, returning any responses generated by user code.

websocket-disconnect.ts
loading...

3) onClose – Cleaning Up the Connection​

When a client disconnects, a dedicated function handles the cleanup. It removes the channel’s state from distributed memory and notifies the EventHub that the connection has been closed.

websocket-disconnect.ts
loading...

This section completes the overview of custom channel runtime in Pikku. For local channels, state is managed directly via WebSocket lifecycle events, while in serverless environments, the framework leverages distributed memory to maintain state and ensure that user code interacts with channels as if they were local.

Feel free to explore and provide feedback as we continue to refine these integrations!