Meshmakers.Octo.ConstructionKit.Models.System.Communication 3.4.38

Octo Communication Controller Services

ASP.NET Core web service for managing communication adapters and pools in the Octo Mesh ecosystem.

Table of Contents

• Overview
• Architecture
• Quick Start
• Project Structure
• Core Components
• API Reference
• Configuration
• Development
• Testing
• Deployment
• System Events

Overview

The Communication Controller Service is the central hub for coordinating communication between external adapters (data pipeline executors) and pools (device groups) in a multi-tenant environment.

Key Features

• Adapter Management: Registration, configuration, and monitoring of communication adapters
• Pool Management: Organization and management of device groups
• Pipeline Orchestration: Deployment and control of data pipelines
• Real-time Communication: Bidirectional SignalR connections for live updates
• Multi-Tenancy: Complete tenant isolation at all levels

Architecture

┌─────────────────────────────────────────────────────────────────────┐
│                        External Clients                             │
│    ┌─────────────────┐              ┌─────────────────┐             │
│    │  Edge Adapter   │              │  Pool Operator  │             │
│    └────────┬────────┘              └────────┬────────┘             │
└─────────────┼────────────────────────────────┼──────────────────────┘
              │ SignalR                        │ SignalR
              ▼                                ▼
┌─────────────────────────────────────────────────────────────────────┐
│                  Communication Controller Service                   │
│  ┌─────────────────────────────────────────────────────────────┐    │
│  │                       SignalR Hubs                          │    │
│  │   ┌──────────────┐              ┌──────────────┐            │    │
│  │   │  AdapterHub  │              │   PoolHub    │            │    │
│  │   │ /adapterHub  │              │  /poolHub    │            │    │
│  │   └──────┬───────┘              └──────┬───────┘            │    │
│  └──────────┼─────────────────────────────┼────────────────────┘    │
│             ▼                             ▼                         │
│  ┌──────────────────────────────────────────────────────────────┐   │
│  │                      Service Layer                           │   │
│  │  ┌───────────────┐ ┌───────────────┐ ┌────────────────────┐  │   │
│  │  │AdapterService │ │  PoolService  │ │PipelineDebugService│  │   │
│  │  └───────┬───────┘ └───────┬───────┘ └─────────┬──────────┘  │   │
│  └──────────┼─────────────────┼───────────────────┼─────────────┘   │
│             │                 │                   │                 │
│  ┌──────────┴─────────────────┴───────────────────┴─────────────┐   │
│  │                       Cache Layer                            │   │
│  │  ┌───────────────┐              ┌───────────────┐            │   │
│  │  │ AdapterCache  │              │   PoolCache   │            │   │
│  │  └───────────────┘              └───────────────┘            │   │
│  └──────────────────────────────────────────────────────────────┘   │
│             │                                                       │
│  ┌──────────┴───────────────────────────────────────────────────┐   │
│  │                    Repository Layer                          │   │
│  │              ┌─────────────────────────┐                     │   │
│  │              │ CommunicationRepository │                     │   │
│  │              └───────────┬─────────────┘                     │   │
│  └──────────────────────────┼───────────────────────────────────┘   │
└─────────────────────────────┼───────────────────────────────────────┘
                              │
                              ▼
                    ┌─────────────────┐
                    │    MongoDB      │
                    │ (via Octo RE)   │
                    └─────────────────┘

Layered Architecture

Quick Start

Prerequisites

• .NET 10.0 SDK
• MongoDB (via Octo Runtime Engine)
• RabbitMQ (for message bus)
• Valid OAuth2/JWT token for authentication

Build

# Restore and build solution
dotnet restore Octo.CommunicationController.sln
dotnet build Octo.CommunicationController.sln --configuration Release

Run

# Development mode
dotnet run --project src/CommunicationControllerServices/CommunicationControllerServices.csproj

# With specific configuration
dotnet run --project src/CommunicationControllerServices/CommunicationControllerServices.csproj --configuration Debug

Run Tests

# All tests
dotnet test --configuration Release

# Specific test class
dotnet test --filter "FullyQualifiedName~AdapterServiceTests"

# Specific test method
dotnet test --filter "FullyQualifiedName~RegisterAdapterTests.ShouldRegisterNewAdapter"

