sse

Server-Sent Events для OneScript: серверная запись событий и клиент, который ведёт себя как браузерный EventSource.

SSE - самый простой способ доставить изменения с сервера на клиент: одно долгоживущее HTTP-соединение, текстовый формат, автоматическое переподключение и продолжение потока с последнего доставленного события. В отличие от WebSocket не требует смены протокола и проходит через обычные HTTP-прокси.

Установка

opm install sse

Использование

Сервер: отправка событий

#Использовать sse
Поток= SSE.ОткрытьПоток(Соединение);
Поток.ОтправитьСобытие("flag.changed",Новый Структура("key,enabled","new-checkout",Истина),"17");

Структуры и соответствия сериализуются в JSON автоматически. Разрыв соединения клиентом не выбрасывает исключение: методы возвращают Ложь, а поток переходит в неактивное состояние - его достаточно убрать из пула подписчиков.

ЕслиНе Поток.Пинг() Тогда
Подписчики.Удалить(Идентификатор);КонецЕсли;

При работе поверх сырого TCP заголовки отправляет сам поток:

Поток.ОтправитьЗаголовки();
Поток.УстановитьВремяПереподключения(3000);

Если заголовки формирует веб-сервер, возьмите их готовыми:

Для Каждого Заголовок Из SSE.ЗаголовкиОтвета() Цикл
Ответ.Заголовки[Заголовок.Ключ] = Заголовок.Значение;КонецЦикла;

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

Клиент= SSE.ПодключитьсяКПотоку("http://localhost:3333/stream");ПокаИстинаЦиклСобытие= Клиент.ПрочитатьСледующее();Если Событие =НеопределеноТогдаПрервать;КонецЕсли;Если Событие.ТипСобытия ="flag.changed"Тогда
Обновить(Событие.ДанныеКакJson());КонецЕсли;КонецЦикла;

Либо через обработчик:

Клиент.Слушать(Новый Действие(ЭтотОбъект,"ПриСобытии"));

При обрыве клиент переподключается сам и передаёт серверу Last-Event-ID, поэтому поток продолжается ровно с того места, где прервался. Сохранив Клиент.ПоследнийИдентификатор(), можно продолжить и после перезапуска приложения:

Параметры=Новый Структура();
Параметры.Вставить("ПоследнийИдентификатор", СохранённыйИдентификатор);
Параметры.Вставить("Заголовки",Новый Структура("Authorization","Bearer "+ Токен));Клиент= SSE.ПодключитьсяКПотоку(Адрес, Параметры);

Разбор потока без сети

Парсер инкрементальный: принимает произвольные фрагменты и отдаёт только собранные события. Это удобно, когда транспорт свой.

Парсер= SSE.СоздатьПарсер();Для Каждого Событие Из Парсер.Добавить(ОчереднойФрагмент) Цикл
Обработать(Событие);КонецЦикла;

Публичный API

Модуль SSE

Метод	Возвращает	Описание
ОткрытьПоток(Приёмник)	SseStream	Поток записи поверх соединения
СоздатьСобытие(ТипСобытия, Данные, Идентификатор)	SseEvent	Новое событие
Сформировать(ТипСобытия, Данные, Идентификатор)	Строка	Текстовое представление события
СоздатьПарсер(ПоследнийИдентификатор)	SseStreamParser	Инкрементальный парсер
Разобрать(Текст)	Массив	Разбор готового текста потока
ПодключитьсяКПотоку(Адрес, Параметры)	SseClient	Подключённый клиент
СоздатьКлиент(Адрес, Параметры)	SseClient	Клиент без подключения
ЗаголовкиОтвета()	Соответствие	Заголовки ответа для потока событий
ТипСодержимого()	Строка	text/event-stream

Класс SseStream

ОтправитьЗаголовки(), Отправить(Событие), ОтправитьСобытие(ТипСобытия, Данные, Идентификатор), ОтправитьКомментарий(Текст), Пинг(), УстановитьВремяПереподключения(Миллисекунды), Закрыть(); свойства Активен, ОтправленоСобытий.

Приёмником может быть любой объект с методом ОтправитьДвоичныеДанные или ОтправитьСтроку - этому условию удовлетворяет TCPСоединение, которое winow передаёт в топик серверных событий.

Класс SseClient

Подключиться(), ПрочитатьСледующее(Переподключаться = Истина), Слушать(Обработчик, МаксимумСобытий = 0, Переподключаться = Истина), ПоследнийИдентификатор(), Закрыть(); свойства Подключен, КоличествоПереподключений.

Параметр	По умолчанию	Описание
Заголовки	-	Дополнительные заголовки запроса
ТаймаутЧтения	60000	Таймаут чтения в миллисекундах
ПаузаПереподключения	3000	Пауза перед переподключением, мс
МаксимумПопыток	0	Предел попыток, 0 - без ограничения
ПоследнийИдентификатор	""	С какого события продолжить

ТаймаутЧтения должен превышать интервал keep-alive сервера, иначе тишина в потоке будет принята за обрыв.

Класс SseEvent

Свойства ТипСобытия, Данные, Идентификатор, ВремяПереподключения; методы ДанныеКакJson(), Сформировать().

Класс SseStreamParser

Добавить(Фрагмент) возвращает массив собранных событий, Сбросить() очищает незавершённое состояние, свойства ПоследнийИдентификатор и ВремяПереподключения.

Соответствие спецификации

Реализован алгоритм разбора из раздела Interpreting an event stream:

• Разделителями строк считаются CRLF, LF и одиночный CR; разрыв фрагмента между CR и LF не порождает лишнего события.
• Строка, начинающаяся с :, - комментарий; поля с неизвестными именами игнорируются.
• После двоеточия отбрасывается ровно один пробел.
• Строка без двоеточия трактуется как поле с пустым значением.
• Несколько полей data склеиваются через перевод строки; событие без данных не доставляется, но тип события при этом сбрасывается.
• id не сбрасывается между событиями и передаётся при переподключении в Last-Event-ID; значение с нулевым символом игнорируется.
• retry принимается только из ASCII-цифр.
• Метка порядка байтов в начале потока отбрасывается.

При формировании событий каждая строка данных отправляется отдельным полем data, а переводы строк в event и id заменяются пробелом - иначе они разорвали бы событие.

Ограничения

• HTTPS не поддерживается: платформа не даёт TLS поверх TCPСоединение. Для внешнего доступа используйте обратный прокси.
• Клиент построен на TCPСоединение, а не на HTTPСоединение, потому что HTTP-клиент платформы дочитывает ответ целиком - для бесконечного потока это означало бы вечное ожидание.
• Кодирование Transfer-Encoding: chunked разбирается корректно, пока сервер отправляет каждое событие целым фрагментом (так делают все распространённые реализации SSE). Событие, разорванное на границе чанка, может быть прочитано неверно.

Тесты

opm install -l
oneunit execute -d ./tests

Проверка клиента и сервера в связке, включая переподключение с Last-Event-ID:

oscript tools/СерверДляПроверки.os 18340 2 3
oscript tools/КлиентДляПроверки.os 18340 6

Лицензия

MIT