Port & Adapter Pattern

Hexagonal Architecture in TAHO

TAHO embraces hexagonal architecture (also known as Ports and Adapters) to create components that are:

Testable: Business logic isolated from infrastructure
Flexible: Swap implementations without changing core logic
Maintainable: Clear separation of concerns
Reusable: Adapters can be shared across components

Key Principle: Your business logic should not know or care whether it’s reading from PostgreSQL or MongoDB, serving HTTP or gRPC, or running on AWS or Azure.

Architecture Overview

Core Concepts

Ports

Ports define the interfaces between your domain logic and the outside world:

Input Ports: How the outside world interacts with your domain
Output Ports: How your domain interacts with external systems

Adapters

Adapters implement the ports for specific technologies:

Primary Adapters: Drive the application (HTTP handlers, CLI, message consumers)
Secondary Adapters: Are driven by the application (databases, APIs, file systems)

Domain Core

Domain Core contains pure business logic with no infrastructure dependencies

Practical Example: Order Service

Let’s build an order processing service using TAHO’s port & adapter pattern:

1. Define the Domain Model

// Pure domain models - no infrastructure dependencies
#[derive(Debug, Clone)]
pub struct Order {
    pub id: OrderId,
    pub customer_id: CustomerId,
    pub items: Vec<OrderItem>,
    pub status: OrderStatus,
    pub total: Money,
    pub created_at: Timestamp,
}

#[derive(Debug, Clone)]
pub struct OrderItem {
    pub product_id: ProductId,
    pub quantity: u32,
    pub price: Money,
}

#[derive(Debug, Clone)]
pub enum OrderStatus {
    Pending,
    Confirmed,
    Processing,
    Shipped,
    Delivered,
    Cancelled,
}

// Domain value objects
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
pub struct OrderId(pub String);

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CustomerId(pub String);

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ProductId(pub String);

#[derive(Debug, Clone, PartialEq)]
pub struct Money {
    pub amount: i64, // in cents
    pub currency: String,
}

2. Define Ports

Input Ports
Output Ports

// ports/input.rs
use async_trait::async_trait;
use crate::domain::*;

#[async_trait]
pub trait OrderService: Send + Sync {
    async fn create_order(&self, req: CreateOrderRequest) -> Result<Order, OrderError>;
    async fn get_order(&self, id: &OrderId) -> Result<Order, OrderError>;
    async fn update_status(&self, id: &OrderId, status: OrderStatus) -> Result<Order, OrderError>;
    async fn list_orders(&self, customer_id: &CustomerId) -> Result<Vec<Order>, OrderError>;
    async fn cancel_order(&self, id: &OrderId) -> Result<Order, OrderError>;
}

pub struct CreateOrderRequest {
    pub customer_id: CustomerId,
    pub items: Vec<OrderItemRequest>,
}

pub struct OrderItemRequest {
    pub product_id: ProductId,
    pub quantity: u32,
}

#[derive(Debug, thiserror::Error)]
pub enum OrderError {
    #[error("Order not found")]
    NotFound,
    #[error("Invalid order state")]
    InvalidState,
    #[error("Insufficient inventory")]
    InsufficientInventory,
    #[error("Payment failed")]
    PaymentFailed,
    #[error("Internal error: {0}")]
    Internal(String),
}

// ports/output.rs
use async_trait::async_trait;
use crate::domain::*;

// Repository port
#[async_trait]
pub trait OrderRepository: Send + Sync {
    async fn save(&self, order: &Order) -> Result<(), RepositoryError>;
    async fn find_by_id(&self, id: &OrderId) -> Result<Option<Order>, RepositoryError>;
    async fn find_by_customer(&self, customer_id: &CustomerId) -> Result<Vec<Order>, RepositoryError>;
    async fn update(&self, order: &Order) -> Result<(), RepositoryError>;
}

// Inventory port
#[async_trait]
pub trait InventoryService: Send + Sync {
    async fn check_availability(&self, items: &[OrderItem]) -> Result<bool, InventoryError>;
    async fn reserve_items(&self, order_id: &OrderId, items: &[OrderItem]) -> Result<(), InventoryError>;
    async fn release_items(&self, order_id: &OrderId) -> Result<(), InventoryError>;
}