Project Structure

octo-communication-controller-services/
├── src/
│   ├── CommunicationControllerServices/      # Main service
│   │   ├── Hubs/                             # SignalR Hubs
│   │   ├── Services/                         # Business Logic
│   │   ├── Caches/                           # In-Memory Caches
│   │   ├── Repository/                       # Data Access
│   │   ├── Consumers/                        # Message Bus Consumers
│   │   ├── SystemApi/v1/Controllers/         # System-Level API
│   │   ├── TenantApi/v1/Controllers/         # Tenant-Level API
│   │   └── Options/                          # Configuration Classes
│   │
│   ├── SystemCommunicationCkModel/           # Construction Kit Model
│   │   ├── ConstructionKit/
│   │   │   ├── ckModel.yaml                  # Main model definition
│   │   │   ├── types/                        # Entity types
│   │   │   ├── enums/                        # Enumerations
│   │   │   ├── attributes/                   # Attribute definitions
│   │   │   └── associations/                 # Relationships
│   │   └── Generated/                        # Generated C# code
│   │
│   └── CommunicationControllerServices.Resources/  # Localization
│
├── tests/
│   └── CommunicationControllerService.Tests/ # Unit Tests (TUnit)
│
├── devops-build/                             # CI/CD Pipelines
├── Directory.Build.props                      # Build Configuration
├── CLAUDE.md                                  # AI Assistant Guidelines
└── Octo.CommunicationController.sln          # Solution File

Core Components

SignalR Hubs

AdapterHub (/{tenantId}/adapterHub)

Manages adapter lifecycle and pipeline debugging.

PoolHub (/{tenantId}/poolHub)

Manages pool operator connections. Workloads (Adapters + Applications) are not pushed back through this hub — the Communication Operator subscribes to WorkloadDeployedAsync / WorkloadUndeployedAsync on /operatorHub for the Helm-based deploy flow.

Services

AdapterService

Core business logic for adapter operations:

public interface IAdapterService
{
    Task RegisterAdapterAsync(string tenantId, RtEntityId adapterRtEntityId, string connectionId);
    Task UnregisterAsync(string tenantId, RtEntityId adapterRtEntityId);
    Task<AdapterConfiguration> GetAdapterConfigurationAsync(string tenantId, RtEntityId adapterRtEntityId);
    Task DeployPipelineAsync(string tenantId, RtEntityId pipelineRtEntityId);
    Task UndeployPipelineAsync(string tenantId, RtEntityId pipelineRtEntityId);
    Task SetAdapterCommunicationStateOnlineAsync(string tenantId, RtEntityId adapterRtEntityId);
    Task SetAdapterCommunicationStateOfflineAsync(string tenantId, RtEntityId adapterRtEntityId);
}

PoolService

Business logic for pool management. Workloads managed by the pool (Adapters + Applications) are fanned out as Helm deploys by DeployPoolAsync via the /operatorHub SignalR channel — no adapter list is returned to the operator on pool registration.

public interface IPoolService
{
    Task<OctoObjectId> RegisterPoolOperatorAsync(string tenantId, string poolName, string connectionId);
    Task UnregisterPoolOperatorAsync(string tenantId, string poolName);
    Task DeployPoolAsync(string tenantId, OctoObjectId poolRtId);
    Task UndeployPoolAsync(string tenantId, OctoObjectId poolRtId);
    Task SetCommunicationStateOnlineAsync(string tenantId, OctoObjectId poolRtId);
    Task SetCommunicationStateOfflineAsync(string tenantId, OctoObjectId poolRtId);
}

PipelineDebugService

Debug information for pipeline executions:

public interface IPipelineDebugService
{
    Task StartDebugPointAsync(string executionId, string nodeId, object data);
    Task<IReadOnlyCollection<DebugPoint>> GetDebugPointsAsync(string executionId);
    Task<IReadOnlyCollection<DebugHistory>> GetDebugHistoryAsync(string tenantId);
}

CommunicationEventService

System event logging for auditing and monitoring:

