Глава 107 SSE: push с сервера

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

Для одностороннего push от сервера к клиенту у Web-мира два варианта: WebSocket (главы 95–97, двунаправленный, отдельный протокол) и SSE (Server-Sent Events — протокол событий поверх потока HTTP-ответа). Философия реализации xhttp одной фразой: никаких выделенных SSE-соединений, потоков, очередей или сетевых машин состояний — xrtHttpSseReplyCreate создаёт обычный xhttpreply со статусом 200, Content-Type: text/event-stream и одноразовым телом неизвестной длины; фактическую отправку ведут единственная машина состояний ответа HTTP Server, противодавление TCP/TLS и срок записи; возвращённый xhttpbodystream — независимый конец-производитель, отдаваемый фоновой задаче или любому потоку на запись. Клиентская сторона — то же зеркало: слой соединения плюс парсер событий (инкрементальная машина синтаксиса text/event-stream), разрывное возобновление Last-Event-ID встроено. Правило выбора: когда push односторонний — берите SSE — протокол прост, едет по порту 80, автопереподключение встроено в браузер.

107.1引入

Push прогресса, потоки котировок, хвост журнала, потоковая выдача LLM по токенам — «сервер говорит без остановки, клиент только слушает» покрывает большую часть потребностей push. Тащить сюда WebSocket — зенитка по комарам: одностороннее использование двунаправленного протокола плюс полный груз handshake-обновления, сердцебиений и протокола закрытия. Ответ SSE — «не меняйте протокол»: по-прежнему HTTP-ответ, просто не заканчивающийся — в chunked-потоке с Content-Type: text/event-stream события составляются из строк полей event:/data:/id:/retry:, разделённых пустыми строками.

Инженерная трудность на обоих концах. Сервер: поток ответа должен сотрудничать с фоновыми производителями — события приходят из потока задачи, соединение пишется на Worker — как путешествует противодавление? Клиент: синтаксис событий инкрементальный (одно поле data может быть разбито на несколько data:-строк; события пересекают границы chunk) — парсер обязан быть потоковым. xhttp даёт структурированные ответы на обоих концах: серверный SseSendEvent (одно полное событие за вызов, сперва валидация потом кодирование, половина события при провале не фиксируется) + конец-производитель xhttpbodystream (противодавление AGAIN + восстановление WaitWritable); клиентский парсер продвигается поле за полем под ограничениями потолков.

107.2概念

Сервер: поезд на единственной машине состояний

Три проектные дисциплины. (1) Граница протокольно-необходимого и прикладного выбора: ReplyCreate ставит только Content-Type; Cache-Control: no-cache, аутентификацию, CORS, политику буферизации прокси (X-Accel-Buffering: no) приложение ставит до фиксации обычными Header-API — слой SSE не гадает за вашу среду развёртывания. (2) Атомарность: каждый Send сперва полностью валидирует и измеряет, потом кодирует прямо в один узел Body Stream — временная строка события не строится, половина события при провале не фиксируется; жёсткие бюджеты байтов и Chunk покрывают конкурентные резервирования, очередь и активные лизинги. (3) Раздельные жизненные циклы: после фиксации Respond Reply можно уничтожить, Stream живёт дальше — «объект ответа» и «конец-производитель» — два жизненных цикла.

Структура события и упаковка

Четыре элемента xhttpsseevent: Type (строка event: — имя типа события), Data (строки data: — может быть многострочной), Id (строка id: — курсор возобновления), Retry (строка retry: — миллисекунды переподключения, рекомендуемые клиенту); Flags объявляет, какие элементы присутствуют. SseSendEvent упаковывает полное событие одним вызовом; SseSendComment (строка комментария : heartbeat — поддержание жизни); xrtHttpBodyStreamWrite/WriteRef/WriteTake пишут сырые байты напрямую (предкодированные события, пересылка прокси, собственные расширения — структурированный Writer не обязательный путь).

Противодавление и ожидание

Запись производителя, упёршаяся в полный бюджет, возвращает AGAIN — ни одной ссылки на ввод не удержано, вызывающий повторяет в точности; xrtHttpBodyStreamWaitWritable возвращает общий Future следующего поколения записываемости — отмена одного ждущего влияет на остальных ждущих, поэтому напрямую отменять нельзя (общая семантика Future главы 58). Это «AGAIN + WaitWritable» — тот же словарь, что у Stream WebSocket главы 96.

Клиент: соединение и парсер

