eth::EthWatchService¶

Ties together EventWatcher, ABI decoding, and eth message dispatch. More...

#include <eth_watch_service.hpp>

Public Functions¶

	Name
	EthWatchService() =default
	~EthWatchService()
	EthWatchService(const EthWatchService & ) =delete
EthWatchService &	operator=(const EthWatchService & ) =delete
	EthWatchService(EthWatchService && ) =default
EthWatchService &	operator=(EthWatchService && ) =default
bool	initialize(EthWatchServiceConfig config, WatchEventNotificationCallback callback) Initialize production orchestration from chain/watch config.
void	run(boost::asio::io_context & io) Start production eth-watch orchestration on `io`.
void	stop() Stop schedulers and discovery clients created by run().
bool	initialized() const
size_t	runtime_chain_count() const
size_t	scheduler_count() const
size_t	peer_queue_count() const
size_t	discovery_client_count() const
size_t	discv4_fallback_count() const
size_t	active_runner_count() const
WatchStatsSnapshot	aggregate_runtime_stats() const
EthWatchRuntimeStatsSnapshot	aggregate_connection_stats() const
std::shared_ptr< EthPeerQueue >	peer_queue(const std::string & chain_name) const
void	set_send_callback(SendCallback cb) Provide a callback used to send outgoing eth messages.
void	set_eth_message_schemas(std::vector< EthMessageSchema > schemas) Configure schema-aware runtime decoders for this service/session.
EventWatchId	watch_event(const codec::Address & contract_address, const std::string & event_signature, const std::vector< abi::AbiParam > & params, DecodedEventCallback callback, std::optional< uint64_t > from_block =std::nullopt, std::optional< uint64_t > to_block =std::nullopt) Register a watch for a specific contract event.
uint64_t	tip() const Return the highest block number seen so far (0 if none).
std::optional< Hash256 >	tip_hash() const Return the hash of the highest block seen, if any.
void	unwatch(EventWatchId id) Remove a previously registered subscription.
size_t	subscription_count() const Return the number of active subscriptions.
WatchStatsSnapshot	stats() const Return a snapshot of the live watcher counters.
void	process_message(uint8_t eth_msg_id, rlp::ByteView payload) Process a raw eth wire message payload.
void	process_receipts(const std::vector< codec::Receipt > & receipts, const std::vector< Hash256 > & tx_hashes, uint64_t block_number, const Hash256 & block_hash, const std::vector< std::vector< uint32_t > > & log_indices ={}) Directly process a batch of receipts for a known block.
void	process_new_block(const NewBlockMessage & msg, const Hash256 & block_hash) Directly process a NewBlock message.
void	request_receipts(const Hash256 & block_hash, uint64_t block_number) Request receipts for a block by hash.

Detailed Description¶

class eth::EthWatchService;

Ties together EventWatcher, ABI decoding, and eth message dispatch.

Usage:

Call set_send_callback() so the service can emit GetReceipts requests.
Register subscriptions via watch_event().
Feed incoming eth wire messages via process_message().
Matching logs trigger the registered DecodedEventCallback.

Thread-safety: not thread-safe; all calls must be externally synchronized.

Public Functions Documentation¶

function EthWatchService¶

EthWatchService() =default

function ~EthWatchService¶

~EthWatchService()

function EthWatchService¶

EthWatchService(
    const EthWatchService & 
) =delete

function operator=¶

EthWatchService & operator=(
    const EthWatchService & 
) =delete

function EthWatchService¶

EthWatchService(
    EthWatchService && 
) =default

function operator=¶

EthWatchService & operator=(
    EthWatchService && 
) =default

function initialize¶

bool initialize(
    EthWatchServiceConfig config,
    WatchEventNotificationCallback callback
)

Initialize production orchestration from chain/watch config.

The method stores config only. Call run() with an io_context to create schedulers, peer queues, optional discovery fallback, and live sessions.

function run¶

void run(
    boost::asio::io_context & io
)

Start production eth-watch orchestration on io.

function stop¶

void stop()

