Struct TapNode

pub struct TapNode { /* private fields */ }

The TAP Node

The TAP Node is the core component responsible for coordinating message processing, routing, and delivery to TAP Agents. It serves as a central hub for all TAP communications and transaction coordination.

Core Responsibilities

• Agent Management: Registration and deregistration of TAP Agents
• PlainMessage Processing: Processing incoming and outgoing messages through middleware chains
• PlainMessage Routing: Determining the appropriate recipient for each message
• Event Publishing: Broadcasting node events to subscribers
• Scalability: Managing concurrent message processing through worker pools

Lifecycle

1. Create a node with appropriate configuration
2. Register one or more agents with the node
3. Start the processor pool (if high throughput is required)
4. Process incoming/outgoing messages
5. Publish and respond to events

Thread Safety

The TapNode is designed to be thread-safe and can be shared across multiple threads using an Arc<TapNode>. All internal mutability is handled through appropriate synchronization primitives.

Implementations

impl TapNode

pub fn new(config: NodeConfig) -> Self

Create a new TAP node with the given configuration

pub async fn init_storage(&mut self) -> Result<()>

Initialize storage asynchronously

pub async fn start(&mut self, config: ProcessorPoolConfig) -> Result<()>

Start the node

pub async fn receive_message(&self, message: Value) -> Result<()>

Receive and process an incoming message

This method handles the complete lifecycle of an incoming message:

1. Determining the message type (plain, signed, or encrypted)
2. Verifying signatures or routing to agents for decryption
3. Processing the resulting plain messages through the pipeline
4. Routing and dispatching to the appropriate agents

Parameters

• message - The message as a JSON Value (can be plain, JWS, or JWE)

Returns

• Ok(()) if the message was successfully processed
• Err(Error) if there was an error during processing

pub async fn receive_message_from_source( &self, message: Value, source_type: SourceType, source_identifier: Option<&str>, ) -> Result<()>

Receive and process an incoming message with source information

This method handles the complete lifecycle of an incoming message with tracking of where the message came from.

Parameters

• message - The message as a JSON Value (can be plain, JWS, or JWE)
• source_type - The type of source (https, internal, websocket, etc.)
• source_identifier - Optional identifier for the source (URL, agent DID, etc.)

Returns

• Ok(()) if the message was successfully processed
• Err(Error) if there was an error during processing

pub async fn dispatch_message( &self, target_did: String, message: PlainMessage, ) -> Result<()>

Dispatch a message to an agent by DID

pub async fn send_message( &self, sender_did: String, message: PlainMessage, ) -> Result<String>

Send a message to an agent

This method now includes comprehensive delivery tracking and actual message delivery. For internal recipients (registered agents), messages are delivered directly. For external recipients, messages are delivered via HTTP with tracking.

pub async fn register_agent(&self, agent: Arc<TapAgent>) -> Result<()>

Register a new agent with the node

This method registers an agent with the TAP Node and automatically initializes DID-specific storage for the agent. The storage directory structure follows:

• ~/.tap/{sanitized_did}/transactions.db (default)
• {tap_root}/{sanitized_did}/transactions.db (if custom TAP root is configured)

Storage Initialization

When an agent is registered, a dedicated SQLite database is created for that agent’s DID. This ensures transaction isolation between different agents while maintaining a consistent storage structure. If storage initialization fails, the agent registration continues but a warning is logged.

Arguments

• agent - The TapAgent to register with the node

Returns

• Ok(()) if the agent was successfully registered
• Err(Error) if agent registration fails

pub async fn unregister_agent(&self, did: &str) -> Result<()>

Unregister an agent from the node

pub fn list_agents(&self) -> Vec<String>

Get a list of registered agent DIDs

pub fn agents(&self) -> &Arc<AgentRegistry>

Get a reference to the agent registry

pub fn event_bus(&self) -> &Arc<EventBus>

Get a reference to the event bus

pub fn resolver(&self) -> &Arc<MultiResolver>

Get a reference to the resolver

pub fn processor_pool_mut(&mut self) -> &mut Option<ProcessorPool>

Get a mutable reference to the processor pool This is a reference to Option<ProcessorPool> to allow starting the pool after node creation

pub fn config(&self) -> &NodeConfig

Get the node configuration

pub fn storage(&self) -> Option<&Arc<Storage>>

Get a reference to the storage (if available)

pub fn agent_storage_manager(&self) -> Option<&Arc<AgentStorageManager>>

Get a reference to the agent storage manager (if available)

pub async fn set_storage(&mut self, storage: Storage) -> Result<()>

Set storage for testing purposes This allows injecting in-memory databases for complete test isolation

Trait Implementations

impl Clone for TapNode

fn clone(&self) -> TapNode

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.