Клиентская сторона — два слоя: слой соединения (http_sse_client — инициирует GET, поддерживает переподключение и заголовок Last-Event-ID) и парсер (инкрементальная машина синтаксиса text/event-stream — строки полей, границы событий, склейка многострочных data). Конфигурация парсера ограничивает одиночную строку, данные события, тип и ID — структурный потолок памяти (SSE-версия «что ограничивать, решает структура содержимого» главы 99: тело представления выбрано безграничным, структурированная память ограничена). Разрывное возобновление: отправленный сервером id: записывается клиентом и при переподключении приносится обратно заголовком Last-Event-ID — сервер продолжает после курсора.

Выбор SSE против WebSocket

107.3示例

Первая полная программа: события, сердцебиение и прямая запись — три пути

Программа ниже из examples/http/sse_server — полная грань структурированной отправки событий:

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



/* 读取并打印 Reply 的完整异步正文。 */
static bool exampleReadReply(xhttpreply* pReply)
{
	xhttpbodyreader* pReader = xrtHttpBodyOpen(
		xrtHttpReplyBody(pReply)
	);
	xhttpbodychunk Chunk;
	xhttpbodystatus Status;

	if ( pReader == NULL ) {
		return false;
	}
	for ( ;; ) {
		Status = xrtHttpBodyNext(pReader, 8, &Chunk);
		if ( Status == XHTTP_BODY_DATA ) {
			(void)fwrite(Chunk.Data, 1, Chunk.Size, stdout);
			xrtHttpBodyChunkRelease(&Chunk);
			continue;
		}
		if ( Status == XHTTP_BODY_AGAIN ) {
			xfuture* pReady = xrtHttpBodyReaderWait(pReader);

			if ( (pReady == NULL) ||
				(xrtFutureWait(pReady) != XWAIT_OK) ||
				(xrtFutureState(pReady) != XFUTURE_RESOLVED) ) {
				xrtFutureDestroy(pReady);
				xrtHttpBodyReaderDestroy(pReader);
				return false;
			}
			xrtFutureDestroy(pReady);
			continue;
		}
		break;
	}
	xrtHttpBodyReaderDestroy(pReader);
	return Status == XHTTP_BODY_EOF;
}



/* 创建 SSE Reply，发送结构化事件、心跳和预封装字节。 */
int main(void)
{
	xhttpbodystreamconfig Config;
	xhttpbodystream* pStream = NULL;
	xhttpreply* pReply;
	xhttpsseevent Event;
	bool bResult;

	xrtHttpBodyStreamConfigInit(&Config);
	Config.MaxBytes = 4096;
	Config.MaxChunks = 32;
	pReply = xrtHttpSseReplyCreate(&Config, &pStream);
	if ( (pReply == NULL) || (pStream == NULL) ) {
		xrtHttpBodyStreamDestroy(pStream);
		xrtHttpReplyDestroy(pReply);
		return 1;
	}

	memset(&Event, 0, sizeof(Event));
	Event.Type = XRT_STR_LITERAL("progress");
	Event.Data = XRT_STR_LITERAL("{\"percent\":75}");
	Event.Id = XRT_STR_LITERAL("job-42:3");
	Event.Retry = 2000;
	Event.Flags = XHTTP_SSE_EVENT_TYPE |
		XHTTP_SSE_EVENT_DATA |
		XHTTP_SSE_EVENT_ID |
		XHTTP_SSE_EVENT_RETRY;
	if ( (xrtHttpSseSendEvent(
			pStream, &Event
		) != XHTTP_BODY_STREAM_OK) ||
		(xrtHttpSseSendComment(
			pStream, XRT_STR_LITERAL("heartbeat")
		) != XHTTP_BODY_STREAM_OK) ||
		(xrtHttpBodyStreamWrite(
			pStream, XRT_BYTES_LITERAL("data: ready\n\n")
		) != XHTTP_BODY_STREAM_OK) ) {
		xrtHttpBodyStreamDestroy(pStream);
		xrtHttpReplyDestroy(pReply);
		return 1;
	}

	xrtHttpBodyStreamDestroy(pStream);
	bResult = exampleReadReply(pReply);
	xrtHttpReplyDestroy(pReply);
	return bResult ? 0 : 1;
}

Что только что произошло. (1) BodyStreamConfigInit ставит MaxBytes/MaxChunks (бюджет конца-производителя — источник порога AGAIN), затем SseReplyCreate даёт пару объектов Reply и Stream. (2) Три формы записи одной очередью: SseSendEvent (событие progress со всеми четырьмя элементами — type/data/id/retry упакованы разом), SseSendComment (строка комментария heartbeat — поддержание жизни без события), BodyStreamWrite (предупакованные байты data: ready\n\n записаны напрямую — путь для прокси/статических событий). (3) Проверка обратным чтением (exampleReadReply образца): читателем тела главы 108 тело Reply вычитывается потоково и печатается по кускам — байты, упакованные сервером, и есть синтаксис, который ест клиентский парсер; один образец сверяет оба конца. (4) Жизненный цикл пары объектов: Stream по завершении Destroy (EOF опубликован), Reply уничтожается следом — согласовано с контрактом «после фиксации Reply можно уничтожить, Stream независим» (этот образец не фиксирует на соединение, а читает обратно локально).

