Глава 96 WebSocket Stream: перехват и поток событий

Ориентация главы

Кадры и сообщения главы 95 — машины состояний (транспортно-независимый уровень библиотеки); Stream этой главы (xwsstream) подключает их к реальному транспорту: xrtWsStreamAttach/AttachTls перехватывают ссылку вызывающего на TCP/TLS-поток уже после завершённого обновления (с точной стыковкой «в буфере остались данные кадров»), события публикуются как MessageBegin → MessageData×N → MessageEnd — представление тела действительно только внутри текущего обратного вызова; Stream не агрегирует целое сообщение и не выделяет фиксированный буфер на каждое соединение — существующая цепочка блоков TCP/TLS входит прямо в машину состояний кадров. Отправка даёт пути copy и ref по ссылке (ссылка освобождается только после фактического завершения записи в сокет — в групповой рассылке одна ссылка висит на N соединениях со счётным освобождением); SendLimit — единая жёсткая граница ожидания WebSocket и нижележащего транспорта, Backpressure/Writable/Drain дают фронты восстановления. Ping отвечает автоматически, Close гарантированно одноразово — полный жизненный цикл объекта соединения.

96.1引入

Упражнение в конце главы 95 предлагало собрать кадрировочный цикл руками — вы обнаружили, что 80% кода возит «транспортную коммутацию»: в буфере после обновления уже могут лежать данные кадров, заголовок кадра может пересекать TLS-record, противодавление записи надо смотреть в двух слоях сразу, Ping просит автоответа, Close можно отправить лишь раз... Эти трубопроводы, переписываемые каждой реализацией, слой Stream собирает в один вызов Attach плюс четыре обратных вызова событий.

Первое, что стоит понять, — стыковка перехвата: ответ обновления только что отправлен, а в приёмном буфере уже может ждать первый кадр от нетерпеливого клиента (браузеры так делают часто). Параметр iPrefix сообщает Stream «длину уже проверенного, но не потреблённого HTTP-заголовка» — при перехвате копируется согласованный подпротокол, заголовок потребляется точно, и остаток кадров в том же буфере входит прямо в машину состояний кадров — ни один байт не пропадает и не смещается. Отказ никогда не перехватывает наполовину (ссылка транспорта возвращается нетронутой).

96.2概念

Перехват: точная стыковка Attach

Особенность TLS: заголовок кадра может пересекать границу TLS-record — Stream накапливает до 14 байтов полного заголовка кадра (максимальный размер заголовка) через путь ReadMore, ограниченный PlainLimit, — «не останавливаться из-за сохранения неполного префикса» — стыковка деталей TLS-потока главы 87 с машиной состояний кадров этой главы уже отлажена.

Модель событий: представление использовал — выбросил

Представление тела MessageData действительно только внутри текущего синхронного обратного вызова — Stream не агрегирует целое сообщение (большое сообщение не занимает память) и не копирует (цепочка блоков проходит насквозь). Логика потребления обязана «завершиться внутри обратного вызова» (разбор, пересылка, запись на диск); нужно сохранить — копируйте сами. Это та же дисциплина, что потребление открытого текста TLS-потока главы 87 и представления Data тела главы 91 — в мире заимствованных представлений сохранение это явное действие.

Три формы отправки и единое противодавление

• copy (Text/Binary/Send): нагрузка копируется в очередь — простой путь малых сообщений.
• ref (TextRef/BinaryTake/SendRef и др.): передаётся xnetref{данные, длина, освобождающий обратный вызов, контекст} — ноль копирования, пока стоит в очереди; освобождающий обратный вызов вызывается только после фактической записи в сокет. В групповой рассылке одна ссылка висит на N соединениях (счётное освобождение) — одно распределение, N отправок.
• сжатая отправка (TextCompressed и др.): Deflater создаётся по надобности — путь после согласования permessage-deflate (глава 97).

Противодавление — единый уровень воды: SendLimit одновременно считает ожидание WebSocket и ожидание нижележащего транспорта — достижение потолка возвращает XNET_RESULT_AGAIN; события Backpressure/Writable/Drain дают фронты восстановления (вход в противодавление, запись снова возможна, осушение). В одном ряду с бюджетом записи TCP главы 68 и SendLimit TLS главы 87 — по уровню воды на слой, семантика одна.

Контрольные кадры и протокол закрытия