// Payment port
#[async_trait]
pub trait PaymentService: Send + Sync {
    async fn process_payment(&self, order: &Order) -> Result<PaymentResult, PaymentError>;
    async fn refund_payment(&self, order: &Order) -> Result<RefundResult, PaymentError>;
}

// Notification port
#[async_trait]
pub trait NotificationService: Send + Sync {
    async fn send_order_confirmation(&self, order: &Order) -> Result<(), NotificationError>;
    async fn send_status_update(&self, order: &Order) -> Result<(), NotificationError>;
}

3. Implement Domain Logic

// domain/service.rs
use crate::ports::{input::*, output::*};

pub struct OrderServiceImpl {
    repository: Arc<dyn OrderRepository>,
    inventory: Arc<dyn InventoryService>,
    payment: Arc<dyn PaymentService>,
    notifications: Arc<dyn NotificationService>,
}

#[async_trait]
impl OrderService for OrderServiceImpl {
    async fn create_order(&self, req: CreateOrderRequest) -> Result<Order, OrderError> {
        // Pure business logic - no infrastructure concerns
        
        // 1. Validate request
        if req.items.is_empty() {
            return Err(OrderError::InvalidRequest("No items in order".into()));
        }
        
        // 2. Calculate total
        let items = self.build_order_items(&req.items).await?;
        let total = self.calculate_total(&items);
        
        // 3. Check inventory through port
        if !self.inventory.check_availability(&items).await? {
            return Err(OrderError::InsufficientInventory);
        }
        
        // 4. Create order
        let order = Order {
            id: OrderId::generate(),
            customer_id: req.customer_id,
            items,
            status: OrderStatus::Pending,
            total,
            created_at: Timestamp::now(),
        };
        
        // 5. Reserve inventory
        self.inventory.reserve_items(&order.id, &order.items).await?;
        
        // 6. Process payment through port
        match self.payment.process_payment(&order).await {
            Ok(_) => {
                // 7. Update status
                let mut confirmed_order = order;
                confirmed_order.status = OrderStatus::Confirmed;
                
                // 8. Save through repository port
                self.repository.save(&confirmed_order).await?;
                
                // 9. Send notification through port
                self.notifications.send_order_confirmation(&confirmed_order).await?;
                
                Ok(confirmed_order)
            }
            Err(e) => {
                // Rollback inventory reservation
                self.inventory.release_items(&order.id).await?;
                Err(OrderError::PaymentFailed)
            }
        }
    }
    
    // Other methods implementation...
}

4. Create Adapters

HTTP Adapter
Database Adapter
Message Queue Adapter

// adapters/http.rs
use warp::{Filter, Reply};
use crate::ports::input::*;

pub fn create_routes(
    service: Arc<dyn OrderService>,
) -> impl Filter<Extract = impl Reply, Error = warp::Rejection> + Clone {
    let create = warp::post()
        .and(warp::path("orders"))
        .and(warp::body::json())
        .and(with_service(service.clone()))
        .and_then(handle_create_order);
        
    let get = warp::get()
        .and(warp::path("orders"))
        .and(warp::path::param())
        .and(with_service(service.clone()))
        .and_then(handle_get_order);
        
    create.or(get)
}

async fn handle_create_order(
    req: CreateOrderDto,
    service: Arc<dyn OrderService>,
) -> Result<impl Reply, warp::Rejection> {
    // Convert HTTP DTO to domain request
    let domain_req = CreateOrderRequest {
        customer_id: CustomerId(req.customer_id),
        items: req.items.into_iter()
            .map(|i| OrderItemRequest {
                product_id: ProductId(i.product_id),
                quantity: i.quantity,
            })
            .collect(),
    };
    
    // Call domain service
    match service.create_order(domain_req).await {
        Ok(order) => {
            // Convert domain model to HTTP response
            let response = OrderDto::from(order);
            Ok(warp::reply::json(&response))
        }
        Err(e) => {
            // Map domain errors to HTTP errors
            Err(map_domain_error(e))
        }
    }
}

// adapters/postgres.rs
use sqlx::{PgPool, postgres::PgRow};
use crate::ports::output::*;

pub struct PostgresOrderRepository {
    pool: PgPool,
}

