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.СоздатьПарсер();
Для Каждого Событие Из Парсер.Добавить(ОчереднойФрагмент) Цикл
Обработать(Событие);
КонецЦикла;| Метод | Возвращает | Описание |
|---|---|---|
ОткрытьПоток(Приёмник) |
SseStream |
Поток записи поверх соединения |
СоздатьСобытие(ТипСобытия, Данные, Идентификатор) |
SseEvent |
Новое событие |
Сформировать(ТипСобытия, Данные, Идентификатор) |
Строка |
Текстовое представление события |
СоздатьПарсер(ПоследнийИдентификатор) |
SseStreamParser |
Инкрементальный парсер |
Разобрать(Текст) |
Массив |
Разбор готового текста потока |
ПодключитьсяКПотоку(Адрес, Параметры) |
SseClient |
Подключённый клиент |
СоздатьКлиент(Адрес, Параметры) |
SseClient |
Клиент без подключения |
ЗаголовкиОтвета() |
Соответствие |
Заголовки ответа для потока событий |
ТипСодержимого() |
Строка |
text/event-stream |
ОтправитьЗаголовки(), Отправить(Событие), ОтправитьСобытие(ТипСобытия, Данные, Идентификатор), ОтправитьКомментарий(Текст), Пинг(), УстановитьВремяПереподключения(Миллисекунды), Закрыть(); свойства Активен, ОтправленоСобытий.
Приёмником может быть любой объект с методом ОтправитьДвоичныеДанные или ОтправитьСтроку - этому условию удовлетворяет TCPСоединение, которое winow передаёт в топик серверных событий.
Подключиться(), ПрочитатьСледующее(Переподключаться = Истина), Слушать(Обработчик, МаксимумСобытий = 0, Переподключаться = Истина), ПоследнийИдентификатор(), Закрыть(); свойства Подключен, КоличествоПереподключений.
| Параметр | По умолчанию | Описание |
|---|---|---|
Заголовки |
- | Дополнительные заголовки запроса |
ТаймаутЧтения |
60000 |
Таймаут чтения в миллисекундах |
ПаузаПереподключения |
3000 |
Пауза перед переподключением, мс |
МаксимумПопыток |
0 |
Предел попыток, 0 - без ограничения |
ПоследнийИдентификатор |
"" |
С какого события продолжить |
ТаймаутЧтения должен превышать интервал keep-alive сервера, иначе тишина в потоке будет принята за обрыв.
Свойства ТипСобытия, Данные, Идентификатор, ВремяПереподключения; методы ДанныеКакJson(), Сформировать().
Добавить(Фрагмент) возвращает массив собранных событий, Сбросить() очищает незавершённое состояние, свойства ПоследнийИдентификатор и ВремяПереподключения.
Реализован алгоритм разбора из раздела 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