public interface ICommunicationEventService
{
    Task StoreEventAsync(string tenantId, RtEventLevelsEnum level, string message);
    Task StoreInformationEventAsync(string tenantId, string message);
    Task StoreWarningEventAsync(string tenantId, string message);
    Task StoreErrorEventAsync(string tenantId, string message);
    void StoreSystemInformationEvent(string message);
    void StoreSystemWarningEvent(string message);
    void StoreSystemErrorEvent(string message);
}

This service wraps IEventRepository from Meshmakers.Octo.Services.Notifications to handle scoped lifetime requirements for singleton services.

Caches

The cache system uses a dual-interface pattern for separation of concerns:

// Read interface
public interface IAdapterCache
{
    bool TryGetTenant(string tenantId, out AdapterTenant? adapterTenant);
    AdapterTenant AddOrUpdateTenant(string tenantId);
    void RemoveTenant(string tenantId);
}

// Publish interface
public interface IAdapterCachePublish
{
    Task LoadConfigurationAsync(string tenantId);
    Task PublishConfigurationAsync(string tenantId);
    Task ReloadConfigurationAsync(ComControllerAdapterUpdate configuration);
}

Registration as singleton with multiple interfaces:

services.AddSingletonMultipleInterfaces<AdapterCache, IAdapterCache, IAdapterCachePublish>();

Repository

Abstraction of MongoDB persistence via Octo Runtime Engine:

public interface ICommunicationRepository
{
    // Adapter operations
    Task<IReadOnlyCollection<RtAdapter>> GetAdaptersAsync(string tenantId);
    Task<RtAdapter> GetAdapterAsync(string tenantId, RtEntityId adapterRtEntityId);

    // Pool operations
    Task<IReadOnlyCollection<RtPool>> GetPoolsAsync(string tenantId);
    Task CreatePoolAsync(string tenantId, string poolName);
    Task SetPoolDeploymentStateAsync(string tenantId, OctoObjectId poolRtId, RtDeploymentStateEnum state);

    // Pipeline operations
    Task<IReadOnlyCollection<RtPipeline>> GetPipelinesAsync(string tenantId, RtEntityId adapterRtEntityId);
    Task SetPipelineDeploymentStateAsync(string tenantId, RtEntityId pipelineRtId, RtDeploymentStateEnum state, object? error);
}

API Reference

System API

Base route: system/v{version}/

Tenant API

Base route: {tenantId}/v{version}/

Adapter Endpoints

Pool Endpoints

Pipeline Endpoints

Pipeline Debug Endpoints

Configuration

Environment Variables

All settings can be overridden via environment variables with prefix OCTO_.

appsettings.json

{
  "CommunicationController": {
    "PublicUrl": "https://localhost:5015",
    "AuthorityUrl": "https://localhost:5003",
    "BrokerHost": "localhost",
    "BrokerPort": 5672,
    "BrokerVirtualHost": "/",
    "BrokerUser": "guest",
    "BrokerPassword": "guest",
    "MinLogLevel": "Information",
    "InstancePrefix": ""
  },
  "System": {
    // Octo System Configuration
  }
}

CommunicationControllerOptions

public class CommunicationControllerOptions
{
    public string PublicUrl { get; set; }           // Service public URL
    public string AuthorityUrl { get; set; }        // OAuth authority
    public string BrokerHost { get; set; }          // RabbitMQ host
    public int BrokerPort { get; set; }             // RabbitMQ port
    public string BrokerVirtualHost { get; set; }   // RabbitMQ vhost
    public string BrokerUser { get; set; }          // RabbitMQ user
    public string BrokerPassword { get; set; }      // RabbitMQ password
    public string MinLogLevel { get; set; }         // Log level
    public string InstancePrefix { get; set; }      // Multi-instance prefix
}

Authentication & Authorization

JWT Bearer authentication with scope-based authorization:

Development

Construction Kit Model

The CK model defines domain objects in YAML:

# types/adapter.yaml
id: Adapter
ckTypeId: system-communication
inherits: DeployableEntity
abstract: true
attributes:
  - $ref: "CommunicationState"
  - $ref: "ConfigurationState"
associations:
  - $ref: "Manages"

Generate model:

dotnet build src/SystemCommunicationCkModel/SystemCommunicationCkModel.csproj

Generated C# types are located in src/SystemCommunicationCkModel/Generated/.

Exception Handling Pattern

