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

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

Methods

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")

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

def connect(self, *dest_ports:Union["Node", "Port"]) -> "Port"

# connect the default output port to a input port of a node
# if dest_port element is a Node, then the default input port is used

def emit(self, data:Any, name:str=DEFAULT_PORT_NAME)

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

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

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