polyflux-websocket 0.0.1a0
WebSocket client implementation for Polyflux
License: Apache-2.0
Requires Python: >=3.12

# Polyflux WebSocket

WebSocket client implementation for Polyflux, providing full bidirectional communication with elegant mapping of WebSocket's event-driven nature to the Protocol interfaces.

## Features

- **Full Protocol Conformance**: Implements all 10 umbrella protocols from the core package
- **Bidirectional Communication**: Maps WebSocket's nature to exchange patterns with correlation IDs
- **Streaming Support**: Native streaming, subscription, and portal patterns
- **Auto-Reconnection**: Built-in reconnection with exponential backoff
- **Heartbeat**: Automatic ping/pong for connection health monitoring
- **Compression**: Per-message deflate support
- **Type Safety**: Full type annotations with Python 3.12+ features
- **Async/Await**: Comprehensive async support with sync wrappers

## Installation

```bash
uv pip install polyflux-websocket
```

## Quick Start

```python
import asyncio
from polyflux.websocket import WebSocketClient, WebSocketMessage

async def main():
    # Initialize client
    client = WebSocketClient(
        uri="wss://api.example.com/ws",
        auto_reconnect=True,
        heartbeat_interval=30.0
    )
    
    # Connect to WebSocket server
    async with client:
        # Send a message
        msg = WebSocketMessage(data={"action": "hello"}, message_type="text")
        await client.asend("greeting", msg)
        
        # Receive a message
        response = await client.areceive("messages")
        print(f"Received: {response.data}")
        
        # Request-response pattern with correlation
        request = WebSocketMessage(data={"query": "status"}, message_type="text")
        response = await client.aexchange("status", request)
        print(f"Status: {response.data}")

# Run the async main function
asyncio.run(main())
```

## Usage Examples

### Basic Connection Management

```python
from polyflux.websocket import WebSocketClient

# Create client
client = WebSocketClient(uri="ws://localhost:8000")

# Connect
await client.aconnect()

# Check connection status
if client.connected():
    print("Connected!")

# Disconnect
await client.adisconnect()
```

### Message Exchange Patterns

```python
# Simple send without response
msg = WebSocketMessage(data={"event": "update"}, message_type="text")
success = await client.asend("events", msg)

# Receive next available message
message = await client.areceive("queue")

# Request-response with correlation
request = WebSocketMessage(data={"method": "getUser", "id": 123})
response = await client.aexchange("api", request)
# Response is automatically correlated using correlation_id
# Works with echo servers too - the envelope is preserved!
```

### Subscription Pattern

```python
# Subscribe to a topic
async for message in client.asubscribe("updates.user.*"):
    print(f"Update: {message.data}")
    # Process each subscription message
    if message.data.get("type") == "logout":
        break

# Unsubscribe
await client.aunsubscribe("updates.user.*")
```

### Streaming Pattern

```python
# Start a stream
stream_msg = WebSocketMessage(data={"stream": "sensor_data"})
async for chunk in client.astream("sensors", stream_msg):
    sensor_value = chunk.data.get("value")
    print(f"Sensor: {sensor_value}")
    
    # Break on end-of-stream
    if chunk.data.get("end"):
        break
```

### Portal Pattern (Bidirectional Channel)

```python
# Open a portal for bidirectional communication
portal_id = await client.aopen_portal("chat_room")

# Send through portal
msg = WebSocketMessage(data={"message": "Hello room!"})
await client.asend_through_portal(portal_id, msg)

# Receive from portal
response = await client.areceive_from_portal(portal_id)

# Close portal when done
await client.aclose_portal(portal_id)
```

### Error Handling and Retry

```python
# Configure retry policy
client.with_retry({
    "max_attempts": 3,
    "base_delay": 1.0,
    "exponential_base": 2,
    "jitter": True
})

# Execute with automatic retry
async def risky_operation():
    return await client.aexchange("api", request)

result = await client.aretry(risky_operation)
```

### Authentication

```python
# Bearer token authentication
await client.aauthenticate(
    type="bearer",
    token="your-api-token"
)

# Custom authentication
await client.aauthenticate(
    type="custom",
    custom_handler=my_auth_handler,
    credentials={"user": "alice", "key": "secret"}
)

# Check authentication status
if await client.aauthenticated():
    print("Authenticated!")
```

### Metrics and Observability

```python
# Configure logging
client.configure_logging(level="INFO", format="json")

# Collect metrics
metrics = await client.acollect_metrics()
print(f"Messages sent: {metrics['messages_sent']}")
print(f"Messages received: {metrics['messages_received']}")
print(f"Bytes transferred: {metrics['bytes_sent'] + metrics['bytes_received']}")

# Distributed tracing
async with client.trace("process_order", order_id="123") as span:
    # Traced operations
    response = await client.aexchange("orders", order_request)
    span.set_attribute("response_time", response.metadata.get("processing_time"))
```