Ping отвечает Pong автоматически (приложение не замечает); Ping можно отправить и по своей инициативе. Гарантии Close: одноразовая отправка (повторный Close не выходит в сеть), механизм тайм-аута (молчащая противоположная сторона завершается по тайм-ауту), снимок финального состояния противоположной стороны (CloseInfo: код + причина, для чтения приложением). Семантические домены кодов закрытия и проверка на стороне записи — материал главы 97; здесь запомните «Close вызывается один раз, протокол ведает машина состояний».

Жизненный цикл и самосознание

Ref/Destroy со счётом ссылок (как у прочих объектов соединения); семейство самосознания: State/Role/Protocol/Worker/Tcp/TcpRef/Tls/TlsRef/Deflate/Pending/Writable — все публичные факты соединения; Pause/Paused/Resume — управление потоком (приостановка приёма в паре с ограничением скорости ниже по течению).

96.3示例

Первая полная программа: отправка по ссылке — освобождение после записи

Программа ниже из examples/websocket/stream_ref/main.c — стандартный образец пути отправки без копий:

#include <stdio.h>
#include <string.h>

#include <xrt.h>





















 */


/* 引用发送完成后归还业务负载。 */
static void releasePayload(
	ptr pContext,
	cbytes pData,
	size_t iSize
)
{
	(void)pContext;
	(void)iSize;
	xrtFree((ptr)pData);
}



/* 在 Connection 所属 Worker 上发送一个可延迟释放的业务负载。 */
static xnetresult sendPayload(
	xwsstream* pConnection,
	xbytesview Payload
)
{
	bytes pOwned;
	xnetref Ref;
	xnetresult Result;

	pOwned = (bytes)xrtMalloc(Payload.Size);
	if ( pOwned == NULL ) {
		return XNET_RESULT_ERROR;
	}
	memcpy(pOwned, Payload.Data, Payload.Size);
	Ref = (xnetref) {
		pOwned,
		Payload.Size,
		releasePayload,
		NULL
	};
	Result = xrtWsStreamBinaryRef(pConnection, &Ref);
	if ( Result != XNET_RESULT_OK ) {
		xrtFree(pOwned);
	}
	return Result;
}



/* 示例函数由已经建立连接的 Worker 回调调用。 */
int main(void)
{
	(void)sendPayload;
	printf("WebSocket Connection reference example is ready\n");
	return 0;
}

Что только что произошло. (1) Бизнес-нагрузка выделена xrtMalloc-ом (имитация сообщения от верхнего потока); xnetref пакует {указатель, длина, обратный вызов releasePayload, контекст}. (2) xrtWsStreamBinaryRef(连接, 负载, Ref) (соединение, нагрузка, Ref) ставит в очередь по ссылке — нагрузка не копируется, пока стоит в очереди Worker; только после фактического завершения отправки (записи в сокет) вызывается releasePayload и xrtFree возвращает память. (3) Почему это важно: в высокопроизводительных сценах путь copy — это распределение + копия на каждую отправку; путь ref — одно распределение, ноль транзитных копий.Групповое продвижение (шапка комментария указывает путь): одна ссылка вешается на N соединений — внутреннее счётное освобождение; только последний дописавшийся действительно освобождает. (4) Pending/Writable в паре ждут осушения — так читается единое противодавление.

Сопроводительный examples/websocket/stream_tour — полный обход возможностей слоя Stream (петля, оба конца): проверка конфигурации, Attach на обоих концах, эхо 64 байтов, автоматический и ручной Pong, приостановка-восстановление, три формы сжатия, чистое закрытие (code=1000) — прогоните его как исполняемую таблицу контрактов главы.

Вторая полная программа: предвестие согласования расширений

Вторая программа из examples/websocket/extension/main.c — обход параметров расширений после обновления, подготовка главы 97:

#include <stdio.h>

#include <xrt.h>






















 */


