← Cloudflare Durable Objects / durable-objects / examples

Použití Alarms API

Tento příklad implementuje alarm() handler, který umožňuje dávkové zpracování požadavků do jednoho Durable Object.

Pokud je přijat požadavek a není nastaven žádný alarm, nastaví se alarm na 10 sekund do budoucna. Obslužná metoda alarm() handler zpracuje všechny požadavky přijaté v tomto 10sekundovém okně.

Pokud nepřijdou žádné nové požadavky, žádné další alarmy se nenastaví, dokud nedorazí další požadavek.

import { DurableObject } from "cloudflare:workers";

// Worker
export default {
	async fetch(request, env) {
		return await env.BATCHER.getByName("foo").fetch(request);
	},
};

// Durable Object
export class Batcher extends DurableObject {
	constructor(ctx, env) {
		super(ctx, env);
		this.storage = ctx.storage;
		this.ctx.blockConcurrencyWhile(async () => {
			let vals = await this.storage.list({ reverse: true, limit: 1 });
			this.count = vals.size == 0 ? 0 : parseInt(vals.keys().next().value);
		});
	}

	async fetch(request) {
		this.count++;

		// If there is no alarm currently set, set one for 10 seconds from now
		// Any further POSTs in the next 10 seconds will be part of this batch.
		let currentAlarm = await this.storage.getAlarm();
		if (currentAlarm == null) {
			this.storage.setAlarm(Date.now() + 1000 * 10);
		}

		// Add the request to the batch.
		await this.storage.put(this.count, await request.text());
		return new Response(JSON.stringify({ queued: this.count }), {
			headers: {
				"content-type": "application/json;charset=UTF-8",
			},
		});
	}

	async alarm() {
		let vals = await this.storage.list();
		await fetch("http://example.com/some-upstream-service", {
			method: "POST",
			body: Array.from(vals.values()),
		});
		await this.storage.deleteAll();
		this.count = 0;
	}
}

from workers import DurableObject, Response, WorkerEntrypoint, fetch
import time

# Worker
class Default(WorkerEntrypoint):
	async def fetch(self, request):
		stub = self.env.BATCHER.getByName("foo")
		return await stub.fetch(request)

# Durable Object
class Batcher(DurableObject):
	def __init__(self, ctx, env):
		super().__init__(ctx, env)
		self.storage = ctx.storage

		@self.ctx.blockConcurrencyWhile
		async def initialize():
			vals = await self.storage.list(reverse=True, limit=1)
			self.count = 0
			if len(vals) > 0:
			    self.count = int(vals.keys().next().value)

	async def fetch(self, request):
		self.count += 1

		# If there is no alarm currently set, set one for 10 seconds from now
		# Any further POSTs in the next 10 seconds will be part of this batch.
		current_alarm = await self.storage.getAlarm()
		if current_alarm is None:
			self.storage.setAlarm(int(time.time() * 1000) + 1000 * 10)

		# Add the request to the batch.
		await self.storage.put(self.count, await request.text())
		return Response.json(
			{"queued": self.count}
		)

	async def alarm(self):
		vals = await self.storage.list()
		await fetch(
			"http://example.com/some-upstream-service",
			method="POST",
			body=list(vals.values())
		)
		await self.storage.deleteAll()
		self.count = 0

alarm() handler se zavolá jednou za 10 sekund. Pokud neočekávaná chyba ukončí Durable Object, alarm() handler se znovu vytvoří na jiném stroji. Po krátké prodlevě se alarm() handler spustí od začátku na jiném stroji.

Nakonec nakonfigurujte soubor Wrangler tak, aby obsahoval Durable Object binding a migrace na základě dříve zvoleného namespace a názvu třídy.

{
	"$schema": "./node_modules/wrangler/config-schema.json",
	"name": "durable-object-alarm",
	"main": "src/index.ts",
	"durable_objects": {
		"bindings": [
			{
				"name": "BATCHER",
				"class_name": "Batcher"
			}
		]
	},
	"migrations": [
		{
			"tag": "v1",
			"new_sqlite_classes": [
				"Batcher"
			]
		}
	]
}

"$schema" = "./node_modules/wrangler/config-schema.json"
name = "durable-object-alarm"
main = "src/index.ts"

[[durable_objects.bindings]]
name = "BATCHER"
class_name = "Batcher"

[[migrations]]
tag = "v1"
new_sqlite_classes = [ "Batcher" ]