#[async_trait]
impl OrderRepository for PostgresOrderRepository {
    async fn save(&self, order: &Order) -> Result<(), RepositoryError> {
        let mut tx = self.pool.begin().await?;
        
        // Save order
        sqlx::query(
            "INSERT INTO orders (id, customer_id, status, total_amount, total_currency, created_at) 
             VALUES ($1, $2, $3, $4, $5, $6)"
        )
        .bind(&order.id.0)
        .bind(&order.customer_id.0)
        .bind(&order.status.to_string())
        .bind(order.total.amount)
        .bind(&order.total.currency)
        .bind(order.created_at.0)
        .execute(&mut tx)
        .await?;
        
        // Save order items
        for item in &order.items {
            sqlx::query(
                "INSERT INTO order_items (order_id, product_id, quantity, price_amount, price_currency) 
                 VALUES ($1, $2, $3, $4, $5)"
            )
            .bind(&order.id.0)
            .bind(&item.product_id.0)
            .bind(item.quantity as i32)
            .bind(item.price.amount)
            .bind(&item.price.currency)
            .execute(&mut tx)
            .await?;
        }
        
        tx.commit().await?;
        Ok(())
    }
    
    async fn find_by_id(&self, id: &OrderId) -> Result<Option<Order>, RepositoryError> {
        // Query and map to domain model
        let row = sqlx::query_as::<_, OrderRow>(
            "SELECT * FROM orders WHERE id = $1"
        )
        .bind(&id.0)
        .fetch_optional(&self.pool)
        .await?;
        
        match row {
            Some(r) => {
                let items = self.load_items(&id).await?;
                Ok(Some(self.map_to_domain(r, items)))
            }
            None => Ok(None)
        }
    }
    
    // Other methods...
}

// adapters/rabbitmq.rs
use lapin::{Connection, Channel};
use crate::ports::output::*;

pub struct RabbitMQNotificationService {
    channel: Channel,
    exchange: String,
}

#[async_trait]
impl NotificationService for RabbitMQNotificationService {
    async fn send_order_confirmation(&self, order: &Order) -> Result<(), NotificationError> {
        let event = OrderConfirmedEvent {
            order_id: order.id.0.clone(),
            customer_id: order.customer_id.0.clone(),
            total: order.total.amount,
            currency: order.total.currency.clone(),
            timestamp: Timestamp::now().0,
        };
        
        let payload = serde_json::to_vec(&event)?;
        
        self.channel
            .basic_publish(
                &self.exchange,
                "order.confirmed",
                Default::default(),
                &payload,
                Default::default(),
            )
            .await?;
            
        Ok(())
    }
    
    // Other methods...
}

5. Wire Everything Together

// main.rs
use crate::{
    domain::OrderServiceImpl,
    adapters::{
        http::create_routes,
        postgres::PostgresOrderRepository,
        rabbitmq::RabbitMQNotificationService,
        grpc::GrpcInventoryService,
        stripe::StripePaymentService,
    },
};

#[tokio::main]
async fn main() {
    // Create adapters
    let db_pool = PgPool::connect(&env::var("DATABASE_URL").unwrap()).await.unwrap();
    let repository = Arc::new(PostgresOrderRepository::new(db_pool));
    
    let amqp = Connection::connect(&env::var("AMQP_URL").unwrap()).await.unwrap();
    let notifications = Arc::new(RabbitMQNotificationService::new(amqp).await.unwrap());
    
    let inventory = Arc::new(GrpcInventoryService::new(&env::var("INVENTORY_URL").unwrap()));
    
    let payment = Arc::new(StripePaymentService::new(&env::var("STRIPE_KEY").unwrap()));
    
    // Create domain service with injected adapters
    let service = Arc::new(OrderServiceImpl::new(
        repository,
        inventory,
        payment,
        notifications,
    ));
    
    // Start HTTP server with service
    let routes = create_routes(service);
    warp::serve(routes)
        .run(([0, 0, 0, 0], 8080))
        .await;
}

TAHO-Specific Features

Dynamic Adapter Loading

TAHO allows adapters to be loaded dynamically at runtime:

// adapters.wit
interface adapter-registry {
    register-adapter: func(
        port-type: string,
        adapter-name: string,
        adapter-component: component-id
    ) -> result<adapter-id, error>;
    
    list-adapters: func(port-type: string) -> list<adapter-info>;
    
    select-adapter: func(
        port-type: string,
        criteria: adapter-criteria
    ) -> result<adapter-id, error>;
}

