← Cloudflare Workers / workers / runtime-apis

Vzdálené volání procedur (RPC)

Workers poskytují vestavěné, javascriptově nativní RPC (Remote Procedure Call) ↗ systém, který umožňuje:

• Definujte veřejné metody na svém Workeru, které mohou volat jiné Workers ve stejném účtu Cloudflare, a to přes Service Bindings
• Definujte veřejné metody na Durable Objects kterou mohou volat jiní workeři ve stejném účtu Cloudflare, kteří k ní deklarují vazbu.

Systém RPC je navržený tak, aby působil co nejpodobněji volání funkce JavaScript ve stejném Workeru. Ve většině případů můžete psát kód stejným způsobem, jako kdyby vše bylo v jediném Workeru.

Pokud například Worker B implementuje veřejnou metodu add(a, b):

{
	"$schema": "./node_modules/wrangler/config-schema.json",
	"name": "worker_b",
	"main": "./src/workerB.js"
}

"$schema" = "./node_modules/wrangler/config-schema.json"
name = "worker_b"
main = "./src/workerB.js"

import { WorkerEntrypoint } from "cloudflare:workers";

export default class extends WorkerEntrypoint {
	async fetch() {
		return new Response("Hello from Worker B");
	}

	add(a, b) {
		return a + b;
	}
}

import { WorkerEntrypoint } from "cloudflare:workers";

export default class extends WorkerEntrypoint {
	async fetch() {
		return new Response("Hello from Worker B");
	}

	add(a: number, b: number) {
		return a + b;
	}
}

from workers import WorkerEntrypoint, Response

class Default(WorkerEntrypoint):
    async def fetch(self, request):
        return Response("Hello from Worker B")

    def add(self, a: int, b: int) -> int:
        return a + b

Worker A může deklarovat binding k Workeru B:

{
	"$schema": "./node_modules/wrangler/config-schema.json",
	"name": "worker_a",
	"main": "./src/workerA.js",
	"services": [
		{
			"binding": "WORKER_B",
			"service": "worker_b"
		}
	]
}

"$schema" = "./node_modules/wrangler/config-schema.json"
name = "worker_a"
main = "./src/workerA.js"

[[services]]
binding = "WORKER_B"
service = "worker_b"

Jak umožnit, aby Worker A volal add() metodu z Workeru B:

export default {
	async fetch(request, env) {
		const result = await env.WORKER_B.add(1, 2);
		return new Response(result);
	},
};

export default {
	async fetch(request, env) {
		const result = await env.WORKER_B.add(1, 2);
		return new Response(result);
	},
};

from workers import WorkerEntrypoint, Response

class Default(WorkerEntrypoint):
    async def fetch(self, request):
        result = await self.env.WORKER_B.add(1, 2)
        return Response(f"Result: {result}")

Klient, v tomto případě Worker A, zavolá Worker B a řekne mu, aby provedl konkrétní proceduru s konkrétními argumenty, které klient poskytne. Toho se dosahuje pomocí standardních tříd JavaScriptu.

Bez ohledu na to, zda byla metoda, kterou voláte, na serverové straně deklarována jako asynchronní, bude se na klientské straně chovat stejně. Musíte await výsledek.

Mějte na paměti, že volání RPC ve skutečnosti nevracejí Promises, ale vrací typ, který se chová jako Promise. Jde o „custom thenable“, tedy typ implementující metodu then(). JavaScript podporuje await nad libovolným typem thenable, takže návratovou hodnotu můžete ve většině případů považovat za Promise.