## WebSocket Message Format

The `WebSocketMessage` PDU supports various message types and automatic envelope handling:

```python
from polyflux.websocket import WebSocketMessage

# Text message with JSON data
text_msg = WebSocketMessage(
    data={"key": "value"},
    message_type="text",
    correlation_id="custom-id-123"
)

# Binary message
binary_msg = WebSocketMessage(
    data=b"\x00\x01\x02\x03",
    message_type="binary"
)

# Control frames
ping_msg = WebSocketMessage(
    data="keepalive",
    message_type="ping"
)

# Messages are automatically wrapped in envelopes with metadata
envelope = {
    "data": {"key": "value"},
    "correlation_id": "custom-id-123",
    "timestamp": "2024-01-01T00:00:00Z",
    "metadata": {...}
}
```

## Configuration Options

```python
client = WebSocketClient(
    # Connection
    uri="wss://api.example.com/ws",
    subprotocols=["chat.v1", "json"],  # WebSocket subprotocols
    compression="deflate",             # Enable compression
    
    # Reconnection
    auto_reconnect=True,               # Automatic reconnection
    max_reconnect_attempts=10,         # Maximum attempts
    reconnect_delay=1.0,               # Initial delay (seconds)
    max_reconnect_delay=60.0,          # Maximum delay (seconds)
    
    # Timeouts
    request_timeout=30.0,              # Default exchange timeout
    heartbeat_interval=30.0,           # Ping/pong interval
    
    # Queues
    max_queue_size=1000,               # Maximum queued messages
    
    # Additional options
    headers={"Authorization": "Bearer token"},  # Custom headers
    custom_option="value"              # Any additional options
)
```

## Advanced Features

### Middleware Support

```python
# Add middleware for message processing
def logging_middleware(next_func):
    def wrapper(msg):
        print(f"Processing: {msg.correlation_id}")
        result = next_func(msg)
        print(f"Completed: {msg.correlation_id}")
        return result
    return wrapper

client.with_middleware(logging_middleware)
```

### Message Transformation

```python
# Add transformers
def add_timestamp(msg):
    msg.metadata["processed_at"] = datetime.now().isoformat()
    return msg

client.pdu_processor_manager.add_outbound_transformer(add_timestamp)
```

### Connection State Persistence

```python
# Save connection state for recovery
state = await client.asave_state()

# Later: restore state after crash/restart
await client.aload_state(state)
```

### Circuit Breaker

```python
# Configure circuit breaker for fault tolerance
client.with_circuit_breaker({
    "failure_threshold": 5,
    "recovery_timeout": 60,
    "expected_exception": ConnectionError
})

# Check circuit status
if client.circuit_open():
    print("Circuit breaker is OPEN - backing off")
```

## Architecture

The WebSocket implementation follows a manager-based architecture:

- **ConnectionManager**: Handles connection lifecycle and reconnection
- **ExchangeManager**: Manages request-response correlation
- **StreamManager**: Controls bidirectional streaming with backpressure
- **SubscriptionManager**: Pattern-based event subscriptions  
- **PortalManager**: Multiplexed communication channels
- **ReliabilityManager**: Error handling, retry, and circuit breaker
- **StateManager**: State persistence and transfer
- **ObservabilityManager**: Logging, metrics, and tracing

## Protocol Conformance

The WebSocket client implements all umbrella protocols from `polyflux-core`:

- ✅ **BaseExchange**: Send/receive with correlation
- ✅ **BaseConnection**: Connection lifecycle  
- ✅ **BaseDiscovery**: Service discovery
- ✅ **BaseState**: State persistence
- ✅ **BaseLifecycle**: Context management
- ✅ **BaseConduit**: Streaming, subscription, portal patterns
- ✅ **BaseReliability**: Retry, circuit breaker, caching
- ✅ **BaseConfiguration**: Authentication and config
- ✅ **BasePDUProcessor**: Middleware and transformation
- ✅ **BaseObservability**: Logging, metrics, tracing

## Protocol Data Units

The `WebSocketMessage` class implements both `Outbound` and `Inbound` protocols, supporting:

- Text and binary frame types
- Control frames (close, ping, pong)
- Message fragmentation and reassembly
- Correlation IDs for request-response mapping
- Full JSON Schema generation for AI/agent integration
- Automatic envelope wrapping for correlation and metadata

## Development

```bash
# Install dependencies
uv pip install -e .[dev]

# Run tests
uv run pytest

# Type checking
uv run mypy .

# Format code
uv run ruff format .

# Lint
uv run ruff check --fix .
```

## License

Apache 2.0 - See LICENSE file for details