Services use factory methods for exceptions:

// Correct - use factory methods
throw AdapterServiceException.TenantNotEnabled(tenantId);
throw AdapterServiceException.PipelineNotFound(tenantId, pipelineId);

// Wrong - do not use generic exceptions
throw new AdapterServiceException("Tenant not enabled");  // Don't use!

State Management

Deployment States

enum RtDeploymentStateEnum
{
    Undeployed = 0,  // Not yet deployed
    Pending = 1,     // Deployment requested
    Deployed = 2,    // Actively deployed
    Error = 3        // Deployment error
}

Communication States

enum RtCommunicationStateEnum
{
    Unregistered = 0,  // Not registered
    Online = 1,        // Connected
    Offline = 2        // Disconnected
}

Service Registration

Multi-interface registration for services:

// Extension methods
services.AddSingletonMultipleInterfaces<AdapterCache, IAdapterCache, IAdapterCachePublish>();
services.AddScopedMultipleInterfaces<DefaultConfigurationCreatorService,
    IDefaultConfigurationCreatorService, IConfigurationService>();

Testing

Framework

• TUnit (not xUnit or NUnit)
• NSubstitute for mocking

Test Structure

public class MyServiceTests : MyServiceTestsBase
{
    [Test]
    public async Task ShouldDoSomething_WhenCondition()
    {
        // Arrange - mocks configured in base class
        _mockRepository.GetAsync(Arg.Any<string>()).Returns(expectedResult);

        // Act
        var result = await _service.DoSomethingAsync();

        // Assert
        await Assert.That(result).IsNotNull();
        await Assert.That(result.Value).IsEqualTo(expected);
    }
}

Test Base Classes

AdapterServiceTestsBase:

• Pre-configured mock dependencies
• Test data setup
• Helper methods for entity creation

Running Tests

# All tests with verbose output
dotnet test --configuration Release --logger "console;verbosity=detailed"

# With code coverage
dotnet test --configuration Release --collect:"XPlat Code Coverage"

Deployment

Docker Build

# Build image
docker build \
  --build-arg OCTO_VERSION=1.0.0 \
  --build-arg OCTO_PRIVATE_NUGET_SERVICE=https://nuget.example.com \
  -f src/CommunicationControllerServices/Dockerfile \
  -t octo-communication-controller:1.0.0 .

Dockerfile Multi-Stage Build

1. base: ASP.NET 10.0 Runtime
2. build: .NET 10.0 SDK + Private NuGet
3. publish: Release Build
4. final: Production Runtime

Kubernetes Deployment

The service supports horizontal scaling with:

• Distributed caches via message bus
• SignalR with Redis backplane (optional)
• Stateless design for load balancing

Health Checks

• /system/v1/communication/ping - Basic health check
• Custom AdapterHealthChecks for detailed status monitoring

External Dependencies

Octo package versions are centrally managed via $(OctoVersion) in Directory.Build.props.

System Events

The service logs important business events for auditing and monitoring using the Octo Notification system.

Event Architecture

Events are stored per-tenant in MongoDB via IEventRepository. The service uses ICommunicationEventService as a wrapper to handle scoped lifetime requirements for singleton services.

// Registration in Program.cs
builder.Services.AddOctoNotification();
builder.Services.AddSingleton<ICommunicationEventService, CommunicationEventService>();

// Service start event
app.Lifetime.ApplicationStarted.Register(() =>
{
    var eventRepository = app.Services.GetRequiredService<IEventRepository>();
    eventRepository.StoreSystemInformationEvent(RtEventSourcesEnum.CommunicationService,
        "Communication Controller Services started.");
});

Logged Events

Usage Example

// In a singleton service
internal class MyService(ICommunicationEventService eventService)
{
    public async Task DoOperationAsync(string tenantId)
    {
        try
        {
            // ... perform operation ...

            await eventService.StoreInformationEventAsync(tenantId,
                "Operation completed successfully.");
        }
        catch (Exception e)
        {
            await eventService.StoreErrorEventAsync(tenantId,
                $"Operation failed: {e.Message}");
            throw;
        }
    }
}

Event Levels

License

Proprietary - Meshmakers GmbH

Support

For questions or issues, please contact the development team.