Module kflow_gen_map

Description
Data Types

This module implements map-like stream processing node that applies a pure function to each incoming message.

This module defines the kflow_gen_map behaviour.
Required callback functions: map/3.
Optional callback functions: init/1, terminate/1.

Description

This module implements map-like stream processing node that applies a pure function to each incoming message.

This behavior can be used in two modes: full and simplified. In simplified mode stream processing node is defined like following:

      {map, fun(Offset, Message) -> Message end}

In full mode one has to create a callback module with kflow_gen_map behavior.

map callback takes 3 arguments: first is offset of a message, second is the message itself and the third one is state of the callback module. This state is created in init callback and remains the same through the lifetime of the pipe. Return value of map callback is a processed message that is passed downstream.

init and terminate callbacks can be used e.g. when some resource should be obtained to process messages. Both callbacks are optional; configuration will be passed as is to map callback when init is omitted.

Example

   -module(my_map).
  
   -behavior(kflow_gen_map).
  
   -export([init/1, map/3, terminate/1]).
  
   init(Config) ->
     State = do_init(Config),
     State.
  
   map(Offset, Message, State) ->
     Message.
  
   terminate(State) ->
     do_cleanup(State).

NOTE: Since state is immutable, it's actually shared between the routes.

Data Types

callback_fun()

callback_fun() = fun((kflow:offset(), _InputMessage) -> _OutputMessage)

Generated by EDoc