record adapter-criteria {
    performance: option<performance-requirement>,
    cost: option<cost-limit>,
    region: option<string>,
    features: list<string>,
}

Adapter Composition

Combine multiple adapters for advanced scenarios:

// Compose adapters for resilience
pub struct ResilientRepository {
    primary: Arc<dyn OrderRepository>,
    secondary: Arc<dyn OrderRepository>,
    cache: Arc<dyn CacheRepository>,
}

#[async_trait]
impl OrderRepository for ResilientRepository {
    async fn find_by_id(&self, id: &OrderId) -> Result<Option<Order>, RepositoryError> {
        // Try cache first
        if let Ok(Some(order)) = self.cache.find_by_id(id).await {
            return Ok(Some(order));
        }
        
        // Try primary
        match self.primary.find_by_id(id).await {
            Ok(order) => {
                if let Some(ref o) = order {
                    let _ = self.cache.save(o).await;
                }
                Ok(order)
            }
            Err(_) => {
                // Fallback to secondary
                self.secondary.find_by_id(id).await
            }
        }
    }
    
    async fn save(&self, order: &Order) -> Result<(), RepositoryError> {
        // Write through to both
        self.primary.save(order).await?;
        self.secondary.save(order).await?;
        let _ = self.cache.save(order).await;
        Ok(())
    }
}

Hot-Swappable Adapters

Change adapters without restarting:

// Runtime adapter switching
pub struct DynamicRepository {
    current: RwLock<Arc<dyn OrderRepository>>,
    registry: Arc<AdapterRegistry>,
}

impl DynamicRepository {
    pub async fn switch_adapter(&self, criteria: AdapterCriteria) -> Result<(), Error> {
        let new_adapter = self.registry.select_adapter("repository", criteria).await?;
        let mut current = self.current.write().await;
        *current = new_adapter;
        Ok(())
    }
}

Testing with Ports & Adapters

The architecture makes testing straightforward:

Unit Tests
Integration Tests

#[cfg(test)]
mod tests {
    use super::*;
    use mockall::mock;
    
    // Mock the ports
    mock! {
        TestRepo {}
        
        #[async_trait]
        impl OrderRepository for TestRepo {
            async fn save(&self, order: &Order) -> Result<(), RepositoryError>;
            async fn find_by_id(&self, id: &OrderId) -> Result<Option<Order>, RepositoryError>;
        }
    }
    
    #[tokio::test]
    async fn test_create_order_success() {
        // Arrange
        let mut mock_repo = MockTestRepo::new();
        mock_repo
            .expect_save()
            .times(1)
            .returning(|_| Ok(()));
            
        let service = OrderServiceImpl::new(
            Arc::new(mock_repo),
            Arc::new(MockInventory::new()),
            Arc::new(MockPayment::new()),
            Arc::new(MockNotifications::new()),
        );
        
        // Act
        let result = service.create_order(test_request()).await;
        
        // Assert
        assert!(result.is_ok());
    }
}

#[tokio::test]
async fn test_end_to_end_order_flow() {
    // Use test adapters
    let repo = Arc::new(InMemoryRepository::new());
    let inventory = Arc::new(MockInventoryService::new());
    let payment = Arc::new(TestPaymentService::new());
    let notifications = Arc::new(NoOpNotificationService);
    
    let service = Arc::new(OrderServiceImpl::new(
        repo.clone(),
        inventory,
        payment,
        notifications,
    ));
    
    // Start test server
    let routes = create_routes(service);
    let server = warp::test::request()
        .method("POST")
        .path("/orders")
        .json(&create_order_request())
        .reply(&routes)
        .await;
        
    assert_eq!(server.status(), 201);
    
    // Verify order was saved
    let order_id = extract_order_id(&server);
    let saved = repo.find_by_id(&order_id).await.unwrap();
    assert!(saved.is_some());
}

Best Practices

Keep Ports Focused

One port per concern (don’t mix repository and messaging)
Use domain language in port definitions
Avoid leaking infrastructure details into ports
Keep port interfaces small and cohesive

Domain Purity

No infrastructure imports in domain code
Use value objects for type safety
Validate at domain boundaries
Express business rules clearly
Avoid anemic domain models

Adapter Guidelines

