Firebird/Concepts

Firebird

Overview

pipeline

• A Pipe is a data processing unit, it may have zero or more input ports, and may have zero or more output ports
• A Generator is a Pipe that does not have any input ports
• A Sink is a Pipe that does not have any output ports

wiring

You can connect an output port to one ore more input port to set the data flow direction.

Models

Node

Node

Represent a node in the pipeline. A node can have bunch of input ports and bunch of output ports. Each port has a unique id within the node.

Properties

id:              The id of the node, it is unique within a pipeline
title:           A human readable name of this node
description:     Detailed description of this node
pipeline:        The pipeline this node belongs to
input_port_ids:  tuple of all input port ids
output_port_ids: tuple of all output port ids
input:           The default input port
output:          The default output port

method: get_port

def get_port(self, id:str) -> Optional["Port"]
# each port has a unique id within the node.
# return a port given port id

# You can also use [] to get port by name
node["foo"]  # same as node.get_port("foo")

method: on_message

on_message(self, port_name:str, data:Any)

Called when this node receives data. port_name is the name of the port which receive the data, data is the JSON payload of the data.

connect

# You can use >>, << to connect nodes, here are examples:
nodeA >> nodeB               # connect default output port of nodeA to default input port of node B
nodeA["foo"] >> nodeB        # connect port "foo" of nodeA to default input port of node B
nodeA["foo"] >> nodeB["bar"] # connect port "foo" of nodeA to port "bar" of node B

# You can also use << as a reverse of >>

emit

def emit(self, data:Any, port_id:str=DEFAULT_OUTPUT_PORT_ID)

# emit data to it's output port specified by port_id

Generator

Represent a node that does not have input ports.

Methods:

def pump(self)

# derived class must override this method to pump data to the pipeline.

Sink

Sink

Represent a node that does not have output ports.

Port

Represent a input port or output port

Properties:

• type: either PortType.INPUT or PortType.OUTPUT, represent it is a input port or output port
• name: name of the port, it is unique within the node
• owner: the node which this port belongs to
• connected_ports: other ports connected to this port

Methods:

def connect(self, *ports: Union["Port", Node])

# connect this port to other port

def emit(self, json_data:Any)

# emit data to this port

Pipeline

Represent a pipeline, which has

• bunch of nodes
• A node's output port can connect to another node's input port

Properties:

• mq: message queue
• node_dict: dictionary, key is node name, value is node

Methods:

def message_loop(self):

# enters a message loop, allow each node to process data

def add_node(self, name:str, node_class, *argc, **kwargs)

# add a node to the pipeline, *argc, and **kwargs will be passed to the constructor

def connect(self, *, src_node_name, src_port_name=DEFAULT_PORT_NAME, dst_node_name, dst_port_name=DEFAULT_PORT_NAME)

# connect one node's output port to another node's input port

RabbitMQ