Fan-out/Fan-in

Overview

Fan-out/fan-in lets services invoke an arbitrary number of other services providing a set of callback services to be executed once all of the target services complete.

Optionally, additional callbacks can be invoked each time a service from the original target list finished execution.

Target services run in parallel, asynchronously, and are not notified of each other's execution progress. Any callbacks configured also run asynchronously without blocking each other.

The diagram below depicts the feature in its full scope. A source service fans out to two target services, each with a list of its completion callbacks. When both of the target services finish execution, the flow of processing fans in to a final list of callbacks. Thus, the final callbacks can be certain that all the targets have all been already invoked.

Usage examples and API

Common usage

In the most common case, the usage will be as follows. A dictionary mapping targets and their requests along with a list of callbacks is provided to self.patterns.fanout.invoke. All of the targets are invoked in background and, once they all complete, each of the callbacks is invoked.

from zato.server.service import Service

class MyService(Service):

    def handle(self):

        # A dictionary of services to invoke along with requests they receive
        targets = {
            'service1': {'hello':'from-fan-out1'},
            'service2': {'hello':'from-fan-out2'},
        }

        # Callbacks to invoke when both services above finish
        callbacks = ['my.callback1', 'my.callback2']

        # Call's Correlation ID is returned on output
        cid = self.patterns.fanout.invoke(targets, callbacks)

On-target-completion callbacks

Here additional callbacks are executed after each of the targets completes. This is in addition to the final callbacks that are executed regardless of whether target callbacks are used or not.

Note that target callbacks are always the same for each target, there are no means to specify distinct target callbacks for targets individually. However, callbacks can tell target services apart from the metadata they receive on input which contains the target's name.

from zato.server.service import Service

class MyService(Service):

    def handle(self):

        # A dictionary of services to invoke along with requests they receive
        targets = {
            'service1': {'hello':'from-fan-out1'},
            'service2': {'hello':'from-fan-out2'},
        }

        # Callbacks to invoke each time one of the services above completes
        target_callbacks = ['my.target.callback1', 'my.target.callback2']

        # Callbacks to invoke when both services above finish
        final_callbacks = ['my.callback1', 'my.callback2']

        # Call's Correlation ID is returned on output
        cid = self.patterns.fanout.invoke(targets, final_callbacks, target_callbacks)

User-defined Correlation IDs

Users can provide their own Correlation IDs to invoke the services with. It is user's responsibility to ensure the IDs are sufficiently unique and there will be no duplicates, should that happen: results are undefined.

from zato.server.service import Service

class MyService(Service):

    def handle(self):

        # A dictionary of services to invoke along with requests they receive
        targets = {
            'service1': {'hello':'from-fan-out1'},
            'service2': {'hello':'from-fan-out2'},
        }

        # Callbacks to invoke when both services above finish
        callbacks = ['my.callback1', 'my.callback2']

        # User-provided Correlation ID
        cid = '2963-2645-9715-1719'

        # CID returned is the one received on input
        cid = self.patterns.fanout.invoke(targets, callbacks, cid=cid)

Authoring callbacks

A callback is a regular Zato service that is invoked in specific moments of a fan-out/fan-in call. The same service can serve as a final and target callback. Each callback is invoked asynchronously, in isolation from any other callbacks defined.

Upon execution, each callback's self.channel attribute will be set to either FANOUT_ON_FINAL or FANOUT_ON_TARGET.

On input, a final callback's self.request.payload attribute will be a Python dictionary with the contents in the format as below:

  {
   'source': 'fanout1.my-service',
   'req_ts_utc': '2025-01-19 19:55:46',
   'on_target': '',
   'on_final': ['my.cb1', 'zato.helpers.input-logger'],
   'data':
    {'my.service1': {
      'source': 'fanout1.my-service',
      'target': 'my.service1',
      'response': 'My response',
      'req_ts_utc': '2025-01-19 19:55:45',
      'resp_ts_utc': '2025-01-19 19:55:46',
      'ok': True,
      'exception': None,
      'cid': '63e7c3a2ba616badecb125ce'},
    'my.service2': {
      'source': 'fanout1.my-service',
      'target': 'my.service2',
      'response': 'My response2',
      'req_ts_utc': '2025-01-19 19:55:45',
      'resp_ts_utc': '2025-01-19 19:55:46',
      'ok': True,
      'exception': None,
      'cid': '9ffc1e5d79f6616b2dfb2c35'}}}

The 'data' dictionary will have the following keys:

• source: name of the original service issuing the fan-out/fan-in call (repeated)
• target: name of a service, one of the targets (repeated)
• response: response the service produced
• req_ts_utc: request timestamp in UTC (repeated)
• resp_ts_utc: when was the response created, in UTC
• ok: True/False depending on whether the call was successful (raised no exceptions) or not
• exception: exception's traceback as string, if ok is not True
• cid: Correlation ID, the same one that the source service was given on output from self.patterns.fanout.invoke

On-target callbacks receive self.request.payload equal to an individual entry in the 'data' key above:

  {
    'source': 'fanout1.my-service',
    'target': 'my.service1',
    'response': 'My response',
    'req_ts_utc': '2025-01-19 19:55:45',
    'resp_ts_utc': '2025-01-19 19:55:46',
    'ok': True,
    'exception': None,
    'cid': 'c6722658d6520777849dee0'
  }

Similarities to Parallel Execution

The feature is very similar to the Parallel Execution pattern: the major difference being only that in Fan-out/Fan-in it is the final callbacks that are required while on-target ones are optional while in Parallel Execution there are no final callbacks while on-target are required.

Integration patterns

• Publish/subscribe
• Invoke/retry
• Fan-out/Fan-in
• Parallel execution
• Async invoke with a callback