Node Reference
==============

This section provides detailed documentation for all node types in Flokkit Flow.

Node Lifecycle
--------------

All nodes in Flokkit Flow follow a consistent lifecycle with optional hooks for resource management.

Lifecycle Methods
~~~~~~~~~~~~~~~~~

.. code-block:: python

   class MyNode(Node):
       async def on_start(self) -> None:
           """Called when the node starts.
           
           Use this to:
           - Initialize resources (database connections, file handles)
           - Set up internal state
           - Perform one-time setup operations
           """
           pass
       
       async def on_stop(self) -> None:
           """Called when the node stops.
           
           Use this to:
           - Close connections
           - Flush buffers
           - Clean up resources
           """
           pass
       
       async def on_error(self, error: Exception) -> None:
           """Called when an error occurs during processing.
           
           Use this to:
           - Log errors
           - Implement retry logic
           - Send alerts
           """
           pass
       
       async def on_complete(self) -> None:
           """Called when the node completes processing normally.
           
           Use this to:
           - Finalize results
           - Send completion notifications
           - Update metrics
           """
           pass

Creating Custom Nodes
---------------------

Basic Custom Node
~~~~~~~~~~~~~~~~~

.. code-block:: python

   from flow import Node
   
   class CustomProcessor(Node):
       def __init__(self, name: str, multiplier: int):
           super().__init__(name)
           self.multiplier = multiplier
       
       def _setup_ports(self) -> None:
           self.add_input_port("in", int)
           self.add_output_port("out", int)
       
       async def process(self) -> None:
           value = await self.get_input_port("in").receive()
           if value is not None:
               result = value * self.multiplier
               await self.get_output_port("out").send(result)

Stateful Node with Resources
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

.. code-block:: python

   class DatabaseWriter(Node):
       def __init__(self, name: str, connection_string: str):
           super().__init__(name)
           self.connection_string = connection_string
           self.connection = None
           self.batch = []
           self.batch_size = 100
       
       def _setup_ports(self) -> None:
           self.add_input_port("in", dict)
       
       async def on_start(self) -> None:
           self.connection = await asyncpg.connect(self.connection_string)
       
       async def on_stop(self) -> None:
           # Flush remaining batch
           if self.batch:
               await self._write_batch()
           
           # Close connection
           if self.connection:
               await self.connection.close()
       
       async def process(self) -> None:
           record = await self.get_input_port("in").receive()
           if record is not None:
               self.batch.append(record)
               
               if len(self.batch) >= self.batch_size:
                   await self._write_batch()
       
       async def _write_batch(self) -> None:
           if not self.batch:
               return
           
           await self.connection.executemany(
               "INSERT INTO records (data) VALUES ($1)",
               [(json.dumps(record),) for record in self.batch]
           )
           self.batch.clear()

Port Management
---------------

Adding Ports
~~~~~~~~~~~~

.. code-block:: python

   def _setup_ports(self) -> None:
       # Basic ports
       self.add_input_port("in", str)
       self.add_output_port("out", str)
       
       # With queue configuration
       self.add_input_port(
           "data",
           data_type=dict,
           queue_size=1000,
           full_strategy=QueueFullStrategy.DROP_OLD
       )
       
       # Multiple ports
       for i in range(3):
           self.add_output_port(f"out{i}", int)

Accessing Ports
~~~~~~~~~~~~~~~

.. code-block:: python

   async def process(self) -> None:
       # Get input
       in_port = self.get_input_port("in")
       value = await in_port.receive()
       
       # Check if port has data without blocking
       value = await in_port.receive(timeout=0)
       
       # Send output
       out_port = self.get_output_port("out")
       await out_port.send(result)
       
       # Send to specific port
       if condition:
           await self.get_output_port("out1").send(value)
       else:
           await self.get_output_port("out2").send(value)

Advanced Patterns
-----------------

Multi-Input Processing
~~~~~~~~~~~~~~~~~~~~~~

.. code-block:: python

   class Correlator(Node):
       """Correlates data from multiple sources."""
       
       def _setup_ports(self) -> None:
           self.add_input_port("primary", dict)
           self.add_input_port("secondary", dict)
           self.add_output_port("out", dict)
       
       async def process(self) -> None:
           # Wait for data from both inputs
           primary_task = asyncio.create_task(
               self.get_input_port("primary").receive()
           )
           secondary_task = asyncio.create_task(
               self.get_input_port("secondary").receive()
           )
           
           # Get first available
           done, pending = await asyncio.wait(
               {primary_task, secondary_task},
               return_when=asyncio.FIRST_COMPLETED
           )
           
           # Cancel pending
           for task in pending:
               task.cancel()
           
           # Process result
           for task in done:
               if task.result() is not None:
                   await self.process_data(task.result())

Windowing Node
~~~~~~~~~~~~~~

.. code-block:: python

   class WindowAggregator(Node):
       """Aggregates data in time windows."""
       
       def __init__(self, name: str, window_seconds: float):
           super().__init__(name)
           self.window_seconds = window_seconds
           self.window = []
           self.window_start = None
       
       def _setup_ports(self) -> None:
           self.add_input_port("in", float)
           self.add_output_port("out", dict)
       
       async def process(self) -> None:
           value = await self.get_input_port("in").receive(
               timeout=self.window_seconds
           )
           
           now = asyncio.get_event_loop().time()
           
           if self.window_start is None:
               self.window_start = now
           
           if value is not None:
               self.window.append(value)
           
           # Check if window expired
           if now - self.window_start >= self.window_seconds:
               await self._emit_window()
       
       async def _emit_window(self) -> None:
           if self.window:
               result = {
                   "count": len(self.window),
                   "sum": sum(self.window),
                   "avg": sum(self.window) / len(self.window),
                   "min": min(self.window),
                   "max": max(self.window)
               }
               await self.get_output_port("out").send(result)
           
           self.window.clear()
           self.window_start = asyncio.get_event_loop().time()

Error Handling Node
~~~~~~~~~~~~~~~~~~~

.. code-block:: python

   class ResilientProcessor(Node):
       """Process with retry logic."""
       
       def __init__(self, name: str, max_retries: int = 3):
           super().__init__(name)
           self.max_retries = max_retries
       
       async def process(self) -> None:
           value = await self.get_input_port("in").receive()
           if value is None:
               return
           
           retries = 0
           while retries < self.max_retries:
               try:
                   result = await self._process_with_errors(value)
                   await self.get_output_port("out").send(result)
                   break
               except Exception as e:
                   retries += 1
                   await self.on_error(e)
                   if retries < self.max_retries:
                       await asyncio.sleep(2 ** retries)  # Exponential backoff
                   else:
                       # Send to error output
                       await self.get_output_port("error").send({
                           "value": value,
                           "error": str(e),
                           "retries": retries
                       })

Best Practices
--------------

1. **Resource Management**: Always implement `on_start()` and `on_stop()` for resources
2. **Error Handling**: Use `on_error()` for logging and recovery
3. **Buffering**: Consider batching for I/O operations
4. **Timeouts**: Use receive timeouts to prevent indefinite blocking
5. **Cancellation**: Handle task cancellation gracefully
6. **Type Safety**: Always specify port data types
7. **Testing**: Design nodes to be testable in isolation