# Core Concepts

Understanding Flokkit Flow's core concepts will help you build efficient and maintainable data pipelines.

## Flow-Based Programming (FBP)

Flokkit Flow implements the Flow-Based Programming paradigm, where:

- **Applications** are networks of "black box" processes
- **Processes** communicate via message passing
- **Connections** are defined separately from processes
- **Data** flows through the network as discrete packets

### Benefits of FBP

1. **Modularity**: Each component has a single responsibility
2. **Reusability**: Components can be reused in different flows
3. **Testability**: Components can be tested in isolation
4. **Visual Design**: Flows can be visualized as directed graphs
5. **Natural Parallelism**: Components run concurrently by default

## Nodes

Nodes are the building blocks of a flow. Flokkit Flow provides several node types:

### Source Nodes

Generate data for the pipeline:

```python
def file_reader():
    with open("data.txt") as f:
        for line in f:
            yield line.strip()
```

### Transform Nodes

Process data and produce outputs:

```python
def parse_json(line: str) -> dict:
    return json.loads(line)
```

### Sink Nodes

Consume data without producing outputs:

```python
def write_to_database(record: dict) -> None:
    db.insert(record)
```

### Special Nodes

- **Filter**: Conditionally pass data through
- **Tap**: Observe data without consuming it
- **Split**: Duplicate data to multiple paths
- **Merge**: Combine multiple streams

## Ports and Connections

### Ports

Every node has input and/or output ports:

- **Input Port**: Receives data from upstream nodes
- **Output Port**: Sends data to downstream nodes

Ports are typed, ensuring type safety across connections.

### Connections

Connections link output ports to input ports:

```python
# Connects output of 'source' to input of 'processor'
builder.graph.connect(
    source.name, "out",     # Source node, output port
    processor.name, "in"    # Target node, input port
)
```

## Queues and Backpressure

### Bounded Queues

Each connection uses a bounded `asyncio.Queue` to:

1. Prevent memory exhaustion
2. Provide natural backpressure
3. Enable concurrent execution

### Queue Strategies

When a queue is full, Flokkit Flow supports different strategies:

```python
from flow import QueueFullStrategy

# Block until space available (default)
QueueFullStrategy.BLOCK

# Drop new items
QueueFullStrategy.DROP_NEW  

# Drop oldest items
QueueFullStrategy.DROP_OLD

# Raise an error
QueueFullStrategy.ERROR
```

Example:

```python
# Configure queue size and strategy
(builder
    .source(fast_producer)
    .to(slow_consumer, 
        queue_size=100,
        full_strategy=QueueFullStrategy.DROP_OLD))
```

## Type Safety

Flokkit Flow leverages Python's type system for safety:

```python
from typing import List

def parse_numbers(text: str) -> List[int]:
    return [int(x) for x in text.split(",")]

def sum_numbers(numbers: List[int]) -> int:
    return sum(numbers)

# This will type-check correctly
(builder
    .source(lambda: "1,2,3,4,5")
    .to(parse_numbers)  # str -> List[int]
    .to(sum_numbers)    # List[int] -> int
    .to(print))         # int -> None
```

## Async Execution

### Automatic Adaptation

Flokkit Flow automatically adapts sync functions to async:

```python
def sync_function(x: int) -> int:
    time.sleep(0.1)  # Blocking call
    return x * 2

# Flokkit Flow wraps this in asyncio.to_thread()
builder.to(sync_function, is_blocking=True)
```

### Native Async Support

Async functions run directly on the event loop:

```python
async def fetch_data(url: str) -> dict:
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.json()
```

## Graph Execution

### DAG Verification

Flokkit Flow verifies that your graph is a Directed Acyclic Graph (DAG):

```python
# This will raise an error - cycles aren't allowed
A -> B -> C -> A  # ❌ Cycle detected!
```

### Execution Model

1. **Initialization**: All nodes start their `process()` coroutines
2. **Data Flow**: Data flows through connections via queues
3. **Concurrency**: Nodes process data concurrently
4. **Completion**: Nodes complete when sources are exhausted
5. **Shutdown**: Resources are cleaned up via lifecycle hooks

## Lifecycle Management

Nodes can implement lifecycle methods:

```python
class StatefulNode:
    async def on_start(self):
        """Called before processing starts"""
        self.connection = await database.connect()
    
    async def on_stop(self):
        """Called during shutdown"""
        await self.connection.close()
    
    async def on_error(self, error: Exception):
        """Called when an error occurs"""
        logger.error(f"Error in node: {error}")
    
    async def on_complete(self):
        """Called when processing completes normally"""
        await self.flush_buffers()
```

## Pipeline vs Server Patterns

### Pipeline Pattern (auto_stop=True)

For batch processing that completes:

```python
builder = flow("ETL Pipeline", auto_stop=True)
# Automatically stops when sources are exhausted
```

### Server Pattern (auto_stop=False)

For continuous processing:

```python
builder = flow("API Server", auto_stop=False)
# Runs until manually stopped
```

## Caching and Memoization

Enable caching for expensive operations:

```python
@cache  # or builder.to(expensive_op, cache=True)
def expensive_operation(data: str) -> Result:
    # This will only run once per unique input
    return complex_calculation(data)
```

## Best Practices

1. **Keep nodes simple**: Each node should do one thing well
2. **Use type hints**: Enable type checking for safety
3. **Handle errors**: Implement lifecycle hooks for robustness
4. **Configure queues**: Set appropriate sizes for your workload
5. **Monitor performance**: Use tap nodes for observability
6. **Test components**: Nodes should be testable in isolation

## Next Steps

- Explore [Basic Usage](guide/basic-usage.md) for practical examples
- Learn about [Async Programming](guide/async-programming.md) patterns
- Understand [Backpressure](guide/backpressure.md) handling
- See [Design Patterns](guide/patterns.md) for common solutions