Function createConsumer

apps/desktop/src/utils/shared-frame-buffer.ts:280–579

(buffer: SharedArrayBuffer)

}

export function createConsumer(buffer: SharedArrayBuffer): Consumer {
	const controlView = new Uint32Array(buffer, 0, 8);

	const storedVersion = controlView[CONTROL_VERSION];
	if (storedVersion !== PROTOCOL_VERSION) {
		throw new Error(
			`SharedFrameBuffer protocol version mismatch: expected ${PROTOCOL_VERSION}, got ${storedVersion}`,
		);
	}

	const slotCount = controlView[CONTROL_SLOT_COUNT];
	const slotSize = controlView[CONTROL_SLOT_SIZE];
	const metadataOffset = controlView[CONTROL_METADATA_OFFSET];
	const dataOffset = controlView[CONTROL_DATA_OFFSET];
	const metadataView = new Int32Array(buffer);

	return {
		read(timeoutMs: number = 100): ArrayBuffer | null {
			const MAX_CAS_RETRIES = 3;

			for (let attempt = 0; attempt < MAX_CAS_RETRIES; attempt++) {
				const shutdownFlag = Atomics.load(controlView, CONTROL_SHUTDOWN);
				if (shutdownFlag) {
					return null;
				}

				const readIdx = Atomics.load(controlView, CONTROL_READ_INDEX);
				const slotMetaIdx =
					(metadataOffset + readIdx * METADATA_ENTRY_SIZE) / 4;

				let state = Atomics.load(metadataView, slotMetaIdx + META_SLOT_STATE);

				if (state !== SLOT_STATE.READY) {
					const waitResult = Atomics.wait(
						metadataView,
						slotMetaIdx + META_SLOT_STATE,
						state,
						timeoutMs,
					);
					if (waitResult === "timed-out") {
						return null;
					}

					const shutdownCheck = Atomics.load(controlView, CONTROL_SHUTDOWN);
					if (shutdownCheck) {
						return null;
					}

					state = Atomics.load(metadataView, slotMetaIdx + META_SLOT_STATE);
					if (state !== SLOT_STATE.READY) {
						continue;
					}
				}

				const exchangedState = Atomics.compareExchange(
					metadataView,
					slotMetaIdx + META_SLOT_STATE,
					SLOT_STATE.READY,