hypermosh-shell / lib / frames.mjs

// frames — the typed wire format shared by both modes, carried over one reliable
// hyperdht stream:
//
//   [uint8 type][uint32 BE len][payload]
//
//   T_CTRL (0): JSON control message
//   T_DATA (1): one raw mosh UDP datagram (one frame == one datagram)
//   T_SSH  (2): a chunk of the tunnelled SSH TCP byte stream (SSH-style mode)
//   T_SSH_END (3): the SSH TCP side closed (empty payload)
//
// A single stream multiplexes the SSH bootstrap (T_SSH) and, once mosh-server is
// up, the mosh datagrams (T_DATA) — no protomux needed, the type byte demuxes.
import b4a from 'b4a'

export const T_CTRL = 0
export const T_DATA = 1
export const T_SSH = 2
export const T_SSH_END = 3

export function frame (type, payload) {
  const h = b4a.alloc(5)
  h.writeUInt8(type, 0)
  h.writeUInt32BE(payload ? payload.length : 0, 1)
  return payload && payload.length ? b4a.concat([h, payload]) : h
}

export const ctrlFrame = (obj) => frame(T_CTRL, b4a.from(JSON.stringify(obj)))
export const dataFrame = (dg) => frame(T_DATA, dg)
export const sshFrame = (chunk) => frame(T_SSH, chunk)
export const sshEndFrame = () => frame(T_SSH_END, null)

// makeParser(handlers) -> (chunk) => void. handlers: { onCtrl, onData, onSsh, onSshEnd }.
// Unknown/absent handlers are ignored so old and new peers stay wire-tolerant.
export function makeParser ({ onCtrl, onData, onSsh, onSshEnd } = {}) {
  let buf = b4a.alloc(0)
  return (chunk) => {
    buf = b4a.concat([buf, chunk])
    while (buf.length >= 5) {
      const type = buf.readUInt8(0)
      const len = buf.readUInt32BE(1)
      if (buf.length < 5 + len) break
      const payload = buf.subarray(5, 5 + len)
      buf = buf.subarray(5 + len)
      if (type === T_CTRL) { if (onCtrl) { try { onCtrl(JSON.parse(b4a.toString(payload))) } catch {} } }
      else if (type === T_DATA) { if (onData) onData(payload) }
      else if (type === T_SSH) { if (onSsh) onSsh(payload) }
      else if (type === T_SSH_END) { if (onSshEnd) onSshEnd() }
    }
  }
}