← Cloudflare Pipelines / pipelines / streams

Запись в потоки

Отправка событий в потоки с помощью Привязки Worker или HTTP-конечные точки для клиентских приложений и внешних систем.

Привязки Worker обеспечивают безопасный способ отправки данных в потоки из Workers без управления API-токенами или учётными данными.

Добавьте в файл Wrangler pipeline binding, который указывает на ваш stream:

{
	"pipelines": [
		{
			"binding": "STREAM",
			"stream": "<STREAM_ID>"
		}
	]
}

[[pipelines]]
binding = "STREAM"
stream = "<STREAM_ID>"

Привязка конвейера предоставляет метод для отправки данных в поток:

Отправляет в поток массив записей, пригодных для сериализации в JSON. Возвращает Promise, который разрешается после подтверждения приёма записей.

export default {
	async fetch(request, env, ctx) {
		const events = await request.json();

		await env.STREAM.send(events);

		return new Response("Events sent");
	},
};

export default {
	async fetch(request, env, ctx): Promise<Response> {
		const events = await request.json<Record<string, unknown>[]>();

		await env.STREAM.send(events);

		return new Response("Events sent");
	},
} satisfies ExportedHandler<Env>;

Если у потока задана схема, выполнение wrangler types генерирует специфичные для схемы типы TypeScript для привязок конвейера. Вместо общего Pipeline<PipelineRecord>, ваши bindings получают именованный тип записи с полным автозаполнением и проверкой типов на этапе компиляции. Подробнее см. в wrangler types документация чтобы узнать больше.

После выполнения wrangler types, сгенерированный worker-configuration.d.ts файл содержит именованный тип записи внутри Cloudflare пространстве имён. Имя типа образуется от имени потока (а не имени привязки), преобразованного в PascalCase с Record суффикс.

Ниже приведён пример того, как выглядят сгенерированные типы в worker-configuration.d.ts для потока с именем ecommerce_stream:

declare namespace Cloudflare {
	type EcommerceStreamRecord = {
		user_id: string;
		event_type: string;
		product_id?: string;
		amount?: number;
	};
	interface Env {
		STREAM: import("cloudflare:pipelines").Pipeline<Cloudflare.EcommerceStreamRecord>;
	}
}

wrangler types возвращается к общему Pipeline<PipelineRecord> тип в следующих сценариях:

• Нет аутентификации: Выполните wrangler login чтобы включить типизированные привязки pipeline.
• Поток не найден: Идентификатор потока в конфигурации Wrangler не соответствует ни одному существующему потоку.
• Неструктурированный поток: Поток был создан без схемы.

Каждый stream предоставляет необязательный HTTP-эндпоинт для приёма данных от внешних приложений, браузеров или любой системы, способной отправлять HTTP-запросы.

HTTP-эндпоинты имеют следующий формат:

https://{stream-id}.ingest.cloudflare.com

URL конечной точки вашего stream можно найти на панели управления Cloudflare в разделе Pipelines > Streams или с помощью Wrangler CLI, указав ID или имя потока:

npx wrangler pipelines streams get <STREAM_NAME_OR_ID>

Отправка событий в виде массивов JSON через запросы POST:

curl -X POST https://{stream-id}.ingest.cloudflare.com \
  -H "Content-Type: application/json" \
  -d '[
    {
      "user_id": "12345",
      "event_type": "purchase",
      "product_id": "widget-001",
      "amount": 29.99
    }
  ]'

Если для вашего потока включена аутентификация, укажите API-токен в Authorization заголовок:

curl -X POST https://{stream-id}.ingest.cloudflare.com \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer YOUR_API_TOKEN" \
  -d '[{"event": "test"}]'

API токен должен иметь Workers Pipeline Send разрешение. Подробнее см. в Создание API-токена документация.

Способ проверки данных в потоке зависит от его конфигурации:

• Структурированные потоки: События должны соответствовать полям и типам, определённым в схеме.
• Неструктурированные потоки: Принимает любую допустимую структуру JSON. Данные хранятся в одном value столбец.

Для структурированных stream убедитесь, что события соответствуют определению схемы. Некорректные события будут приняты, но затем отклонены, поэтому проверяйте данные перед отправкой, чтобы избежать потери событий. При использовании Worker bindings выполните wrangler types чтобы сгенерировать типизированные привязки Pipelines которые выявляют нарушения схемы на этапе компиляции. Вы также можете выполнить запрос к метрики пользовательских ошибок для мониторинга отброшенных событий и диагностики проблем проверки схемы.