Вторая полная программа: форма ожидания при потоковом чтении клиента

Вторая программа из examples/http/sse_client — стандартный цикл потребляющей стороны клиента:

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



/* 示例上下文只同步唯一终态，不在回调和宿主线程间共享借用视图。 */
typedef struct sse_example {
	xatomic32 Closed;
} sse_example;



/* 输出一段不保证以零字符结尾的 SSE 文本视图。 */
static void printView(cstr sName, xstrview Value)
{
	printf("%s: ", sName);
	if ( Value.Size != 0 ) {
		(void)fwrite(Value.Data, 1, Value.Size, stdout);
	}
	(void)fputc('\n', stdout);
}



/* 报告已经通过状态码和媒体类型校验的响应。 */
static bool onOpen(
	xhttpsseclient* pClient,
	const xhttpresponse* pResponse,
	ptr pData
)
{
	(void)pClient;
	(void)pData;
	printf("open: HTTP %u\n", (unsigned)xrtHttpResponseStatus(pResponse));
	return true;
}



/* 在网络 Worker 回调内消费借用消息，不把 Parser 视图带出回调。 */
static bool onMessage(
	xhttpsseclient* pClient,
	const xhttpssemessage* pMessage,
	ptr pData
)
{
	(void)pClient;
	(void)pData;
	printView("event", pMessage->Type);
	printView("id", pMessage->LastEventId);
	printView("data", pMessage->Data);
	(void)fputc('\n', stdout);
	return true;
}



/* 报告暂态断开；Client 随后按协商后的毫秒延迟自动重连。 */
static void onRetrying(
	xhttpsseclient* pClient,
	size_t iReconnect,
	uint64 iDelay,
	const xerror* pError,
	ptr pData
)
{
	(void)pClient;
	(void)pData;
	fprintf(
		stderr,
		"reconnect %zu after %llu ms: %s\n",
		iReconnect,
		(unsigned long long)iDelay,
		pError != NULL ? xrtErrorMessage(pError) : "stream ended"
	);
}



/* 发布唯一终态并唤醒示例宿主线程。 */
static void onClose(
	xhttpsseclient* pClient,
	xhttpsseclosereason Reason,
	const xerror* pError,
	ptr pData
)
{
	sse_example* pExample = (sse_example*)pData;

	(void)pClient;
	fprintf(
		stderr,
		"closed (%d): %s\n",
		(int)Reason,
		pError != NULL ? xrtErrorMessage(pError) : "normal stop"
	);
	xrtAtomic32Store(&pExample->Closed, 1, XMEMORY_RELEASE);
}



/* 连接一个 SSE URL，打印消息并等待服务端 204 或永久错误终止会话。 */
int main(int argc, char** argv)
{
	xnetengineconfig EngineConfig;
	xhttpclientconfig HttpConfig;
	xhttpsseclientevents Events;
	sse_example Example;
	xnetengine* pEngine = NULL;
	xhttpclient* pHttp = NULL;
	xhttpsseclient* pSse = NULL;
	int iExit = 1;

	if ( argc != 2 ) {
		fprintf(stderr, "usage: sse_client <http-url>\n");
		return 0;
	}
	memset(&Example, 0, sizeof(Example));
	memset(&Events, 0, sizeof(Events));
	Events.Open = onOpen;
	Events.Message = onMessage;
	Events.Retrying = onRetrying;
	Events.Close = onClose;
	Events.Data = &Example;

	xrtNetEngineConfigInit(&EngineConfig);
	pEngine = xrtNetEngineCreate(&EngineConfig);
	if ( (pEngine == NULL) || !xrtNetEngineStart(pEngine) ) {
		goto cleanup;
	}
	xrtHttpClientConfigInit(&HttpConfig);
	pHttp = xrtHttpClientCreate(pEngine, &HttpConfig);
	if ( pHttp == NULL ) {
		goto cleanup;
	}
	pSse = xrtHttpSseConnect(
		pHttp,
		(xstrview){ argv[1], strlen(argv[1]) },
		NULL,
		&Events
	);
	if ( pSse == NULL ) {
		goto cleanup;
	}

	while ( xrtAtomic32Load(&Example.Closed, XMEMORY_ACQUIRE) == 0 ) {
		xrtSleep(10);
	}
	iExit = xrtHttpSseClientError(pSse) == NULL ? 0 : 1;

cleanup:
	xrtHttpSseClientDestroy(pSse);
	xrtHttpClientDestroy(pHttp);
	if ( pEngine != NULL ) {
		while ( !xrtNetEngineDestroy(pEngine) ) {
			xrtClearError();
			xrtThreadYield();
		}
	}
	return iExit;
}