Adapters should be thin wrappers
Handle all infrastructure errors
Map between domain and infrastructure types
Don’t put business logic in adapters
Make adapters replaceable

Testing Strategy

Unit test domain logic with mocked ports
Integration test adapters against real systems
Use test doubles for external dependencies
Test error scenarios thoroughly
Verify adapter contracts

Common Patterns

Repository Pattern

// Generic repository trait
#[async_trait]
pub trait Repository<T, ID>: Send + Sync {
    async fn find_by_id(&self, id: &ID) -> Result<Option<T>, RepositoryError>;
    async fn find_all(&self) -> Result<Vec<T>, RepositoryError>;
    async fn save(&self, entity: &T) -> Result<(), RepositoryError>;
    async fn delete(&self, id: &ID) -> Result<(), RepositoryError>;
}

// Specialized for each aggregate
pub trait OrderRepository: Repository<Order, OrderId> {
    async fn find_by_customer(&self, customer_id: &CustomerId) -> Result<Vec<Order>, RepositoryError>;
    async fn find_by_status(&self, status: OrderStatus) -> Result<Vec<Order>, RepositoryError>;
}

Unit of Work Pattern

#[async_trait]
pub trait UnitOfWork: Send + Sync {
    async fn begin(&self) -> Result<Box<dyn Transaction>, Error>;
}

#[async_trait]
pub trait Transaction: Send + Sync {
    async fn commit(self: Box<Self>) -> Result<(), Error>;
    async fn rollback(self: Box<Self>) -> Result<(), Error>;
    
    fn orders(&self) -> &dyn OrderRepository;
    fn inventory(&self) -> &dyn InventoryService;
}

Event Publishing

#[async_trait]
pub trait EventPublisher: Send + Sync {
    async fn publish(&self, event: DomainEvent) -> Result<(), PublishError>;
}

#[derive(Debug, Clone)]
pub enum DomainEvent {
    OrderCreated(OrderCreatedEvent),
    OrderCancelled(OrderCancelledEvent),
    PaymentProcessed(PaymentProcessedEvent),
}

Next Steps

Creating Components

Build your first TAHO component using ports & adapters

Component Communication

Learn how components interact in TAHO’s mesh

Adapter Catalog

Browse pre-built adapters for common services

Testing Guide

Comprehensive testing strategies for TAHO components

Getting Started

Core Concepts

Developer Guide

Configuration

Release Notes

Port & Adapter Pattern

Hexagonal Architecture in TAHO

Architecture Overview

Core Concepts

Ports

Adapters

Domain Core

Practical Example: Order Service

1. Define the Domain Model

2. Define Ports

3. Implement Domain Logic

4. Create Adapters

5. Wire Everything Together

TAHO-Specific Features

Dynamic Adapter Loading

Adapter Composition

Hot-Swappable Adapters

Testing with Ports & Adapters

Best Practices

Common Patterns

Repository Pattern

Unit of Work Pattern

Event Publishing

Next Steps

Creating Components

Component Communication

Adapter Catalog

Testing Guide

Getting Started

Core Concepts

Developer Guide

Configuration

Release Notes

​Hexagonal Architecture in TAHO

​Architecture Overview

​Core Concepts

​Ports

​Adapters

​Domain Core

​Practical Example: Order Service

​1. Define the Domain Model

​2. Define Ports

​3. Implement Domain Logic

​4. Create Adapters

​5. Wire Everything Together

​TAHO-Specific Features

​Dynamic Adapter Loading

​Adapter Composition

​Hot-Swappable Adapters

​Testing with Ports & Adapters

​Best Practices

​Common Patterns

​Repository Pattern

​Unit of Work Pattern

​Event Publishing

​Next Steps

Creating Components

Component Communication

Adapter Catalog

Testing Guide

Hexagonal Architecture in TAHO

Architecture Overview

Core Concepts

Ports

Adapters

Domain Core

Practical Example: Order Service

1. Define the Domain Model

2. Define Ports

3. Implement Domain Logic

4. Create Adapters

5. Wire Everything Together

TAHO-Specific Features

Dynamic Adapter Loading

Adapter Composition

Hot-Swappable Adapters

Testing with Ports & Adapters

Best Practices

Common Patterns

Repository Pattern

Unit of Work Pattern

Event Publishing

Next Steps