27:null 28:["$","title",null,{"children":"Protobuf Remote Procedure Calls (and gRpc) | Mikhail Kotelnikov | Observable"}] 2d:T5aa,{ const requestDelay = 300; // Timeout between request and response const responseDelay = 300; const serializedRequests = []; const serializedResponses = []; // Method implementation performing real operations: const impl = serviceImplementation.pingPong; // Server stub - it deserializes and call the real method: const serverStub = newPingPongServerAdapter(impl); // Transport layer - in our case it just tracks all requests and responses // and adds some delays between them: const transportLayer = newFakeTransportLayer(serverStub, { // In these arrays the transport layer will add all requests/reponses requests : serializedRequests, responses : serializedResponses, // Transport-related slow down requestDelay, responseDelay }); // Client stub serializes requests and recieves responses const clientStub = newPingPongClientAdapter(transportLayer); // Now we are ready to perform real calls. const requests = ['A', 'B', 'C', 'D', 'E'].map((message, id) => ({ id, message })) ; const responses = []; const result = { lastMessage : '', requests, responses, serializedRequests, serializedResponses }; for await (let response of clientStub(requests)) { result.lastMessage = response.message; // Show the current response and the whole stack of serialized requests and responses: responses.push(response); yield result; } yield result; }2e:T536,{ await restart; const descriptors = methodDescriptors; const serd = jsonSerd; const requests = ['A', 'B', 'C', 'D', 'E'] .map((message, id) => ({ id, message, extraField : `${message} - ${id}` })) ; const service = { test : { EchoService : serviceImplementation } }; const serverStub = rpc.newServerStub(serd, descriptors)(service); const newServerCaller = ({ packageName, serviceName, methodName }) => { const method = serverStub[packageName][serviceName][methodName]; return async function*(request) { for await (let replyObject of method(request)) { // Artificially slow down responses... await Promises.delay(500) yield replyObject; } }; } let serverCaller; serverCaller = newServerCaller; // // We can call the server stub directly, without going through the a network layer // serverCaller = rpc.getServiceMethod(serverStub); const clientStub = rpc.newClientStub(serd, descriptors)(serverCaller); const formatReplies = (list) => { return html`

${JSON.stringify(reply)}

` } const responses = []; for await (let response of clientStub.test.EchoService.pingPong(requests)) { responses.push(response); yield formatReplies(responses); } yield formatReplies(responses); }2f:T644,{ await protobufRestart; const timeout = 500; // Slowdown responses... const ast = parseSchema(serviceSchema); // return yield ast; const descriptors = getMethodsDescriptors(ast); const serd = newPbfSerd(ast); const requests = ['A', 'B', 'C', 'D', 'E'] .map((message, id) => ({ id, message, // This field is dropped when messages are serialized-deserialized extraField : `${message} - ${id}` })); const service = { test : { EchoService : serviceImplementation } }; const serverStub = rpc.newServerStub(serd, descriptors)(service); const newServerCaller = ({ packageName, serviceName, methodName }) => { const method = serverStub[packageName][serviceName][methodName]; return async function*(request) { for await (let replyObject of method(request)) { yield replyObject; // Artificially slow down responses... await Promises.delay(timeout); } }; } let serverCaller; serverCaller = newServerCaller; // // We can call the server stub directly, without going through the a network layer // serverCaller = rpc.getServiceMethod(serverStub); const clientStub = rpc.newClientStub(serd, descriptors)(serverCaller); const formatReplies = (list) => { return html`

${JSON.stringify(reply)}

` } const responses = []; for await (let response of clientStub.test.EchoService.pingPong(requests)) { responses.push(response); yield formatReplies(responses); } yield formatReplies(responses); return descriptors; }30:T52b,{ // Step 1: create a left channel and pannel sending events to it const leftColumn = html`

`; const leftChannel = newChannel(); leftColumn.addEventListener('mousemove', (ev) => leftChannel.emit('update-left', `LEFT: ${ev.x}:${ev.y}`)); // Listen messages from the right: leftChannel.on('update-right', (msg) => appendMessage(leftColumn, msg)); // Step 2: right channel send events from the right pannel const rightColumn = html`

` const rightChannel = newChannel(); rightColumn.addEventListener('mousemove', (ev) => rightChannel.emit('update-right', `RIGHT: ${ev.x}:${ev.y}`)); rightChannel.on('update-left', (msg) => appendMessage(rightColumn, msg)); // Step 3: plug channels between them leftChannel.connect(rightChannel); return (html`

Right column events:

${leftColumn}

Left column events:

${rightColumn}

`) function appendMessage(panel, message) { const list = panel.innerText.split('\n'); list.push(message); if (list.length > 10) list.shift(); panel.innerText = list.join('\n'); } }31:T97b,{ await messagesExchangeRestart; // -------------------------- // Step 0: common part used on the server side and on the client: // Parse the schema const ast = parseSchema(serviceSchema); // Serialization used on the server as well as on the client: // const serd = newPbfSerd(ast); const serd = jsonSerd; // Method description. On the server these descriptors are used to define // how the real code should be called by the generated stub. // The client uses them to automatically generate an object implementing // the API and re-directing real calls to the server. const descriptors = getMethodsDescriptors(ast); const service = { test : { EchoService : serviceImplementation } }; // -------------------------- // Step 1: Implementing the server-side part. // The server stub recieves notifications by the server-side channel, // and re-directs them to a real method implementation. // Data are loaded from the channel by subscribing on specific events // and responses are send back using the emit channel method. const serverChannel = newChannel(); const stopServer = calls.newServer({ channel: serverChannel, service, serd, descriptors, }); // -------------------------- // Step 2: Implementing the client-side part. // This code sends request messages and recieves replies via a dedicated channel. // Create a client-side channel: let idCounter = 0; const newCallId = () => `call-${Date.now()}-${idCounter++}`; const clientChannel = newChannel(); const clientStub = calls.newClient({ channel : clientChannel, newCallId, serd, descriptors }); // -------------------------- // Step 3: Connect client with the server. // Plug input from one channel to the output of the other and vice versa serverChannel.connect(clientChannel); // -------------------------- // Now is everything ready to perform remote calls! const requests = ['Foo', 'Bar', 'Baz', 'Boo', 'Bea'] .map((message, id) => ({ id, message, extraField : `${message} - ${id}` })) ; const formatReplies = (list) => { return html`

${JSON.stringify(reply)}

` } const responses = []; yield responses; for await (let response of clientStub.test.EchoService.pingPong(requests)) { responses.push(response); yield formatReplies(responses); } yield formatReplies(responses); }32:T861,md`## Library Description This notebook provides a set of methods implementing Remote Procedure Calls utilities. The main idea is to to use bi-directional AsyncGenerators to perform all kind of calls and data transformations and provide generic adapters for simple JavaScript functions. The base of all RPC implementations are methods of this type: \`\`\`javascript async function\*(iterator:AsyncIterator):AsyncIterator{ ... } \`\`\` These methods accept just one parameter - an asynchronous iterator providing request objects and returning asynchronously a sequence of resulting objects. This approache allows to automatically generate client/server stubs and implement message marchalling based on: - schema defining methods - simple JavaScript implementation Schema for all service methods should have the following structure: * *packageName* - name of the package containing individual services * *serviceName* - name of the service regrouping multiple methods * *methodName* - name of the callable method * *requestType* - type name of request objects * *requestStream* - this flag defines if the method request is an AsyncIterator or a simple object * *responseType* - type name for response objects * *responseStream* - if this method is an AsyncGenerator or it returns a single object Using such kind of method definitions it is easy to implement universal adapters for real JavaScript methods matching the following constraints: * it accepts just one parameter: a single request object or an AsyncIterator over objects * it returns just one object or it is an AsyncIterator over results Method descriptors allow to transform JavaScript methods to universal format based on the following table:

	requestStream	responseStream
\`async function(Object):Object\`	false	false
\`async function(AsyncIterator):Object\`	true	false
\`async function*(Object):AsyncIterator\`	false	true
\`async function*(AsyncIterator):AsyncIterator\`	true	true

`33:T42f,md`## Stream Transformation These methods create methods transforming provided function calls from one type to another. There are two type of transformations: * Type 1: Tranforming input/output objects for bi-directional generators: * *toSerializedForm({ requestType, responseType, newSerializer, newDeserializer })* serializes requests and deserializes responses * *fromSerializedForm({ requestType, responseType, newSerializer, newDeserializer })* deserializes request objects and serializes responses * Type 2: Transforming simple method calls to bi-directional generators: * *serverCallsAdapter({ requestStream, responseStream })* transform bi-directional iterator function to JavaScript types of calls * *clientCallsAdapter({ requestStream, responseStream })* transforms javascript calls to bi-directional iterator functions An internal utility method used by all implementations defined above: * *newCallsAdapter(getRequest, getResponse)* - transforms all incomming objects with the *getRequest* function and responses with the *getResponse* function `34:T10b8,rpc = { // 4.25kb => 1.45kb // Standard methods are `async function*(it : AsyncIterator)` return { // Main methods providing client/server stubs newServerStub, newClientStub, // Returns individual method adapters to call functionality on the client or on the server newServerMethodProvider, newClientMethodProvider, // Utility methods getJsServiceMethodName, getServiceMethod, } function getJsServiceMethodName(name) { return name[0].toLowerCase() + name.substring(1); } function getServiceMethod(service, getMethodName = (_=>_)) { return ({ packageName, serviceName, methodName }) => { return service[packageName][serviceName][getMethodName(methodName)]; } } function newServerStub(serd, descriptors, getMethodName = getJsServiceMethodName) { return (impl) => buildServiceStub( descriptors, newServerMethodProvider( serd, (typeof impl === 'function') ? (methodInfo) => impl(methodInfo, getMethodName) : getServiceMethod(impl, getMethodName) ) ); } function newClientStub(serd, descriptors, getMethodName = getJsServiceMethodName) { return (getMethod) => buildServiceStub( descriptors, newClientMethodProvider(serd, getMethod), getMethodName ) } function newServerMethodProvider(serd, getImplementation) { return (descriptor) => transform(getImplementation(descriptor), [ // Transform generic AsyncGenerator methods to JavaScript specific calls serverCallsAdapter(descriptor), // We need to parse requests before sending them to the real implementation. // Responses should be serialized before sending them back. fromSerializedForm({...descriptor, ...serd }), ]); } function newClientMethodProvider(serd, getTransport) { return (descriptor) => transform(getTransport(descriptor), [ // Serialize requests before sending them and deserialize responses toSerializedForm({...descriptor, ...serd }), // Transform JavaScript-specific method calls to generic ones clientCallsAdapter(descriptor), ]); } function newCallsAdapter(getRequest, getResponse) { return (method) => { const callMethod = async (value) => method(await value); return (value) => getResponse(callMethod(getRequest(value))); } } function fromIterator(stream) { return stream ? (value) => (async function* () { yield* (await value); })() : (value) => (async function () { for await (let item of await value) { return item; } })() } function toIterator(stream) { return stream ? (value) => (async function* () { yield* (await value); })() : (value) => (async function* () { yield (await value); })(); } function serverCallsAdapter({ requestStream, responseStream }) { return newCallsAdapter(fromIterator(requestStream), toIterator(responseStream)) } function clientCallsAdapter({ requestStream, responseStream }) { return newCallsAdapter(toIterator(requestStream), fromIterator(responseStream)); } function map(f) { return async function* (it) { for await (let item of await it) { yield f(item); } } } function toSerializedForm({ requestType, responseType, newSerializer, newDeserializer }) { return newCallsAdapter(map(newSerializer(requestType)), map(newDeserializer(responseType))); } function fromSerializedForm({ requestType, responseType, newSerializer, newDeserializer }) { return newCallsAdapter(map(newDeserializer(requestType)), map(newSerializer(responseType))); } function transform(method, transforms) { for (let t of transforms) { method = t(method); } return method; } /** * Transforms a sequence of methods with the corresponding information about their signature * to a stub containing the following heirarchies: * package > service > method */ function buildServiceStub(it, getMethod, getMethodName = (_ => _)) { const result = {}; for (let info of it) { const pckg = result[info.packageName] = result[info.packageName] || {}; const service = pckg[info.serviceName] = pckg[info.serviceName] || {}; service[getMethodName(info.methodName)] = getMethod(info); } return result; } }35:Tead,channels = { // 3.7kb => 1.6kb // Returns a method recieving messages from the given channel, transforming them // to a request stream and calling methods. function startServerHandler({ channel, getMethod, messageKey = 'call:message', initKey = 'call:init', }) { const multiplexer = newMultiplexer(channel, messageKey); const handler = async ({ callId, packageName, serviceName, methodName }, resolve) => { const callChannel = multiplexer(callId); try { const method = getMethod({ packageName, serviceName, methodName }); const request = recieveMessages(callChannel, `request`); // resolve(); let it = method(request); await emitMessages(callChannel, `response`, it); } catch (error) { callChannel.emit(`error`, error); } finally { callChannel.close(); } }; channel.on(initKey, handler); return () => channel.off(initKey, handler); } // Returns a method providing client-side part transforming client calls to a sequence of // events on the channel. function newClientMethodsProvider({ channel, newCallId, messageKey = 'call:message', initKey = 'call:init', }) { const multiplexer = newMultiplexer(channel, messageKey); return ({ packageName, serviceName, methodName }) => async function*(request) { const callId = newCallId(); const callChannel = multiplexer(callId); try { channel.emit(initKey, { callId, packageName, serviceName, methodName }); const response = recieveMessages(callChannel, `response`); emitMessages(callChannel, `request`, request); yield* response; } finally { callChannel.close(); } }; } function newEventEmitter() { const index = {}; function on(event, listener) { (index[event] = index[event] || []).push(listener); return () => off(event, listener); } function off(event, listener) { index[event] = (index[event] || []).filter(l => l !== listener); } function emit(event, ...args) { for (let l of (index[event] || [])) { l(...args); } } return { on, off, emit } } function newMultiplexer(channel, messageKey) { const index = {}; const handler = ({ channelId, key, payload }) => { const emitter = index[channelId]; if (!emitter) return ; // TODO: log and throw an error! emitter.emit(key, ...payload); } channel.on(messageKey, handler); return (channelId) => { const emitter = index[channelId] = newEventEmitter(); function close() { delete index[channelId]; } return { on : emitter.on, off : emitter.off, emit(key, ...payload) { channel.emit(messageKey, { channelId, key, payload }); }, close } } } async function emitMessages(channel, prefix, it) { let error; try { for await (let message of it) { channel.emit(`${prefix}:next`, message); } } catch (e) { error = e; } finally { if (error) channel.emit(`${prefix}:error`, error); else channel.emit(`${prefix}:complete`); } } async function* recieveMessages(channel, prefix) { // TODO: Add timeouts/deadlines/throttling yield* agen.iterator(({ next, complete, error }) => { const list = []; list.push(channel.on(`${prefix}:next`, next)); list.push(channel.on(`${prefix}:complete`, complete)); list.push(channel.on(`${prefix}:error`, error)); return () => list.forEach(r => r()); })(); } return { startServerHandler, newClientMethodsProvider, // // Internal methods // newMultiplexer, // emitMessages , // recieveMessages }; }36:[] 2b:{"id":"15e7a8cb4a361293","slug":null,"title":"Protobuf Remote Procedure Calls (and gRpc)","thumbnail":"355bcef9b54d9d6b7f3e8094134d152f23f55c4c0891bfdcfe7c9e7376b42f4b","access_level":"public","create_time":"$D2021-01-18T09:16:23.553Z","update_time":"$D2021-02-10T12:00:10.852Z","view_time":null,"first_public_time":"$D2025-10-10T08:03:18.877Z","has_unpublished_edits":null,"fork":false,"stars":0,"starred":false,"owner":{"type":"team","id":"d4c65cdf77f8f4fd","login":"kotelnikov","name":"Mikhail Kotelnikov","avatar_url":"https://avatars-next.observableusercontent.com/529fa53c5b4acc1414273473d4e049e502c44c1bdbb1a0af81f7faa463dcff4f","tier":"starter_2024"},"version":1102,"publish_version":null,"role":null,"authored":false,"authors":[{"type":"individual","id":"cdbfd308e8310490","login":"kotelnikov","name":"Mikhail Kotelnikov","avatar_url":"https://avatars-next.observableusercontent.com/529fa53c5b4acc1414273473d4e049e502c44c1bdbb1a0af81f7faa463dcff4f","tier":null}],"members":[{"type":"individual","id":"cdbfd308e8310490","login":"kotelnikov","name":"Mikhail Kotelnikov","avatar_url":"https://avatars-next.observableusercontent.com/529fa53c5b4acc1414273473d4e049e502c44c1bdbb1a0af81f7faa463dcff4f","tier":null,"role":"editor"},{"type":"team","id":"d4c65cdf77f8f4fd","login":"kotelnikov","name":"Mikhail Kotelnikov","avatar_url":"https://avatars-next.observableusercontent.com/529fa53c5b4acc1414273473d4e049e502c44c1bdbb1a0af81f7faa463dcff4f","tier":"starter_2024","role":"viewer"}],"comments":[],"custom_thumbnail":false,"trashed":false,"lapsed":null,"subscribed":null,"forks":1,"collection":null,"type":"notebook","body":{"title":"Protobuf Remote Procedure Calls (and gRpc)","cells":[{"id":0,"value":"md`# Protobuf Remote Procedure Calls (and gRpc)`","pinned":false,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","output":"$undefined","database":"$undefined"},{"id":889,"value":"md`*TODO:*\n* message exchange protocol\n - metadata\n - header / trailer messages\n - use with a common schema (messageId, metadata, messageBody : binary)\n - pbf serializer/deserializer\n - how to integrate access control (jwt etc)\n* multiplexer/demultiplexer - transport over one channel\n* backpressure control with locks/unlocks\n* transport on top of the socket.io (adapters)\n* encryption over the stream\n`","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","output":"$undefined","database":"$undefined"},{"id":265,"value":"md`This notebook provides a set of methods implementing Remote Procedure Calls utilities. The main idea is to to use bi-directional AsyncGenerators to perform all kind of calls and data transformations and provide generic adapters for simple JavaScript functions. \n`","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","output":"$undefined","database":"$undefined"},{"id":270,"value":"md`Lets start with a simple service implementation:`","pinned":false,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","output":"$undefined","database":"$undefined"},{"id":273,"value":"serviceImplementation = ({\n \n // This method accepts just one parameter and returns a single response object.\n async sayHello({ name }) { \n return { greeting : `Hello, ${name}` } ;\n },\n \n async* generateMessages({ count = 0 }) {\n for (let i = 0; i < count; i++) {\n yield { idx : i, message : `Ping ${i}` };\n }\n },\n \n // This method accepts a \"stream\" of requests and it respones for each recieved message\n async* pingPong(requests) {\n for await (let req of requests) {\n const { id, message } = req;\n yield { ...req, id, message : `Pong for ${message}` };\n }\n }\n})","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","output":"$undefined","database":"$undefined"},{"id":275,"value":"md`Here are descriptors for service functions. The meaning of individual filds in the Description part below. Note that service method names start with upper case characters to match with the protobuf naming convention.`","pinned":false,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","output":"$undefined","database":"$undefined"},{"id":276,"value":"methodDescriptors = {\n const packageName = 'test';\n const serviceName = 'EchoService';\n const methods = [\n {\n methodName : 'SayHello',\n requestType : 'HelloRequest',\n responseType : 'HelloResponse',\n requestStream : false,\n responseStream : false\n },\n {\n methodName : 'GenerateMessages',\n requestType : 'GenerateMessageRequest', \n responseType : 'GenerateMessageResponse',\n requestStream : false,\n responseStream : true\n },\n {\n methodName : 'PingPong',\n requestType : 'PingRequest',\n responseType : 'PongResponse',\n requestStream : true,\n responseStream : true\n }\n ];\n return methods.map(m => ({...m, packageName, serviceName }))\n}","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","output":"$undefined","database":"$undefined"},{"id":285,"value":"md`For test purposes we are using a simple JSON serializer/deserializer transforming requests and responses to/from strings. It does not take into account object types, it just strigifies and parses them.`","pinned":false,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","output":"$undefined","database":"$undefined"},{"id":287,"value":"jsonSerd = newJsonSerd()","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","output":"$undefined","database":"$undefined"},{"id":350,"value":"md`### Implementing One Client-Server Call\n\nIn this section we will implement a simple client-server type of calls directly in the browser. \nClient-side part accepts request objects, serializes them and sends to the server. The server adapter executes the real code defined in the \\`serviceImplementation\\` service, serializes responses and sends them back. \n`\n","pinned":false,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","output":"$undefined","database":"$undefined"},{"id":314,"value":"\nmd`\nThe full request/response roundtrip:\n\\`\\`\\`\n-> [clientStub] --serializing--> [transport] -deserializing-> [serverStub] --+\n |\n [method implementation]\n |\n<- [clientStub] <-deserializing- [transport] <--serializing-- [serverStub] <-+\n \n\\`\\`\\`\n`","pinned":false,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","output":"$undefined","database":"$undefined"},{"id":321,"value":"pingPongMethodDescriptor = methodDescriptors.find(d => d.methodName === 'PingPong');","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","output":"$undefined","database":"$undefined"},{"id":281,"value":"md`The method below creates a server stub allowing to parse incoming messages, call a service method and return serialized results:`","pinned":false,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","output":"$undefined","database":"$undefined"},{"id":283,"value":"function newPingPongServerAdapter(implementation) {\n return rpc.newServerMethodProvider(jsonSerd, () => implementation)(pingPongMethodDescriptor);\n}","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","output":"$undefined","database":"$undefined"},{"id":308,"value":"md`Client adapters perform inversed operations - serializing requests and deserializing responses. This method creates a such kind of adapters:`","pinned":false,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","output":"$undefined","database":"$undefined"},{"id":312,"value":"function newPingPongClientAdapter(transportLayer) {\n return rpc.newClientMethodProvider(jsonSerd, () => transportLayer)(pingPongMethodDescriptor);\n}","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","output":"$undefined","database":"$undefined"},{"id":400,"value":"md`A simple transport stub transferring all client calls directly to a server stub. It alsow allows to track all request/response messages: `","pinned":false,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","output":"$undefined","database":"$undefined"},{"id":403,"value":"function newFakeTransportLayer(serverStub, {\n requestDelay = 0, // Timeout between request and response\n responseDelay = 0,\n requests = [], // All requests are added here\n responses = [] // All responses go here\n} = {}) {\n return async function*(input) {\n const output = serverStub((async function*() {\n for await (let request of input) {\n requests.push(request);\n requestDelay && await Promises.delay(requestDelay);\n yield request;\n }\n })());\n for await (let response of output) {\n responses.push(response);\n responseDelay && await Promises.delay(responseDelay);\n yield response;\n } \n }\n}","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","output":"$undefined","database":"$undefined"},{"id":356,"value":"md`We have all required blocks to perform a client-server call. Let's put them together and perform a simple \"remote\" invocation:`","pinned":false,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","output":"$undefined","database":"$undefined"},{"id":358,"value":"$2d","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","output":"$undefined","database":"$undefined"},{"id":426,"value":"md`## Full Client-Server Stubs Implementation`","pinned":false,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","output":"$undefined","database":"$undefined"},{"id":687,"value":"viewof restart=html``","pinned":false,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","output":"$undefined","database":"$undefined"},{"id":435,"value":"$2e","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","output":"$undefined","database":"$undefined"},{"id":779,"value":"md`## Protobuf-based Client-Server Stubs`","pinned":false,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","output":"$undefined","database":"$undefined"},{"id":896,"value":"md`In this section we are implementing a client-server roundtrip using Protobuf serialization of messages. To do so we need to define a formal schema for messages and service calls.`","pinned":false,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","output":"$undefined","database":"$undefined"},{"id":784,"value":"md`Let's write down a protobuf schema for the service defined above:`","pinned":false,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","output":"$undefined","database":"$undefined"},{"id":782,"value":"serviceSchema = `\n// See here: https://github.com/grpc/grpc-web/blob/0.4.0/net/grpc/gateway/examples/echo/echo.proto\n\nsyntax = \"proto3\";\n\npackage test;\n\nmessage Hello {\n message Request { string name = 1; }\n message Response { string greeting = 1; }\n}\n\nmessage GenerateMessage {\n message Request { int32 count = 1; }\n message Response { int32 idx = 1; string message = 2; }\n}\n\nmessage PingPong {\n message Request { int32 id = 1; string message = 2; }\n message Response { int32 id = 1; string message = 2; }\n}\n\n// A simple echo service.\nservice EchoService {\n\n rpc SayHello(Hello.Request) returns (Hello.Response);\n\n rpc GenerateMessages(GenerateMessage.Request) returns (stream GenerateMessage.Response);\n\n rpc PingPong(stream PingPong.Request) returns (stream PingPong.Response);\n\n}\n\n`","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","output":"$undefined","database":"$undefined"},{"id":870,"value":"serviceSchemaBackup = `\n// See here: https://github.com/grpc/grpc-web/blob/0.4.0/net/grpc/gateway/examples/echo/echo.proto\n\nsyntax = \"proto3\";\n\npackage test;\n\nmessage HelloRequest { string name = 1; }\nmessage HelloResponse { string greeting = 1; }\n\nmessage GenerateMessageRequest { int32 count = 1; }\nmessage GenerateMessageResponse { int32 idx = 1; string message = 2; }\n\nmessage PingRequest { int32 id = 1; string message = 2; }\nmessage PongResponse { int32 id = 1; string message = 2; }\n\n// A simple echo service.\nservice EchoService {\n\n rpc SayHello(HelloRequest) returns (HelloResponse);\n\n rpc GenerateMessages(GenerateMessageRequest) returns (stream GenerateMessageResponse);\n\n rpc PingPong(stream PingRequest) returns (stream PongResponse);\n\n}\n\n`","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","output":"$undefined","database":"$undefined"},{"id":828,"value":"viewof protobufRestart = html``","pinned":false,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","output":"$undefined","database":"$undefined"},{"id":787,"value":"$2f","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","output":"$undefined","database":"$undefined"},{"id":903,"value":"md`## Message-based request/response service`","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","output":"$undefined","database":"$undefined"},{"id":921,"value":"md`To simulate better classical client-server communication with node-like streams or sockets.io let's implement a basic implementation on top of a channels allowing to emit to and recieve messages from their peers.`","pinned":false,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","output":"$undefined","database":"$undefined"},{"id":937,"value":"md`Basic event emitter:`","pinned":false,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","output":"$undefined","database":"$undefined"},{"id":922,"value":"function newEventEmitter() {\n const index = {};\n function on(event, listener) {\n (index[event] = index[event] || []).push(listener);\n return () => off(event, listener);\n }\n function off(event, listener) {\n index[event] = (index[event] || []).filter(l => l !== listener);\n }\n function emit(event, ...args) {\n // additional 'broadcasting' notification allowing to catch all events\n for (let l of (index['*'] || [])) { l(event, ...args); }\n for (let l of (index[event] || [])) { l(...args); }\n }\n return { on, off, emit }\n}","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","output":"$undefined","database":"$undefined"},{"id":934,"value":"md`Here we implement a \"channel\" with the following main methods:\n- 'emit' - send data to the peer\n- 'on'/'off' - registers/removes listeners recieving data from peers\n\nThere is also an additional method - 'connect' - allowing to recieve and forward all peer chanel messages.\n`","pinned":false,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","output":"$undefined","database":"$undefined"},{"id":930,"value":"function newChannel() {\n const reader = newEventEmitter();\n const writer = newEventEmitter();\n return {\n _reader : reader,\n _writer : writer,\n on : reader.on,\n off : reader.off,\n emit : writer.emit,\n connect : (channel) => {\n const list = [];\n list.push(channel._writer.on('*', reader.emit));\n list.push(writer.on('*', channel._reader.emit));\n return () => list.forEach(r => r()); // Unregister connection\n }\n }\n}","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","output":"$undefined","database":"$undefined"},{"id":953,"value":"md`The example below allows to test communication between two panels using our \"channels\". Mouse movement on top of the left column send messages to the right and right column notifies the left one.`","pinned":false,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","output":"$undefined","database":"$undefined"},{"id":957,"value":"$30","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","output":"$undefined","database":"$undefined"},{"id":975,"value":"md`On top of these \"channels\" implementations it is possible to create client-server communication very similar with APIs like https://socket.io:`","pinned":false,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","output":"$undefined","database":"$undefined"},{"id":980,"value":"agen = require('@agen/utils@^0.8', '@agen/encoding')","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","output":"$undefined","database":"$undefined"},{"id":907,"value":"viewof messagesExchangeRestart=html``","pinned":false,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","output":"$undefined","database":"$undefined"},{"id":905,"value":"$31","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","output":"$undefined","database":"$undefined"},{"id":230,"value":"$32","pinned":false,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","output":"$undefined","database":"$undefined"},{"id":138,"value":"$33","pinned":false,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","output":"$undefined","database":"$undefined"},{"id":147,"value":"md`## Utility methods:\n\n* *transform(iterable)* transform a list of method providers to a single method provider\n* *buildServiceStub(iterable)* recieve an iteratable with method descriptions and generate a stub object with the corresponding methods. Each \n\n`","pinned":false,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","output":"$undefined","database":"$undefined"},{"id":88,"value":"$34","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","output":"$undefined","database":"$undefined"},{"id":755,"value":"function newJsonSerd() {\n return ({ \n newSerializer(name) { return (obj) => JSON.stringify(obj); },\n newDeserializer(name) { return (str) => JSON.parse(str); }\n })\n}","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","output":"$undefined","database":"$undefined"},{"id":1061,"value":"calls = ({ // 0.5kb => 0.3kb\n newClient : function newClient({ serd, descriptors, channel, newCallId }) {\n const newClientMethod = channels.newClientMethodsProvider({\n channel,\n newCallId\n });\n return rpc.newClientStub(serd, descriptors)(newClientMethod);\n },\n newServer : function newServer({ serd, descriptors, channel, service }) {\n const serverStub = rpc.newServerStub(serd, descriptors)(service);\n const getServerMethod = ({ packageName, serviceName, methodName }) => serverStub[packageName][serviceName][methodName];\n return channels.startServerHandler({ channel, getMethod : getServerMethod });\n }\n}) ","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","output":"$undefined","database":"$undefined"},{"id":1039,"value":"$35","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","output":"$undefined","database":"$undefined"},{"id":86,"value":"md`## Schema-Related Methods`","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","output":"$undefined","database":"$undefined"},{"id":764,"value":"function parseProtobufSchema(schema) { // 0.2kb => 0.15kb\n const ast = parseSchema(schema);\n const descriptors = getMethodsDescriptors(ast);\n const serd = newPbfSerd(ast);\n return { descriptors, serd, ast };\n}","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","output":"$undefined","database":"$undefined"},{"id":767,"value":"function newPbfSerd(ast) { // 0.7kb => 0.4kb\n function get(obj, path) {\n path = Array.isArray(path) ? path : path.split('.');\n let o, idx;\n for (o = obj, idx = 0; o && idx < path.length; idx++) {\n o = o[path[idx]];\n }\n return (idx === path.length) ? o : undefined;\n } \n const serializer = compile(ast);\n function newSerializer(type) {\n const write = get(serializer, type).write;\n return (req) => {\n const pbf = new Pbf();\n write(req, pbf);\n return pbf.finish();\n }\n }\n function newDeserializer(type) {\n const read = get(serializer, type).read;\n return (req) => {\n const pbf = new Pbf(req);\n return read(pbf);\n }\n }\n return { newSerializer, newDeserializer };\n}","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","output":"$undefined","database":"$undefined"},{"id":770,"value":"function getMethodsDescriptors(ast) { // 0.18kb => 0.1kb\n const descriptors = [];\n for (let descriptor of visitAstMethods(ast)) {\n descriptors.push(descriptor);\n }\n return descriptors;\n}","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","output":"$undefined","database":"$undefined"},{"id":1077,"value":"`\n4.25kb => 1.45kb\n0.5kb => 0.3kb\n3.7kb => 1.6kb\n0.2kb => 0.15kb\n0.7kb => 0.4kb\n0.18kb => 0.1kb\nO.66kb => 0.3kb\n\nsum = ${Math.round(4.25 + 0.5 + 3.7 + 0.2 + 0.7 + 0.18 + 0.66).toFixed(2)} kb\n`","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","output":"$undefined","database":"$undefined"},{"id":761,"value":"function* visitAstMethods(ast) { // O.66kb => 0.3kb\n const packageName = ast.package;\n for (let service of ast.services) {\n const serviceName = service.name;\n for (let method of service.methods) {\n const methodName = method.name;\n const requestStream = method.client_streaming;\n const responseStream = method.server_streaming;\n const requestType = method.input_type;\n const responseType = method.output_type;\n const methodInfo = {\n packageName, serviceName, methodName,\n requestStream, responseStream, \n requestType, responseType,\n // service, method, ast\n }\n yield methodInfo;\n } \n }\n}","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","output":"$undefined","database":"$undefined"},{"id":760,"value":"function parseSchema(str) {\n return schema.parse(str);\n}\n","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","output":"$undefined","database":"$undefined"},{"id":900,"value":"md`## External Dependencies`","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","output":"$undefined","database":"$undefined"},{"id":35,"value":"Pbf = loadModule('pbf')","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","output":"$undefined","database":"$undefined"},{"id":92,"value":"// schema=require('https://bundle.run/protocol-buffers-schema/index.js')\nschema = loadModule('protocol-buffers-schema', {\n './parse' : './parse.js',\n './tokenize' : './tokenize.js',\n './stringify' : './stringify.js'\n})","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","output":"$undefined","database":"$undefined"},{"id":37,"value":"compile = loadModule('pbf/compile', { './package.json' : { version : '0' } }) ","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","output":"$undefined","database":"$undefined"},{"id":12,"value":"import { loadModule, loadDependencies } from '@kotelnikov/common-modules-loader'","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","output":"$undefined","database":"$undefined"},{"id":78,"value":"md`## Sandbox`","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","output":"$undefined","database":"$undefined"},{"id":845,"value":"{\n const str = `\n message Metadata {\n map values = 1;\n }\n\n message SearchData { // Level 0\n message Request { // Level 1\n message Inner { // Level 2\n required int64 ival = 1;\n optional bool booly = 2;\n }\n // repeated Inner inr = 1;\n }\n message Response { // Level 1\n message Inner { // Level 2\n required int32 ival = 1;\n optional bool booly = 2;\n }\n }\n }\n\n message DataSet { \n message Request {\n message Identity {\n string projectId = 1;\n string dataSetId = 2;\n }\n message DataBlock {\n bytes buf = 1;\n } \n oneof payload {\n Identity identity = 1;\n DataBlock data = 2;\n }\n }\n }\n\n\n service SearchService {\n rpc Search(SearchData.Request.Inner) returns (stream SearchData.Response.Inner);\n }\n`\n const ast = parseSchema(str);\n // return ast;\n \n const serializer = compile.raw(ast);\n \n return serializer;\n \n return { ast, serializer }\n}","pinned":true,"hidden":false,"mode":"ojs","author":"$undefined","summary":"$undefined","output":"$undefined","database":"$undefined"}],"files":"$W36","stdlib":"1","titling":"auto"}} 2c:"$2b:body"