Что только что произошло. (1) Три состояния цикла чтения тела: XHTTP_BODY_DATA (взять кусок, потребить, ChunkRelease), XHTTP_BODY_AGAIN (данных пока нет — xrtHttpBodyReaderWait берёт Future и ждёт прибытия), EOF (поток завершается нормально). Пара AGAIN + Wait — универсальный каркас потокового потребления (симметричен серверному AGAIN + WaitWritable). (2) Слой соединения SSE-клиента поддерживает поверх: при разрыве переподключение по рекомендации Retry, курсор Last-Event-ID приносится обратно; парсер событий превращает поток байтов в структурированные обратные вызовы событий. (3) Сопроводительный examples/http/sse_http (голая HTTP-форма без клиентского слоя — поток ответа потребляется напрямую) закрывает сценарий «нужен поток, а не переподключение».

107.4契约

• Нет выделенной машины состояний: SSE Reply — обычный Reply (200/text-event-stream/неизвестная длина); отправка едет на единственной машине состояний ответа с существующими противодавлением и сроком записи.
• Граница протокол/приложение: ReplyCreate ставит только Content-Type; Cache-Control/CORS/политика буферизации — ставьте сами до фиксации.
• Жизненный цикл: после Respond Reply можно уничтожить, Stream независим (его держит поток задачи/издатель); разрушение последней ссылки = EOF после очереди событий; Close идемпотентен; Fail отбрасывает недоставленное + стабильная Cause.
• Атомарность записи: сперва валидация и измерение, потом кодирование в один узел; временная строка события не строится; половины события при провале нет; бюджет байтов/Chunk покрывает резервирования + очередь + лизинги.
• Противодавление: AGAIN не удерживает ввод; WaitWritable — общий Future (без односторонней отмены); прямые записи байтов Write/WriteRef/WriteTake опциональны.
• Четыре элемента события: event/data/id/retry + Flags; строка комментария поддерживает жизнь; id — курсор возобновления.
• Клиентский разбор: инкрементальная машина синтаксиса; четыре потолка строка/данные/тип/ID — структурированная память ограничена; возобновление Last-Event-ID.
• Цикл потребления: DATA (кусок + Release) / AGAIN (Wait Future) / EOF — каркас трёх состояний.
• Обрезка: HTTP_SSE сервер/клиент независимы; парсер можно использовать отдельно на собственном транспорте.

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

xrtHttpBodyStreamWrite(pStream, XRT_BYTES_LITERAL(
	"data: line1\nline2\n\n"));   /* у line2 нет префикса data: - синтаксическая ошибка */

xhttpsseevent Event = { 0 };
Event.Data = XRT_STR_LITERAL("line1\nline2");   /* упаковщик разобьёт на многострочные data: */
Event.Flags = XHTTP_SSE_EVENT_DATA;
xrtHttpSseSendEvent(pStream, &Event);

while ( xrtHttpSseSend(pStream, Data) == XHTTP_BODY_STREAM_AGAIN ) {
	/* плотный цикл: жжёт CPU и может никогда не протиснуться */
}

xhttpbodystreamresult R;
while ( (R = xrtHttpSseSend(pStream, Data)) ==
		XHTTP_BODY_STREAM_AGAIN ) {
	xfuture* pW = xrtHttpBodyStreamWaitWritable(pStream);
	if ( pW == NULL ) { break; }
	xrtFutureWaitFor(pW, Timeout);   /* ждём, общий Future не отменяем */
	xrtFutureDestroy(pW);
}
/* выход из цикла с R OK/CLOSED - обработать соответственно */

pReply = xrtHttpSseReplyCreate(NULL, &pStream);
respond(pConn, pReply);   /* развёртывание за прокси: события копятся партиями */

pReply = xrtHttpSseReplyCreate(NULL, &pStream);
xrtHttpReplySetHeader(pReply,
	XRT_STR_LITERAL("X-Accel-Buffering"),
	XRT_STR_LITERAL("no"));
xrtHttpReplySetHeader(pReply,
	XRT_STR_LITERAL("Cache-Control"),
	XRT_STR_LITERAL("no-cache"));
respond(pConn, pReply);

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

Итоги