← Cloudflare Pipelines / pipelines / streams

Zápis do streamů

Odesílejte události do streamů pomocí Worker bindings nebo HTTP koncové body pro klientské aplikace a externí systémy.

Worker bindings poskytují bezpečný způsob odesílání dat do streamů z Workers bez nutnosti spravovat API tokeny nebo přihlašovací údaje.

Do souboru Wrangler přidejte pipeline binding, který odkazuje na váš stream:

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

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

Pipeline binding zpřístupňuje metodu pro odesílání dat do vašeho streamu:

Odešle do streamu pole záznamů serializovatelných do JSON. Vrátí Promise, který se vyřeší ve chvíli, kdy je přijetí záznamů potvrzeno.

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>;

Pokud má stream definované schéma, spuštění wrangler types generuje typy TypeScript specifické pro schéma pro vaše vazby pipeline. Namísto obecného Pipeline<PipelineRecord>, vaše bindings získají pojmenovaný typ záznamu s plnou podporou automatického doplňování a kontrolou typů při kompilaci. Podrobnosti najdete v wrangler types dokumentace a zjistěte více.

Po spuštění wrangler types, vygenerovaný worker-configuration.d.ts soubor obsahuje pojmenovaný typ záznamu uvnitř Cloudflare jmenném prostoru. Název typu je odvozen od názvu streamu (nikoli od názvu vazby) a převeden na PascalCase s Record příponu.

Níže je příklad, jak vypadají vygenerované typy v worker-configuration.d.ts pro stream s názvem 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 se vrátí k obecnému Pipeline<PipelineRecord> typ v následujících scénářích:

• Neověřeno: Spusťte wrangler login pro povolení typovaných pipeline bindingů.
• Stream nebyl nalezen: ID streamu v konfiguraci Wrangler neodpovídá žádnému existujícímu streamu.
• Nestrukturovaný stream: Stream byl vytvořen bez schématu.

Každý stream poskytuje volitelný HTTP endpoint pro příjem dat z externích aplikací, prohlížečů nebo libovolného systému, který umí odesílat HTTP požadavky.

HTTP koncové body mají následující formát:

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

Koncový bod URL vašeho streamu najdete v Cloudflare dashboardu v části Pipelines > Streams nebo pomocí Wrangler CLI se zadáním ID streamu nebo názvu streamu:

npx wrangler pipelines streams get <STREAM_NAME_OR_ID>

Odesílejte události jako pole JSON pomocí požadavků 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
    }
  ]'

Pokud je pro váš stream povoleno ověřování, zahrňte API token do Authorization hlavičku:

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

API token musí mít Workers Pipeline Send oprávnění. Další informace najdete v Vytvoření API tokenu dokumentace.

Streams zpracovávají validaci odlišně v závislosti na své konfiguraci:

• Strukturované streamy: Události musí odpovídat definovaným polím a typům schématu.
• Nestrukturované streamy: Přijímá libovolnou platnou strukturu JSON. Data se ukládají do jednoho value sloupec.

U strukturovaných streamů dbejte na to, aby vaše události odpovídaly definici schématu. Neplatné události budou přijaty, ale následně zahozeny, proto si data před odesláním ověřte, abyste zahazování událostí předešli. Při použití Worker bindings spusťte wrangler types k vygenerování typované vazby pipeline které zachytí porušení schématu už při kompilaci. Můžete také dotazovat metriky uživatelských chyb ke sledování zahozených událostí a diagnostice problémů s validací schématu.