AIP Transport Layer

from aip.transport import Transport

class Transport(ABC):
    @abstractmethod
    async def connect(self, url: str, **kwargs) -> None:
        """Connect to remote endpoint"""

    @abstractmethod
    async def send(self, data: bytes) -> None:
        """Send data"""

    @abstractmethod
    async def receive(self) -> bytes:
        """Receive data"""

    @abstractmethod
    async def close(self) -> None:
        """Close connection"""

    @abstractmethod
    async def wait_closed(self) -> None:
        """Wait for connection to fully close"""

    @property
    @abstractmethod
    def is_connected(self) -> bool:
        """Check connection status"""

from aip.transport import WebSocketTransport

# Create and configure
transport = WebSocketTransport(
    ping_interval=30.0,
    ping_timeout=180.0,
    close_timeout=10.0,
    max_size=100 * 1024 * 1024  # 100MB
)

# Connect
await transport.connect("ws://localhost:8000/ws")

# Communicate
await transport.send(b"Hello Server")
data = await transport.receive()

# Cleanup
await transport.close()

from fastapi import WebSocket
from aip.transport import WebSocketTransport

async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()

    # Wrap existing WebSocket
    transport = WebSocketTransport(websocket=websocket)

    # Use unified interface
    data = await transport.receive()
    await transport.send(b"Response")

from aip.transport import TransportState

if transport.state == TransportState.CONNECTED:
    await transport.send(data)
else:
    logger.warning("Transport not connected")

try:
    await transport.connect("ws://localhost:8000/ws")
except ConnectionError as e:
    logger.error(f"Failed to connect: {e}")
    await handle_connection_failure()

try:
    await transport.send(data)
    response = await transport.receive()
except ConnectionError:
    logger.warning("Connection closed during operation")
    await reconnect()
except IOError as e:
    logger.error(f"I/O error: {e}")
    await handle_io_error(e)

try:
    # Close with timeout
    await transport.close()

    # Wait for complete shutdown
    await transport.wait_closed()
except Exception as e:
    logger.error(f"Error during shutdown: {e}")

# Server-side: Automatically uses FastAPIWebSocketAdapter
transport = WebSocketTransport(websocket=fastapi_websocket)

# Client-side: Automatically uses WebSocketsLibAdapter
transport = WebSocketTransport()
await transport.connect("ws://server:8000/ws")

from aip.messages import ClientMessage

# 1. Create Pydantic model
msg = ClientMessage(
    message_type="TASK_RESULT",
    task_id="task_123",
    result={"status": "success"}
)

# 2. Serialize to JSON string
json_str = msg.model_dump_json()

# 3. Encode to bytes
bytes_data = json_str.encode('utf-8')

# 4. Send via transport
await transport.send(bytes_data)

from aip.messages import ServerMessage

# 1. Receive bytes
bytes_data = await transport.receive()

# 2. Decode to JSON string
json_str = bytes_data.decode('utf-8')

# 3. Deserialize to Pydantic model
msg = ServerMessage.model_validate_json(json_str)

# 4. Use typed data
print(f"Task ID: {msg.task_id}")

    import gzip

    compressed = gzip.compress(large_data)
    await transport.send(compressed)
    ```

**Option 2: Chunking**
```python
    chunk_size = 1024 * 1024  # 1MB chunks
    for i in range(0, len(large_data), chunk_size):
        chunk = large_data[i:i+chunk_size]
        await transport.send(chunk)
    ```

**Option 3: Streaming Protocol**

Consider implementing a custom streaming protocol for very large payloads.

[→ See message encoding details in Protocol Reference](./protocols.md)

**High Throughput Strategy:**

For high message rates:

**Batch Messages:**
```python
    batch = [msg1, msg2, msg3, msg4]
    batch_json = json.dumps([msg.model_dump() for msg in batch])
    await transport.send(batch_json.encode('utf-8'))
    ```

**Reduce Ping Frequency:**
```python
transport = WebSocketTransport(
    ping_interval=60.0  # Less overhead
)

    transport = WebSocketTransport(
        ping_interval=10.0,  # Quick detection
        ping_timeout=30.0
    )
    ```

**Dedicated Connections:**
```python
# One transport per device (no sharing)
device_transports = {
    device_id: WebSocketTransport()
    for device_id in devices
}

from aip.transport.base import Transport

class CustomTransport(Transport):
    async def connect(self, url: str, **kwargs) -> None:
        # Custom connection logic
        self._connection = await custom_protocol.connect(url)

    async def send(self, data: bytes) -> None:
        await self._connection.write(data)

    async def receive(self) -> bytes:
        return await self._connection.read()

    async def close(self) -> None:
        await self._connection.shutdown()

    @property
    def is_connected(self) -> bool:
        return self._connection is not None and self._connection.is_open

from aip.protocol import AIPProtocol

# Use custom transport with protocol
transport = CustomTransport()
await transport.connect("custom://server:port")

protocol = AIPProtocol(transport)
await protocol.send_message(message)

transport = WebSocketTransport(
    ping_interval=15.0,   # Quick failure detection
    ping_timeout=45.0,
    close_timeout=5.0
)

transport = WebSocketTransport(
    ping_interval=30.0,   # Balance overhead and detection
    ping_timeout=180.0,
    close_timeout=10.0
)

transport = WebSocketTransport(
    ping_interval=60.0,   # Reduce battery usage
    ping_timeout=180.0,
    max_size=10 * 1024 * 1024  # 10MB for mobile
)

# Check before sending
if not transport.is_connected:
    logger.warning("Transport not connected, attempting reconnection")
    await reconnect_transport()

# Proceed with send
await transport.send(data)

from aip.resilience import ReconnectionStrategy

strategy = ReconnectionStrategy(max_retries=5)

try:
    await transport.send(data)
except ConnectionError:
    # Trigger reconnection
    await strategy.handle_disconnection(endpoint, device_id)

import logging

# Enable transport debug logs
logging.getLogger("aip.transport").setLevel(logging.DEBUG)

# Custom transport event logging
class LoggedTransport(WebSocketTransport):
    async def send(self, data: bytes) -> None:
        logger.debug(f"Sending {len(data)} bytes")
        await super().send(data)

    async def receive(self) -> bytes:
        data = await super().receive()
        logger.debug(f"Received {len(data)} bytes")
        return data

async with WebSocketTransport() as transport:
    await transport.connect("ws://localhost:8000/ws")
    await transport.send(data)
    # Automatic cleanup on exit

transport = WebSocketTransport()
try:
    await transport.connect("ws://localhost:8000/ws")
    await transport.send(data)
finally:
    await transport.close()

from aip.transport import (
    Transport,           # Abstract base class
    WebSocketTransport,  # WebSocket implementation
    TransportState,      # Connection states enum
)