Message bus and RPC

Services talk to each other through Redis Pub/Sub. A channel is an address: every service has a direct channel named after it, plus the shared channels broadcast for distributing events and enrich for enrichment. Messages are JSON-RPC 2.0 with two extra fields, to and from. All of this is implemented in rock-me-ts by the RPCAdapterRedis (transport) and RPCAgnostic (protocol) classes.

Channels#

When creating the adapter, a service passes the list of channels it subscribes to:

const channels = [config.rpc.name, BROADCAST];        // listener: chwriter, kafkastream
const channels = [ENRICH, config.rpc.name];           // enricher: mmgeo

Sending is a PUBLISH to a channel named after the recipient. To write to the chwriter service, all you need is its name: rpc.notify('chwriter', 'method', params).

Message format#

Request:

{
  "jsonrpc": "2.0",
  "id": "1k",
  "from": "front_1",
  "to": "mmgeo_1",
  "method": "mmgeo_1:enrich:front_1",
  "params": { "ip": "93.184.216.34", "ua": "Mozilla/5.0 ..." }
}

Response:

{
  "jsonrpc": "2.0",
  "id": "1k",
  "from": "mmgeo_1",
  "to": "front_1",
  "service_group": "mmgeo",
  "result": { "country_iso": "RU", "city_en": "Moscow" }
}

Details:

• Before sending, method is encoded as to:method:from; the recipient parses the string back. In code you work with the plain method name.
• A request without an id is a notification: the recipient runs the method and sends nothing back.
• If a service has rpc.service_group set, it is added to the response. Front uses it to group the results of several instances of the same enricher.
• An exception in a handler is turned into a response with an error field (code, message, data with the stack), but only for requests that have an id.

API#

register#

Registers a method handler. One method, one function; registering again replaces the previous one:

this.rpc.register(BROADCAST, this.writer.write);
this.rpc.register(METHOD_STATUS, () => this.status());

The handler receives params (or an empty object if there are none) and returns a result. The result is sent to the sender only if the sender is waiting for a response.

notify#

A notification with no response. This is how Front distributes events to listeners, and how services send a heartbeat:

this.rpc.notify(SERVICE_DIRECTOR, METHOD_IAMALIVE, { name: this.name });

request#

A request that waits for a response. Returns a promise:

const status = await this.rpc.request<RPCAppStatus>('mmgeo_1', METHOD_STATUS, {});
const geo = await this.rpc.request('enrich', 'enrich', { ip }, {
  timeout: 300,
  services: ['mmgeo_1', 'uapc_1']
});

Options:

• timeout — milliseconds, default 200. When it elapses, the promise is rejected with a Request timeout error.
• services — the list of service names you need responses from. The request becomes a group request: the promise resolves to an object whose key is the responder's service_group (or its name, if it has no group) and whose value is its result. As soon as everyone listed has answered, the promise completes; on timeout it also resolves, but only with whatever arrived in time. An empty list means the request is not sent and the result is null.

If the result of a single response is an object of the form { type__: 'data', data: {...} } (a response assembled via response.data), the promise resolves to the contents of data.

import { getAppDeps, TheIds } from '@rockstat/rock-me-ts';
getAppDeps().setDep('ids', new TheIds());

Registration in Front#

Front needs to know which services enrich events and which handle keys. For this, a service sends a heartbeat every few seconds, and Front responds by requesting its status:

1. Service: notify('director', '__iamalive', { name }). Front listens on the director channel.
2. Front: request(name, '__status') to the service's direct channel.
3. The service responds with an RPCAppStatus structure containing a register list.
4. Front rebuilds the routing tables. If the heartbeat stops arriving, the registration is removed after a while, so the interval must be shorter than Front's timeout: mmgeo sends its first heartbeat 500 ms after start and then every 9 seconds.

import { RPCAppStatus, MethodRegistration, METHOD_STATUS, METHOD_IAMALIVE, SERVICE_DIRECTOR } from '@rockstat/rock-me-ts';

async status(): Promise<RPCAppStatus> {
  return {
    name: this.name,
    app_state: 'running',
    app_started: Number(this.appStarted),
    app_uptime: Date.now() - Number(this.appStarted),
    register: [{
      method: 'enrich',
      role: 'enricher',
      options: { keys: ['in.gen.track'], props: { ip: 'td.ip', ua: 'td.ua' } }
    }]
  };
}

async setup() {
  this.rpc.register(METHOD_STATUS, () => this.status());
  const aliver = () => this.rpc.notify(SERVICE_DIRECTOR, METHOD_IAMALIVE, { name: this.name });
  setTimeout(aliver, 500);
  setInterval(aliver, 9 * 1000);
}

A register item:

For a handler, Front builds the key in.gen.<alias or service name>.<method> and calls the method with a request carrying the full event. The response is either a response.data, response.redirect, response.pixel, or response.error structure from rock-me-ts, or any object, which Front wraps in response.data. The method * subscribes the handler to all of the service's events.

RPC metrics#

RPCAgnostic itself records the time of every request to the Meter (rpc.request with target and method tags) and an error counter rpc.error. Add your own metrics with meter.tick('kafka.write.success') or meter.timenote('ch.flush'), which returns a function that finishes the measurement.