/* 展示逐项读取 Sec-WebSocket-Extensions 和扩展参数。 */
int main(void)
{
	xstrview Text = XRT_STR_LITERAL(
		"permessage-deflate; client_max_window_bits, x-trace"
	);
	xwsextension Extension;
	xhttpnext Next;
	size_t iOffset = 0;

	for ( ;; ) {
		xhttpparam Param;
		size_t iParam = 0;

		Next = xrtWsExtensionNext(
			Text,
			&iOffset,
			&Extension
		);
		if ( Next == XHTTP_NEXT_END ) {
			break;
		}
		if ( Next == XHTTP_NEXT_ERROR ) {
			return 1;
		}
		printf(
			"extension=%.*s\n",
			(int)Extension.Name.Size,
			Extension.Name.Data
		);
		while ( xrtWsExtensionParamNext(
			&Extension,
			&iParam,
			&Param
		) == XHTTP_NEXT_ITEM ) {
			printf(
				"  parameter=%.*s\n",
				(int)Param.Name.Size,
				Param.Name.Data
			);
		}
	}
	return 0;
}

Что только что произошло. (1) Двухуровневая структура Sec-WebSocket-Extensions: внешний уровень режет расширения по ,, внутренний режет параметры по ; — изоморфно грамматике семейств полей главы 92, но у уровня расширений свой итератор (пользовательские расширения вроде x-trace тоже выражаются как name (имя) + параметры). (2) client_max_window_bits — один из параметров согласования permessage-deflate; глава 97 полностью разбирает согласование и цепочку сжатия; здесь закладывается понимание «расширения — свойство соединения после обновления, и согласование решает поведение Stream (пути сжатия, биты RSV)». (3) Итерация расширений — чисто полевая операция: отладочные прокси и контролёры соответствия аудируют результат согласования, не строя соединения.

96.4契约

• Атомарность перехвата: неудавшийся Attach никогда не перехватывает наполовину (ссылка транспорта возвращается нетронутой); iPrefix точно потребляет проверенный заголовок, остаток кадров того же буфера входит в машину состояний напрямую.
• Представления событий: тело MessageData действительно только внутри текущего обратного вызова; без агрегации целых сообщений, без фиксированного буфера на соединение — цепочка блоков проходит прямо в машину состояний кадров.
• TLS через record: заголовки кадров, пересекающие TLS-record, накапливаются через ReadMore (≤14 байтов, под ограничением PlainLimit) — без остановки и без потерь.
• Три формы отправки: copy (копия в очередь) / ref (ссылка + отложенный освобождающий обратный вызов) / compressed (Deflater по надобности); семантика ёмкости и бюджета общебиблиотечная.
• Единое противодавление: SendLimit считает вместе WS и транспорт; AGAIN + события фронтов Backpressure/Writable/Drain.
• Отправка по ссылке: xnetref{данные, длина, освобождение, контекст}; освобождение только после записи в сокет; одна ссылка может висеть на N соединениях со счётным освобождением.
• Ping/Pong: полученный Ping автоматически отвечает Pong; активный Ping отправить можно; управление потоком Pause/Paused/Resume.
• Гарантии Close: одноразовая отправка, механизм тайм-аута, снимок финального состояния противоположной стороны (CloseInfo); Abort прерывает немедленно.
• Семейство самосознания: State/Role/Protocol/Worker/Tcp(+Ref)/Tls(+Ref)/Deflate/Pending/Writable.
• Обрезка: WEBSOCKET_STREAM независимо от слоя кадров; под-возможности _REF/_TLS/_DEFLATE по надобности.

Типичные ошибки

onMessageData(pStream, Data, ...) {
	queue_push(g_Queue, Data);   /* сохранили представление: висяче по возвращении */
}

onMessageData(pStream, Data, ...) {
	bytes pCopy = (bytes)xrtMalloc(Data.Size);
	memcpy(pCopy, Data.Data, Data.Size);
	queue_push(g_Queue, pCopy);   /* копия может идти асинхронно */
}
/* ещё лучше: завершить потребление синхронно (разбор/пересылка), не вводя очередь */

xrtWsStreamBinaryRef(pStream, PayloadView, &Ref);
xrtFree(pBuffer);   /* освободили сразу после постановки: ссылка в очереди висит */

/* владение переходит со ссылкой; освобождение - внутри обратного вызова releasePayload */
xrtWsStreamBinaryRef(pStream, PayloadView, &Ref);
/* с этого момента pBuffer - забота Stream: не трогать */

if ( my_ws_queue_size() < MY_LIMIT ) {
	send_more();   /* транспортный слой мог давно перейти предел */
}

if ( xrtWsStreamSend(pStream, ...) == XNET_RESULT_AGAIN ) {
	/* ждать события Writable/Drain - фронты восстановления единого уровня */
	return;
}

Практические задания

Итоги