(We'll see why the type is not actually a Promise a bit later.)

Téměř všechny typy, které jsou Structured Cloneable ↗ lze použít jako parametr nebo návratovou hodnotu metody RPC. Patří sem i nejzákladnější "hodnotové" typy v JavaScriptu, tedy objekty, pole, řetězce a čísla.

Jako výjimku ze Structured Clone nelze přes RPC předávat třídy definované aplikací (nebo objekty s vlastními prototypy), kromě případů popsaných níže.

Systém RPC podporuje také řadu typů, které nejsou Structured Cloneable, včetně:

• Funkce, které jsou nahrazeny stuby, jež volají zpět odesílateli.
• Třídy definované aplikací, které rozšiřují RpcTarget, které jsou obdobně nahrazeny stuby.
• ReadableStream a WriteableStream, s automatickým řízením toku streamování.
• Požadavek a Odpověď, pro pohodlnou reprezentaci HTTP zpráv.
• samotné RPC stuby, i když byl stub přijat od třetího Workeru.

Funkci můžete odeslat přes RPC. Při odeslání je funkce nahrazena takzvaným „stub“ objektem. Příjemce může tento stub volat jako funkci, ale takové volání vytvoří nové RPC volání zpět na místo, odkud funkce pochází.

Uvažujme následující dva Workery propojené přes Service Binding. Služba counter poskytuje RPC metodu newCounter(), která vrací funkci:

{
	"$schema": "./node_modules/wrangler/config-schema.json",
	"name": "counter-service",
	"main": "./src/counterService.js"
}

"$schema" = "./node_modules/wrangler/config-schema.json"
name = "counter-service"
main = "./src/counterService.js"

import { WorkerEntrypoint } from "cloudflare:workers";

export default class extends WorkerEntrypoint {
	async fetch() {
		return new Response("Hello from counter-service");
	}

	async newCounter() {
		let value = 0;
		return (increment = 0) => {
			value += increment;
			return value;
		};
	}
}

import { WorkerEntrypoint } from "cloudflare:workers";

export default class extends WorkerEntrypoint {
	async fetch() {
		return new Response("Hello from counter-service");
	}

	async newCounter() {
		let value = 0;
		return (increment = 0) => {
			value += increment;
			return value;
		};
	}
}

from workers import WorkerEntrypoint, Response

class Default(WorkerEntrypoint):
    async def fetch(self, request):
        return Response("Hello from counter-service")

    async def new_counter(self):
        value = 0
        def increment(amount=0):
            nonlocal value
            value += amount
            return value
        return increment

Tuto funkci pak může volat klientský Worker:

{
	"$schema": "./node_modules/wrangler/config-schema.json",
	"name": "client_worker",
	"main": "./src/clientWorker.js",
	"services": [
		{
			"binding": "COUNTER_SERVICE",
			"service": "counter-service"
		}
	]
}

"$schema" = "./node_modules/wrangler/config-schema.json"
name = "client_worker"
main = "./src/clientWorker.js"

[[services]]
binding = "COUNTER_SERVICE"
service = "counter-service"

export default {
	async fetch(request, env) {
		using f = await env.COUNTER_SERVICE.newCounter();
		await f(2); // returns 2
		await f(1); // returns 3
		const count = await f(-5); // returns -2

		return new Response(count);
	},
};

export default {
	async fetch(request: Request, env: Env) {
		using f = await env.COUNTER_SERVICE.newCounter();
		await f(2); // returns 2
		await f(1); // returns 3
		const count = await f(-5); // returns -2

		return new Response(count);
	},
};

from workers import WorkerEntrypoint, Response

class Default(WorkerEntrypoint):
    async def fetch(self, request):
        counter = await self.env.COUNTER_SERVICE.new_counter()
        await counter(2)  # returns 2
        await counter(1)  # returns 3
        count = await counter(-5)  # returns -2

        return Response(str(count))

Jak je to možné? Systém neserializuje samotnou funkci. Když funkce vrácená CounterService je volána, běží v rámci CounterService, even if it is called by another Worker.

Volající ve skutečnosti nevolá funkci přímo, ale volá to, čemu se říká „stub“. „Stub“ je Proxy ↗ objekt, který klientovi umožňuje volat vzdálenou službu, jako by byla lokální a běžela ve stejném Workeru. Na pozadí volá zpět Worker, který implementuje CounterService a požádá ho o spuštění funkční closure, která byla vrácena dříve.

V parametrech RPC můžete také odeslat funkci. To umožňuje, aby "server" zavolal zpět "klienta", čímž se obrátí směr vztahu.

Z tohoto důvodu mohou být pojmy „klient“ a „server“ při hovoru o RPC nejednoznačné. „Server“ je Durable Object nebo WorkerEntrypoint a „klient“ je Worker, který server vyvolal prostřednictvím vazby. RPC ale mohou proudit oběma směry. Když mluvíme o jednotlivém volání RPC, doporučujeme místo toho používat pojmy „volající“ (caller) a „volaný“ (callee).

Chcete-li jako parametr nebo návratovou hodnotu metody RPC použít instanci vlastní třídy, musíte rozšířit vestavěnou RpcTarget třídu.

Uvažujme následující příklad:

{
	"$schema": "./node_modules/wrangler/config-schema.json",
	"name": "counter",
	"main": "./src/counter.js"
}

"$schema" = "./node_modules/wrangler/config-schema.json"
name = "counter"
main = "./src/counter.js"

import { WorkerEntrypoint, RpcTarget } from "cloudflare:workers";

class Counter extends RpcTarget {
	#value = 0;

	increment(amount) {
		this.#value += amount;
		return this.#value;
	}

	get value() {
		return this.#value;
	}
}

export class CounterService extends WorkerEntrypoint {
	async newCounter() {
		return new Counter();
	}
}

export default {
	fetch() {
		return new Response("ok");
	},
};

import { WorkerEntrypoint, RpcTarget } from "cloudflare:workers";

class Counter extends RpcTarget {
	#value = 0;

	increment(amount: number) {
		this.#value += amount;
		return this.#value;
	}

	get value() {
		return this.#value;
	}
}

export class CounterService extends WorkerEntrypoint {
	async newCounter() {
		return new Counter();
	}
}

export default {
	fetch() {
		return new Response("ok");
	},
};

Metoda increment lze volat přímo z klienta, stejně jako veřejnou vlastnost value:

{
	"$schema": "./node_modules/wrangler/config-schema.json",
	"name": "client-worker",
	"main": "./src/clientWorker.js",
	"services": [
		{
			"binding": "COUNTER_SERVICE",
			"service": "counter",
			"entrypoint": "CounterService"
		}
	]
}

"$schema" = "./node_modules/wrangler/config-schema.json"
name = "client-worker"
main = "./src/clientWorker.js"

[[services]]
binding = "COUNTER_SERVICE"
service = "counter"
entrypoint = "CounterService"

export default {
	async fetch(request, env) {
		using counter = await env.COUNTER_SERVICE.newCounter();

		await counter.increment(2); // returns 2
		await counter.increment(1); // returns 3
		await counter.increment(-5); // returns -2

		const count = await counter.value; // returns -2

		return new Response(count);
	},
};

export default {
	async fetch(request: Request, env: Env) {
		using counter = await env.COUNTER_SERVICE.newCounter();

		await counter.increment(2); // returns 2
		await counter.increment(1); // returns 3
		await counter.increment(-5); // returns -2

		const count = await counter.value; // returns -2

		return new Response(count);
	},
};

from workers import WorkerEntrypoint, Response

class Default(WorkerEntrypoint):
    async def fetch(self, request):
        counter = await self.env.COUNTER_SERVICE.new_counter()

        await counter.increment(2)  # returns 2
        await counter.increment(1)  # returns 3
        await counter.increment(-5)  # returns -2

        count = await counter.value  # returns -2

        return Response(str(count))

Třídy, které rozšiřují RpcTarget fungují velmi podobně jako funkce: samotný objekt se neserializuje, ale nahradí se stubem. Samotný stub v tomto případě nelze volat, ale jeho metody ano. Volání jakékoli metody na stubu ve skutečnosti provede RPC zpět na původní objekt, kde byl vytvořen.

Jak je vidět výše, lze přistupovat také k vlastnostem tříd. Vlastnosti se chovají jako RPC metody, které neberou žádné argumenty: jejich aktuální hodnotu asynchronně načtete pomocí await. Mějte na paměti, že samotné čekání na vlastnost (které v zákulisí volá .then() na něm) je to, co způsobí načtení vlastnosti. Pokud nepoužijete await při přístupu k vlastnosti se nenačte.

Když zavoláte metodu RPC a jako odpověď obdržíte objekt, je běžné na tomto objektu ihned zavolat další metodu:

// Two round trips.
using counter = await env.COUNTER_SERVICE.getCounter();
await counter.increment();

// Two round trips.
using counter = await env.COUNTER_SERVICE.getCounter();
await counter.increment();

# Two round trips.
counter = await self.env.COUNTER_SERVICE.get_counter()
await counter.increment()

Zvažte však případ, kdy Worker služba, kterou voláte, může být v síti daleko, jako například v případě Smart Placement nebo Durable Objects. Výše uvedený kód provádí dvě zpáteční cesty, první při volání getCounter(), a znovu při volání .increment(). Tomu bychom se rádi vyhnuli.

U většiny RPC systémů by jediným způsobem, jak se tomuto problému vyhnout, bylo sloučit obě volání do jednoho dávkového („batch“) volání, například nazvaného getCounterAndIncrement(). To ale zhoršuje rozhraní. Lokální rozhraní byste takto nenavrhovali.

Workers RPC umožňuje jiný přístup: stačí vynechat první await:

// Only one round trip! Note the missing `await`.
using promiseForCounter = env.COUNTER_SERVICE.getCounter();
await promiseForCounter.increment();

// Only one round trip! Note the missing `await`.
using promiseForCounter = env.COUNTER_SERVICE.getCounter();
await promiseForCounter.increment();

# Only one round trip! Note the missing await.
promise_for_counter = self.env.COUNTER_SERVICE.get_counter()
await promise_for_counter.increment()

V tomto kódu getCounter() vrací promise pro čítač. Obvykle je s promise jediné, co uděláte, await to. Promises RPC ve Workers jsou ale speciální: umožňují také zahájit spekulativní volání na budoucí výsledek promise. Tato volání se na server odesílají okamžitě, aniž by se čekalo na dokončení původního volání. Díky tomu lze více zřetězených volání dokončit v rámci jediného round tripu.

Jak to funguje? Promise vrácený z RPC není skutečný JavaScript Promise. Místo toho jde o vlastní "Thenable" ↗. Má .then() metoda jako Promise, díky čemuž jej lze použít všude tam, kde byste použili běžný Promise. Můžete například await to. Kromě toho se RPC promise chová i jako stub. Zavolání libovolné metody na promise vytvoří spekulativní volání na budoucí výsledek tohoto promise. Tomu se říká "promise pipelining".

Toto funguje i při volání vlastností objektů vrácených metodami RPC. Například:

import { WorkerEntrypoint } from "cloudflare:workers";

export class MyService extends WorkerEntrypoint {
	async foo() {
		return {
			bar: {
				baz: () => "qux",
			},
		};
	}
}

import { WorkerEntrypoint } from "cloudflare:workers";

export class MyService extends WorkerEntrypoint {
	async foo() {
		return {
			bar: {
				baz: () => "qux",
			},
		};
	}
}

from workers import WorkerEntrypoint

class MyService(WorkerEntrypoint):
    async def foo(self):
        return {
            "bar": {
                "baz": lambda: "qux",
            },
        }

export default {
	async fetch(request, env) {
		using foo = env.MY_SERVICE.foo();
		let baz = await foo.bar.baz();
		return new Response(baz);
	},
};

export default {
	async fetch(request, env) {
		using foo = env.MY_SERVICE.foo();
		let baz = await foo.bar.baz();
		return new Response(baz);
	},
};

from workers import WorkerEntrypoint, Response

class Default(WorkerEntrypoint):
    async def fetch(self, request):
        foo = self.env.MY_SERVICE.foo()
        baz = await foo.bar.baz()
        return Response(baz)

Pokud počáteční RPC nakonec vyvolá výjimku, selžou se stejnou výjimkou i všechna zřetězená volání

Můžete odesílat a přijímat ReadableStream, WriteableStream, Request, a Response pomocí metod RPC. Bajty v těle se přitom automaticky streamují s odpovídajícím řízením toku. Díky tomu můžete přes RPC odesílat zprávy větší než obvyklý limit 32 MiB.

Pouze bajtově orientované streamy ↗ (streamy se základním bajtovým zdrojem type: "bytes") jsou podporovány.

Ve všech případech přechází vlastnictví streamu na příjemce. Odesílatel po odeslání streamu už nemůže stream číst ani do něj zapisovat. Pokud si chce odesílatel ponechat vlastní kopii, může použít tee() metoda objektu ReadableStream ↗ nebo clone() metoda objektu Request nebo Response ↗. Mějte na paměti, že to může přinutit systém bufferovat bajty a přijít o výhody řízení toku.

Stub přijatý přes RPC z jednoho Workeru lze přes RPC přeposlat dalšímu Workeru.

using counter = env.COUNTER_SERVICE.getCounter();
await env.ANOTHER_SERVICE.useCounter(counter);

using counter = env.COUNTER_SERVICE.getCounter();
await env.ANOTHER_SERVICE.useCounter(counter);

counter = self.env.COUNTER_SERVICE.get_counter()
await self.env.ANOTHER_SERVICE.use_counter(counter)

Do tohoto procesu jsou zapojeny tři různé workery:

1. Volající Worker (nazveme ho „introducer“)
2. COUNTER_SERVICE
3. ANOTHER_SERVICE

Když ANOTHER_SERVICE volá metodu na counter která je do něj předána, se toto volání automaticky proxuje přes introducer dál do RpcTarget třída implementovaná COUNTER_SERVICE.

Tímto způsobem může introducer Worker propojit dva Workery, které by jinak neměly žádnou možnost mezi sebou navázat přímé spojení.

V současnosti toto proxování trvá pouze do konce prováděcích kontextů Workers. Proxy připojení nelze uchovat pro pozdější použití.

V tomto videu se podíváme, jak Cloudflare Workers podporují Remote Procedure Calls (RPC) a zjednodušují tak komunikaci mezi Workers. Naučíte se implementovat RPC ve svých aplikacích v JavaScriptu a snadno stavět bezserverová řešení. Ať už spravujete mikroslužby, nebo optimalizujete webovou architekturu, tento tutoriál vám ukáže, jak rychle nastavit a používat Cloudflare Workers pro volání RPC. Na konci videa budete rozumět tomu, jak volat funkce mezi Workers, předávat funkce jako argumenty a implementovat autentizaci uživatelů v Cloudflare Workers.

• Životní cyklus
• Vyhrazené metody
• Model viditelnosti a zabezpečení
• TypeScript
• Zpracování chyb

• Smart Placement se při volání RPC v současnosti ignoruje. Pokud je pro Worker A povoleno Smart Placement a Worker B deklaruje Service Binding k němu, když Worker B volá Worker A přes RPC, Worker A poběží lokálně, na stejném stroji.

• Maximální limit serializovaného RPC je 32 MiB. Zvažte použití ReadableStream při vracení více dat.

  export class MyService extends WorkerEntrypoint {
  	async foo() {
  		// Although this works, it puts a lot of memory pressure on the isolate.
  		// If possible, streaming the data from its original source is much preferred and would yield better performance.
  		// If you must buffer the data into memory, consider chunking it into smaller pieces if possible.
  
  		const sizeInBytes = 33 * 1024 * 1024; // 33 MiB
  		const arr = new Uint8Array(sizeInBytes);
  
  		return new ReadableStream({
  			start(controller) {
  				controller.enqueue(arr);
  				controller.close();
  			},
  		});
  	}
  }

  export class MyService extends WorkerEntrypoint {
  	async foo() {
  		// Although this works, it puts a lot of memory pressure on the isolate.
  		// If possible, streaming the data from its original source is much preferred and would yield better performance.
  		// If you must buffer the data into memory, consider chunking it into smaller pieces if possible.
  
  		const sizeInBytes = 33 * 1024 * 1024; // 33 MiB
  		const arr = new Uint8Array(sizeInBytes);
  
  		return new ReadableStream({
  			start(controller) {
  				controller.enqueue(arr);
  				controller.close();
  			},
  		});
  	}
  }