Stop schedulers and discovery clients created by run().

function initialized¶

bool initialized() const

function runtime_chain_count¶

size_t runtime_chain_count() const

function scheduler_count¶

size_t scheduler_count() const

function peer_queue_count¶

size_t peer_queue_count() const

function discovery_client_count¶

size_t discovery_client_count() const

function discv4_fallback_count¶

size_t discv4_fallback_count() const

function active_runner_count¶

size_t active_runner_count() const

function aggregate_runtime_stats¶

WatchStatsSnapshot aggregate_runtime_stats() const

function aggregate_connection_stats¶

EthWatchRuntimeStatsSnapshot aggregate_connection_stats() const

function peer_queue¶

std::shared_ptr< EthPeerQueue > peer_queue(
    const std::string & chain_name
) const

function set_send_callback¶

void set_send_callback(
    SendCallback cb
)

Provide a callback used to send outgoing eth messages.

Must be called before process_message() if automatic GetReceipts requests are desired. Safe to omit if the caller handles receipts manually via process_receipts().

function set_eth_message_schemas¶

void set_eth_message_schemas(
    std::vector< EthMessageSchema > schemas
)

Configure schema-aware runtime decoders for this service/session.

function watch_event¶

EventWatchId watch_event(
    const codec::Address & contract_address,
    const std::string & event_signature,
    const std::vector< abi::AbiParam > & params,
    DecodedEventCallback callback,
    std::optional< uint64_t > from_block =std::nullopt,
    std::optional< uint64_t > to_block =std::nullopt
)

Register a watch for a specific contract event.

Parameters:

contract_address Contract to watch; empty address = any contract.
event_signature Canonical Solidity signature, e.g. "Transfer(address,address,uint256)".
params Full parameter list in declaration order (mark indexed ones with AbiParam::indexed = true).
callback Called for each matching decoded log.
from_block Optional lower block bound for the filter.
to_block Optional upper block bound for the filter.

Return: WatchId that can be passed to unwatch().

function tip¶

inline uint64_t tip() const

Return the highest block number seen so far (0 if none).

function tip_hash¶

inline std::optional< Hash256 > tip_hash() const

Return the hash of the highest block seen, if any.

function unwatch¶

void unwatch(
    EventWatchId id
)

Remove a previously registered subscription.

function subscription_count¶

inline size_t subscription_count() const

Return the number of active subscriptions.

function stats¶

inline WatchStatsSnapshot stats() const

Return a snapshot of the live watcher counters.

function process_message¶

void process_message(
    uint8_t eth_msg_id,
    rlp::ByteView payload
)

Process a raw eth wire message payload.

Parameters:

eth_msg_id Eth-layer message id (offset already subtracted).
payload Raw message bytes.

Call this from your generic_handler with the eth-layer message id (i.e. already minus the rlpx offset) and the raw payload bytes.

Handles NewBlockHashes (0x01), NewBlock (0x07), Receipts (0x10). When a send callback is registered, automatically emits GetReceipts for new blocks.

function process_receipts¶

void process_receipts(
    const std::vector< codec::Receipt > & receipts,
    const std::vector< Hash256 > & tx_hashes,
    uint64_t block_number,
    const Hash256 & block_hash,
    const std::vector< std::vector< uint32_t > > & log_indices ={}
)

Directly process a batch of receipts for a known block.

Parameters:

receipts The receipts for all transactions in the block.
tx_hashes Corresponding transaction hashes (same order).
block_number Block number.
block_hash Block hash.

function process_new_block¶

void process_new_block(
    const NewBlockMessage & msg,
    const Hash256 & block_hash
)

Directly process a NewBlock message.

function request_receipts¶

void request_receipts(
    const Hash256 & block_hash,
    uint64_t block_number
)

Request receipts for a block by hash.

Parameters:

block_hash Hash of the block whose receipts are wanted.
block_number Block number (stored for context in the callback).

Encodes and sends a GetReceipts message via the send callback. Records the pending request so the Receipts response can be correlated.

Updated on 2026-06-05 at 17:22:18 -0700