Exth.Rpc.MessageHandler (Exth v0.4.2)

View Source

Handles JSON-RPC message correlation between requests and responses.

This module provides a mechanism for tracking JSON-RPC messages and their corresponding callers. Each client/transport gets its own handler instance to ensure proper isolation of message handling.

Features

  • Per-client message handling
  • Automatic request/response correlation
  • Batch request support
  • Process crash resilience
  • Efficient message routing
  • Support for any transport implementation
  • Subscription handling

Usage 

# Create a new handler for a client
{:ok, handler} = MessageHandler.new("wss://eth-mainnet.example.com")

# Send a request and wait for response
{:ok, response} = MessageHandler.call(handler, request, transport)

# Handle incoming responses
:ok = MessageHandler.handle_response(handler, encoded_response)

Message Flow

  1. Client creates its own handler instance with a unique name
  2. Client sends a request through call/4:
    • Determines request type (RPC vs Subscription)
    • Registers request ID with caller's PID
    • Sends request through transport
    • Waits for response
    • Cleans up registration
  3. Transport receives response and calls handle_response/2:
    • Deserializes response
    • Looks up registered caller
    • Sends response to caller
  4. Caller receives response and continues execution

Error Handling

The handler handles several error cases:

  • Process crashes (automatic cleanup via Registry)
  • Orphaned responses (no registered caller)
  • Timeouts (configurable per call)
  • Transport errors (propagated to caller)

Transport Implementation

Any transport implementation can be used with the handler as long as it:

Performance Considerations

  • Uses Registry for efficient process lookup
  • Automatic cleanup of registrations
  • No shared state between calls
  • Configurable timeouts per call

Summary

Types

handler()
request_id()
request_type()

Functions

call(handler, requests, transport, timeout \\ 5000)

Sends a request through the handler and waits for a response.

handle_response(handler, encoded_response)

Handles an incoming response by routing it to the appropriate caller.

new(rpc_url)

Creates a new handler instance for a client.

Types

handler()

@type handler() :: Registry.registry()

request_id()

@type request_id() :: pos_integer() | String.t()

request_type()

@type request_type() :: :rpc | :subscription

Functions

call(handler, requests, transport, timeout \\ 5000)

@spec call(
  handler(),
  [Exth.Rpc.Request.t()],
  Exth.Transport.Transportable.t(),
  timeout()
) ::
  {:ok, [Exth.Rpc.Response.t()]} | {:error, term()}

Sends a request through the handler and waits for a response.

This function handles the full request/response cycle:

  1. Determines request type (RPC vs Subscription)
  2. Registers the request ID with the caller's PID
  3. Sends the request through the transport
  4. Waits for the response
  5. Cleans up the registration

Parameters

  • handler - The handler instance to use
  • requests - The request(s) to send
  • transport - The transport to use
  • timeout - Optional timeout in milliseconds (default: 5000)

Returns

  • {:ok, response} - Successful response
  • {:error, term()} - Request failed

Example 

{:ok, response} = MessageHandler.call(handler, request, transport)

handle_response(handler, encoded_response)

@spec handle_response(handler(), String.t()) :: :ok | :error

Handles an incoming response by routing it to the appropriate caller.

This function is called by the transport when a response is received. It:

  1. Deserializes the response
  2. Looks up the registered caller
  3. Sends the response to the caller

For subscription events, it:

  1. Extracts the subscription ID from the notification
  2. Looks up the process that owns the subscription
  3. Delivers the event to that process

Parameters

  • handler - The handler instance to use
  • encoded_response - The encoded JSON-RPC response

Returns

  • :ok - Response was handled
  • :error - Response could not be handled

Example 

:ok = MessageHandler.handle_response(handler, encoded_response)

new(rpc_url)

@spec new(rpc_url :: String.t()) :: {:ok, handler()} | {:error, term()}

Creates a new handler instance for a client.

Each client should have its own handler instance to ensure proper isolation of message handling. The handler is identified by a unique name derived from the RPC URL.

Parameters

  • rpc_url - The RPC URL to use as the base for the handler name

Returns

  • {:ok, handler()} - Successfully created handler
  • {:error, term()} - Failed to create handler

Example 

{:ok, handler} = MessageHandler.new("wss://eth-mainnet.example.com")