mirror of
https://github.com/ethereum/go-ethereum.git
synced 2026-07-22 12:46:44 +00:00
rpc: add request and response interceptors for middleware support
This commit is contained in:
parent
0287666b7d
commit
d8f51fb0ac
7 changed files with 1576 additions and 28 deletions
493
ethereum_rpc_architecture.md
Normal file
493
ethereum_rpc_architecture.md
Normal file
|
|
@ -0,0 +1,493 @@
|
|||
# Go-Ethereum RPC Client Architecture Overview
|
||||
|
||||
## 1. Client Structure & Initialization
|
||||
|
||||
### Core Client struct (rpc/client.go)
|
||||
The `Client` struct is the main entry point for RPC communication:
|
||||
|
||||
```go
|
||||
type Client struct {
|
||||
idgen func() ID // subscription ID generator
|
||||
isHTTP bool // connection type: http, ws, or ipc
|
||||
services *serviceRegistry // service registry for method resolution
|
||||
|
||||
idCounter atomic.Uint32 // counter for request IDs
|
||||
|
||||
// Connection management
|
||||
reconnectFunc reconnectFunc // function to establish new connections
|
||||
writeConn jsonWriter // current connection (wrapped in httpConn, websocketCodec, or jsonCodec)
|
||||
|
||||
// Dispatch system (for non-HTTP)
|
||||
close chan struct{} // signal to close client
|
||||
closing chan struct{} // closed when client is quitting
|
||||
didClose chan struct{} // closed when client quits
|
||||
reconnected chan ServerCodec // where write/reconnect sends new connections
|
||||
readOp chan readOp // read messages from connection
|
||||
readErr chan error // errors from read loop
|
||||
reqInit chan *requestOp // register response IDs, takes write lock
|
||||
reqSent chan error // signals write completion, releases write lock
|
||||
reqTimeout chan *requestOp // removes response IDs when call timeout expires
|
||||
|
||||
// Configuration
|
||||
batchItemLimit int
|
||||
batchResponseMaxSize int
|
||||
}
|
||||
```
|
||||
|
||||
### Initialization Flow
|
||||
|
||||
1. **Dial** → **DialContext** → **DialOptions** (public API entry points)
|
||||
2. **DialOptions** parses URL and creates appropriate transport:
|
||||
- HTTP/HTTPS → `newClientTransportHTTP()`
|
||||
- WS/WSS → `newClientTransportWS()`
|
||||
- IPC → `newClientTransportIPC()`
|
||||
- stdio → `newClientTransportIO()`
|
||||
|
||||
3. **newClient()** creates the Client and initializes dispatch loop:
|
||||
```go
|
||||
func newClient(initctx context.Context, cfg *clientConfig, connect reconnectFunc) (*Client, error) {
|
||||
conn, err := connect(initctx) // Establish initial connection
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
c := initClient(conn, new(serviceRegistry), cfg)
|
||||
c.reconnectFunc = connect // Store reconnection function
|
||||
return c, nil
|
||||
}
|
||||
```
|
||||
|
||||
4. **initClient()** sets up the Client:
|
||||
- Creates channels for dispatch
|
||||
- Determines if HTTP or not (HTTP doesn't use dispatch loop)
|
||||
- Launches dispatch goroutine for non-HTTP connections
|
||||
|
||||
## 2. Configuration System (client_opt.go)
|
||||
|
||||
### ClientOption Pattern
|
||||
Uses functional options pattern for flexible configuration:
|
||||
|
||||
```go
|
||||
type ClientOption interface {
|
||||
applyOption(*clientConfig)
|
||||
}
|
||||
|
||||
type clientConfig struct {
|
||||
// HTTP settings
|
||||
httpClient *http.Client
|
||||
httpHeaders http.Header
|
||||
httpAuth HTTPAuth
|
||||
|
||||
// WebSocket options
|
||||
wsDialer *websocket.Dialer
|
||||
wsMessageSizeLimit *int64
|
||||
|
||||
// RPC handler options
|
||||
idgen func() ID
|
||||
batchItemLimit int
|
||||
batchResponseLimit int
|
||||
}
|
||||
```
|
||||
|
||||
### Available Options
|
||||
- `WithHTTPClient(client)` - Custom HTTP client
|
||||
- `WithHTTPAuth(authFunc)` - Authentication provider called per request
|
||||
- `WithHeader(key, value)` - Custom HTTP headers
|
||||
- `WithHeaders(header)` - Multiple headers
|
||||
- `WithWebsocketDialer(dialer)` - Custom WS dialer
|
||||
- `WithWebsocketMessageSizeLimit(limit)` - WS message size limit
|
||||
- `WithBatchItemLimit(limit)` - Batch request limits
|
||||
- `WithBatchResponseSizeLimit(limit)` - Batch response size limits
|
||||
|
||||
**Key Insight**: These options only configure the *client-side* creation. HTTPAuth is called during request preparation.
|
||||
|
||||
## 3. Connection Handling
|
||||
|
||||
### Three Main Transport Types
|
||||
|
||||
#### A. HTTP Transport (http.go)
|
||||
- **httpConn struct**: Wrapper that implements ServerCodec interface (but mostly stubbed)
|
||||
- **HTTP-specific behavior**:
|
||||
- No persistent connection (stateless)
|
||||
- No dispatch loop needed
|
||||
- Direct request/response cycle
|
||||
- Headers managed in `httpConn.headers` (protected by mutex)
|
||||
- Authentication via `HTTPAuth` function applied per-request
|
||||
|
||||
- **Request flow** (sendHTTP):
|
||||
```
|
||||
Client.CallContext()
|
||||
→ Client.sendHTTP() // Directly send via HTTP
|
||||
→ httpConn.doRequest() // Marshal, create HTTP request, auth, execute
|
||||
→ http.Client.Do() // Execute HTTP request
|
||||
→ JSON decode response
|
||||
→ op.resp <- response
|
||||
```
|
||||
|
||||
#### B. WebSocket Transport (websocket.go)
|
||||
- **websocketCodec struct**: Implements ServerCodec
|
||||
- **Features**:
|
||||
- Persistent connection
|
||||
- Ping/pong keepalive (30s interval)
|
||||
- Message size limit (default 32MB)
|
||||
- Origin validation
|
||||
- Connection pooling for write buffers
|
||||
|
||||
- **Creation**:
|
||||
```go
|
||||
newClientTransportWS()
|
||||
→ Create websocket.Dialer
|
||||
→ Apply custom headers and auth
|
||||
→ Return connect() function that:
|
||||
- Calls dialer.DialContext()
|
||||
- Wraps in newWebsocketCodec()
|
||||
```
|
||||
|
||||
- **Ping loop**: Separate goroutine in websocketCodec keeps connection alive
|
||||
|
||||
#### C. IPC/Stdio Transport
|
||||
- **jsonCodec struct**: Standard JSON codec wrapper
|
||||
- Simpler than WS, used for local domain sockets and stdio
|
||||
|
||||
### ServerCodec Interface (types.go)
|
||||
The abstraction that all transports must implement:
|
||||
|
||||
```go
|
||||
type ServerCodec interface {
|
||||
peerInfo() PeerInfo // Return connection metadata
|
||||
readBatch() (msgs, isBatch, err) // Read and parse JSON-RPC messages
|
||||
close() // Close the connection
|
||||
jsonWriter // Embedded interface
|
||||
}
|
||||
|
||||
type jsonWriter interface {
|
||||
writeJSON(ctx context.Context, msg interface{}, isError bool) error
|
||||
closed() <-chan interface{} // Channel closed when connection ends
|
||||
remoteAddr() string // Peer address
|
||||
}
|
||||
```
|
||||
|
||||
## 4. RPC Call Flow & Request Handling
|
||||
|
||||
### Single Request Flow (CallContext)
|
||||
```
|
||||
Client.CallContext(ctx, result, method, args...)
|
||||
1. Validate result is pointer or nil
|
||||
2. Create jsonrpcMessage with:
|
||||
- Version: "2.0"
|
||||
- ID: Next ID from counter
|
||||
- Method: Requested method
|
||||
- Params: JSON-encoded arguments
|
||||
3. Create requestOp with:
|
||||
- IDs: [msg.ID]
|
||||
- resp: channel for responses (buffered)
|
||||
- err: any error
|
||||
4. IF HTTP:
|
||||
→ Client.sendHTTP(ctx, op, msg)
|
||||
→ httpConn.doRequest() sends HTTP POST
|
||||
→ response decoded and sent to op.resp
|
||||
ELSE:
|
||||
→ Client.send(ctx, op, msg)
|
||||
→ Send op to reqInit channel (dispatch picks it up)
|
||||
→ Send msg on connection via c.write()
|
||||
→ Handler receives response, routes to op.resp
|
||||
5. op.wait(ctx, c) blocks until:
|
||||
- Context canceled (timeout)
|
||||
- Response received on op.resp
|
||||
6. Decode response and unmarshal into result
|
||||
```
|
||||
|
||||
### Batch Request Flow (BatchCallContext)
|
||||
Similar to single request but:
|
||||
- Creates multiple jsonrpcMessage objects
|
||||
- Sends all at once via sendBatchHTTP or send
|
||||
- Maps response IDs back to original request elements
|
||||
- Stores errors in BatchElem.Error fields
|
||||
|
||||
### Dispatch Loop (Non-HTTP Only)
|
||||
The dispatch goroutine (`Client.dispatch()`) is the heart of non-HTTP clients:
|
||||
|
||||
```go
|
||||
func (c *Client) dispatch(codec ServerCodec) {
|
||||
conn := c.newClientConn(codec) // Create handler for this connection
|
||||
go c.read(codec) // Launch read loop
|
||||
|
||||
for {
|
||||
select {
|
||||
// Close signal
|
||||
case <-c.close:
|
||||
return
|
||||
|
||||
// Read path: incoming messages
|
||||
case op := <-c.readOp: // Messages from read loop
|
||||
if op.batch:
|
||||
conn.handler.handleBatch(op.msgs)
|
||||
else:
|
||||
conn.handler.handleMsg(op.msgs[0])
|
||||
|
||||
case err := <-c.readErr: // Read error
|
||||
conn.close(err, lastOp)
|
||||
reading = false
|
||||
|
||||
// Reconnect path: new connection
|
||||
case newcodec := <-c.reconnected:
|
||||
conn.close(errClientReconnected, lastOp)
|
||||
conn = c.newClientConn(newcodec)
|
||||
conn.handler.addRequestOp(lastOp)
|
||||
|
||||
// Send path: outgoing requests
|
||||
case op := <-c.reqInit: // New request to send
|
||||
reqInitLock = nil // Take write lock
|
||||
conn.handler.addRequestOp(op)
|
||||
|
||||
case err := <-c.reqSent: // Send complete
|
||||
if err != nil:
|
||||
conn.handler.removeRequestOp(lastOp)
|
||||
reqInitLock = c.reqInit // Release write lock
|
||||
|
||||
// Timeout path
|
||||
case op := <-c.reqTimeout:
|
||||
conn.handler.removeRequestOp(op)
|
||||
}
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
### Read Loop
|
||||
```go
|
||||
func (c *Client) read(codec ServerCodec) {
|
||||
for {
|
||||
msgs, batch, err := codec.readBatch() // Block reading from connection
|
||||
if err != nil {
|
||||
c.readErr <- err
|
||||
return
|
||||
}
|
||||
c.readOp <- readOp{msgs, batch} // Send to dispatch
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
### Handler (handler.go)
|
||||
The handler processes messages and manages subscriptions:
|
||||
- Maps request IDs to pending requestOp objects
|
||||
- Routes responses to waiting callers
|
||||
- Manages subscriptions
|
||||
- Handles timeouts
|
||||
- Processes batches with response limits
|
||||
|
||||
## 5. WebSocket Connection Details
|
||||
|
||||
### WebSocket Dial (DialWebsocket / DialOptions with WS URL)
|
||||
1. Parse endpoint URL
|
||||
2. Extract origin and basic auth from URL
|
||||
3. Apply custom headers and auth from config
|
||||
4. Create websocket.Dialer with:
|
||||
- ReadBufferSize: 1024
|
||||
- WriteBufferSize: 1024
|
||||
- WriteBufferPool: Shared sync.Pool for efficiency
|
||||
- Proxy: http.ProxyFromEnvironment
|
||||
5. DialContext with prepared headers
|
||||
6. Wrap connection in websocketCodec
|
||||
7. Codec starts pingLoop goroutine
|
||||
|
||||
### WebSocket Message Size
|
||||
- Default read limit: 32 MB (wsDefaultReadLimit)
|
||||
- Configurable via WithWebsocketMessageSizeLimit
|
||||
- Connection reads with codec.SetReadLimit()
|
||||
|
||||
### WebSocket Ping/Pong
|
||||
- Ping sent every 30s when idle
|
||||
- Pong handler resets read deadline
|
||||
- Write timeout for ping: 5s
|
||||
- Pong expected within: 30s
|
||||
|
||||
### WebSocket Headers
|
||||
- Origin header set (for CORS)
|
||||
- User-Agent preserved
|
||||
- Custom headers from config applied
|
||||
- HTTP auth applied during connection
|
||||
|
||||
## 6. Context & Header Management
|
||||
|
||||
### HTTP Headers in Context (context_headers.go)
|
||||
Headers can be injected via context for per-request customization:
|
||||
|
||||
```go
|
||||
// Create context with headers
|
||||
ctx := NewContextWithHeaders(context.Background(), headers)
|
||||
|
||||
// Called with HTTP client:
|
||||
client.CallContext(ctx, result, "method")
|
||||
|
||||
// In doRequest(), headers are extracted and merged
|
||||
func headersFromContext(ctx context.Context) http.Header
|
||||
func setHeaders(dst http.Header, src http.Header) http.Header
|
||||
```
|
||||
|
||||
**Important**: Headers from context are merged with static headers, context headers override.
|
||||
|
||||
### Client Context Extraction
|
||||
Via `ClientFromContext(ctx)`:
|
||||
- Returns the Client associated with a request context
|
||||
- Used for "reverse calls" in handler methods
|
||||
- Enables handler methods to call back out on the client
|
||||
|
||||
## 7. Middleware Injection Points (Currently Limited)
|
||||
|
||||
### Existing Extension Points
|
||||
|
||||
1. **HTTPAuth Function**
|
||||
- Called during every HTTP request
|
||||
- Has full access to request headers
|
||||
- Can add authentication headers
|
||||
- **Limitation**: Only on HTTP, doesn't apply to WS
|
||||
|
||||
2. **Custom HTTP Client**
|
||||
- Can implement http.RoundTripper wrapper
|
||||
- Can intercept all HTTP traffic
|
||||
- Applied at HTTP client level
|
||||
- **Limitation**: Only HTTP
|
||||
|
||||
3. **Custom WebSocket Dialer**
|
||||
- Can implement custom dialing logic
|
||||
- Called for initial connection + reconnects
|
||||
- Limited middleware capability
|
||||
|
||||
4. **HTTP Headers via Context**
|
||||
- Per-request header injection
|
||||
- Applied in doRequest()
|
||||
- Limited to header manipulation
|
||||
|
||||
### Missing Middleware Patterns
|
||||
|
||||
1. **No request/response interception for non-HTTP**
|
||||
- WebSocket, IPC, Stdio bypass all middleware
|
||||
- Direct ServerCodec interface prevents layering
|
||||
|
||||
2. **No request/response logging hook**
|
||||
- No way to intercept jsonrpcMessage before/after
|
||||
- No built-in tracing/metrics
|
||||
|
||||
3. **No error interception**
|
||||
- No hook to transform or log errors
|
||||
- No metrics collection
|
||||
|
||||
4. **No subscription interception**
|
||||
- Subscribe requests bypass middleware
|
||||
- Subscription messages not intercepted
|
||||
|
||||
5. **No connection-level hooks**
|
||||
- No way to inject before connection established
|
||||
- No way to hook connection failures
|
||||
|
||||
## 8. Critical Code Paths for Middleware
|
||||
|
||||
### HTTP Path (Most Middleware-Friendly)
|
||||
```
|
||||
Client.CallContext()
|
||||
→ Client.sendHTTP()
|
||||
→ httpConn.doRequest()
|
||||
1. json.Marshal(msg) ← Can intercept request
|
||||
2. http.NewRequestWithContext()
|
||||
3. req.Header = hc.headers.Clone()
|
||||
4. setHeaders(req.Header, headersFromContext(ctx)) ← Can add headers
|
||||
5. if hc.auth != nil: hc.auth(req.Header) ← HTTPAuth hook
|
||||
6. resp, err := hc.client.Do(req) ← Standard HTTP transport
|
||||
7. json.Decoder(respBody).Decode(&resp) ← Can intercept response
|
||||
```
|
||||
|
||||
### Non-HTTP Path (Limited Middleware)
|
||||
```
|
||||
Client.send()
|
||||
→ Client.write()
|
||||
→ c.writeConn.writeJSON(ctx, msg, isError)
|
||||
→ jsonCodec.writeJSON()
|
||||
→ c.encode(v, isErrorResponse) ← Direct function call
|
||||
|
||||
In parallel:
|
||||
c.read()
|
||||
→ codec.readBatch()
|
||||
→ c.decode(&rawmsg) ← Direct function call
|
||||
```
|
||||
|
||||
## 9. Key Insights for Middleware Design
|
||||
|
||||
### 1. Transport Asymmetry
|
||||
- HTTP has good middleware hooks (HTTPAuth, context headers, http.Client)
|
||||
- WebSocket/IPC/Stdio have limited hooks (only custom dialer)
|
||||
- Middleware needs transport-specific implementation
|
||||
|
||||
### 2. Channel-Based Architecture
|
||||
- Non-HTTP uses Go channels for dispatch
|
||||
- Messages flow through defined channels (readOp, readErr, reqInit, reqSent)
|
||||
- Could intercept at channel boundaries
|
||||
|
||||
### 3. Two-Layer Codec System
|
||||
- Transport layer (httpConn, websocketCodec, jsonCodec) - implements ServerCodec
|
||||
- Handler layer (handler) - processes jsonrpcMessage structs
|
||||
- Middleware could target either layer
|
||||
|
||||
### 4. Request ID Tracking
|
||||
- Every request assigned unique ID
|
||||
- Can correlate requests/responses
|
||||
- Enables request tracing
|
||||
|
||||
### 5. Connection Lifecycle
|
||||
- Connections can be replaced (reconnect)
|
||||
- New handler created per connection (newClientConn)
|
||||
- Connection metadata available (peerInfo)
|
||||
|
||||
### 6. Error Handling
|
||||
- Transport errors: returned from send/write
|
||||
- RPC errors: returned in jsonError in response
|
||||
- Both should be intercepted separately
|
||||
|
||||
### 7. Subscription Complexity
|
||||
- Subscriptions require persistent connection (not HTTP)
|
||||
- Messages flowing to handler via readOp
|
||||
- Notifier pattern for server-side pushes
|
||||
- Need special handling for subscription responses
|
||||
|
||||
## 10. Architecture Summary
|
||||
|
||||
```
|
||||
┌─────────────────────────────────────────────────────────────┐
|
||||
│ Application Code │
|
||||
│ client.CallContext() / client.Subscribe() / etc. │
|
||||
└────────────────────────┬────────────────────────────────────┘
|
||||
│
|
||||
┌───────────────┼───────────────┐
|
||||
│ │ │
|
||||
▼ ▼ ▼
|
||||
┌────────┐ ┌──────────┐ ┌──────────┐
|
||||
│ HTTP │ │WebSocket │ │ IPC │
|
||||
│Handler │ │ Codec │ │ Codec │
|
||||
└────┬───┘ └────┬─────┘ └────┬─────┘
|
||||
│ │ │
|
||||
▼ ▼ ▼
|
||||
┌────────┐ ┌──────────┐ ┌──────────┐
|
||||
│httpConn│ │websocket │ │ jsonCodec│
|
||||
│ │ │ Codec │ │ │
|
||||
└────┬───┘ └────┬─────┘ └────┬─────┘
|
||||
│ │ │
|
||||
[sendHTTP] [Dispatch Loop] [Dispatch Loop]
|
||||
│ │ │
|
||||
▼ ▼ ▼
|
||||
[HTTP Req] [Channel Send] [Channel Send]
|
||||
│ │ │
|
||||
│ └─────┬─────────┘
|
||||
│ │
|
||||
└───────────┬───────┘
|
||||
│
|
||||
[Network I/O]
|
||||
│
|
||||
┌───────────┴────────────┐
|
||||
▼ ▼
|
||||
[RPC Server] [Other Clients]
|
||||
```
|
||||
|
||||
This architecture shows that middleware injection points exist at:
|
||||
- Application layer (wrapping Client)
|
||||
- HTTP layer (custom client, headers, auth)
|
||||
- Transport layer (custom dialer for WS)
|
||||
- Handler layer (if we extend handler)
|
||||
- Channel layer (if we intercept dispatch channels)
|
||||
BIN
graphql/Screenshot 2025-11-26 at 13.52.56.png
Normal file
BIN
graphql/Screenshot 2025-11-26 at 13.52.56.png
Normal file
Binary file not shown.
|
After Width: | Height: | Size: 94 KiB |
464
middleware_injection_strategies.md
Normal file
464
middleware_injection_strategies.md
Normal file
|
|
@ -0,0 +1,464 @@
|
|||
# RPC Middleware Injection Strategies for Go-Ethereum
|
||||
|
||||
## Overview
|
||||
This document outlines practical strategies for implementing middleware in the go-ethereum RPC client, considering the current architecture and limitations.
|
||||
|
||||
## Strategy 1: Wrapper Client (Recommended for Application-Level)
|
||||
|
||||
### Approach
|
||||
Wrap the native Client with a custom struct that intercepts all method calls.
|
||||
|
||||
### Advantages
|
||||
- Non-invasive (doesn't modify go-ethereum code)
|
||||
- Works for all transports equally
|
||||
- Can add logging, metrics, request/response transformation
|
||||
- Easy to test and compose multiple middlewares
|
||||
|
||||
### Implementation Pattern
|
||||
```go
|
||||
type MiddlewareClient struct {
|
||||
client *rpc.Client
|
||||
middlewares []Middleware
|
||||
}
|
||||
|
||||
type Middleware interface {
|
||||
BeforeCall(ctx context.Context, method string, args ...interface{}) error
|
||||
AfterCall(ctx context.Context, method string, result interface{}, err error) error
|
||||
OnSubscription(ctx context.Context, namespace string, channel interface{}) error
|
||||
}
|
||||
|
||||
func (mc *MiddlewareClient) CallContext(ctx context.Context, result interface{},
|
||||
method string, args ...interface{}) error {
|
||||
|
||||
// Before hooks
|
||||
for _, m := range mc.middlewares {
|
||||
if err := m.BeforeCall(ctx, method, args...); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
// Call
|
||||
err := mc.client.CallContext(ctx, result, method, args...)
|
||||
|
||||
// After hooks
|
||||
for _, m := range mc.middlewares {
|
||||
if hookErr := m.AfterCall(ctx, method, result, err); hookErr != nil {
|
||||
if err == nil {
|
||||
err = hookErr
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return err
|
||||
}
|
||||
```
|
||||
|
||||
### Use Cases
|
||||
- Request logging/tracing
|
||||
- Retry logic
|
||||
- Rate limiting
|
||||
- Authentication token refresh
|
||||
- Request/response transformation
|
||||
|
||||
---
|
||||
|
||||
## Strategy 2: HTTP-Specific Middleware (Best for HTTP Transport)
|
||||
|
||||
### Approach
|
||||
Use http.RoundTripper wrapping when creating HTTP client.
|
||||
|
||||
### Advantages
|
||||
- Transparent to go-ethereum code
|
||||
- Full control over HTTP layer
|
||||
- Can intercept headers, status codes, body
|
||||
- Leverage standard Go HTTP middleware ecosystem
|
||||
|
||||
### Implementation Pattern
|
||||
```go
|
||||
type RoundTripperMiddleware struct {
|
||||
next http.RoundTripper
|
||||
middlewares []HTTPMiddleware
|
||||
}
|
||||
|
||||
type HTTPMiddleware interface {
|
||||
BeforeRequest(req *http.Request) error
|
||||
AfterResponse(resp *http.Response) error
|
||||
}
|
||||
|
||||
func (rtm *RoundTripperMiddleware) RoundTrip(req *http.Request) (*http.Response, error) {
|
||||
// Before hooks
|
||||
for _, m := range rtm.middlewares {
|
||||
if err := m.BeforeRequest(req); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
|
||||
// Call
|
||||
resp, err := rtm.next.RoundTrip(req)
|
||||
|
||||
// After hooks
|
||||
if resp != nil {
|
||||
for _, m := range rtm.middlewares {
|
||||
if hookErr := m.AfterResponse(resp); hookErr != nil {
|
||||
if err == nil {
|
||||
err = hookErr
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return resp, err
|
||||
}
|
||||
|
||||
// Usage
|
||||
func NewHTTPClientWithMiddleware(middlewares ...HTTPMiddleware) *http.Client {
|
||||
base := &http.Client{}
|
||||
rt := &RoundTripperMiddleware{
|
||||
next: base.Transport,
|
||||
middlewares: middlewares,
|
||||
}
|
||||
base.Transport = rt
|
||||
return base
|
||||
}
|
||||
|
||||
// Create RPC client with middleware
|
||||
httpClient := NewHTTPClientWithMiddleware(
|
||||
&LoggingMiddleware{},
|
||||
&RetryMiddleware{},
|
||||
)
|
||||
rpcClient, _ := rpc.DialOptions(ctx, "http://localhost:8545",
|
||||
rpc.WithHTTPClient(httpClient),
|
||||
)
|
||||
```
|
||||
|
||||
### Use Cases
|
||||
- HTTP-specific logging
|
||||
- Response time measurement
|
||||
- Status code handling
|
||||
- Header inspection/modification
|
||||
- Cookie handling
|
||||
- Compression handling
|
||||
|
||||
---
|
||||
|
||||
## Strategy 3: HTTPAuth Hook (Built-in, Limited)
|
||||
|
||||
### Approach
|
||||
Use existing WithHTTPAuth option to add authentication headers and basic logging.
|
||||
|
||||
### Advantages
|
||||
- Built into go-ethereum
|
||||
- No additional dependencies
|
||||
- Applied per-request
|
||||
|
||||
### Limitations
|
||||
- Only for HTTP
|
||||
- Only manipulates headers
|
||||
- No response interception
|
||||
- No error handling
|
||||
|
||||
### Implementation Pattern
|
||||
```go
|
||||
type AuthMiddleware struct {
|
||||
token string
|
||||
logger Logger
|
||||
}
|
||||
|
||||
func (am *AuthMiddleware) Authenticate(h http.Header) error {
|
||||
// Log the request
|
||||
am.logger.Debug("auth middleware: adding token")
|
||||
|
||||
// Add auth header
|
||||
h.Set("Authorization", "Bearer " + am.token)
|
||||
return nil
|
||||
}
|
||||
|
||||
// Usage
|
||||
rpcClient, _ := rpc.DialOptions(ctx, "http://localhost:8545",
|
||||
rpc.WithHTTPAuth(authMiddleware.Authenticate),
|
||||
)
|
||||
```
|
||||
|
||||
### Use Cases
|
||||
- Token/API key injection
|
||||
- Basic auth setup
|
||||
- Header logging
|
||||
|
||||
---
|
||||
|
||||
## Strategy 4: Context-Based Header Injection
|
||||
|
||||
### Approach
|
||||
Use NewContextWithHeaders to inject per-request headers without modifying client config.
|
||||
|
||||
### Advantages
|
||||
- Per-request granularity
|
||||
- No global state
|
||||
- Works with existing client
|
||||
- Can be combined with other approaches
|
||||
|
||||
### Implementation Pattern
|
||||
```go
|
||||
func CallWithTraceID(client *rpc.Client, ctx context.Context,
|
||||
result interface{}, method string, args ...interface{}) error {
|
||||
|
||||
traceID := generateTraceID()
|
||||
headers := http.Header{
|
||||
"X-Trace-ID": []string{traceID},
|
||||
"X-Request-ID": []string{generateRequestID()},
|
||||
}
|
||||
|
||||
ctx = rpc.NewContextWithHeaders(ctx, headers)
|
||||
return client.CallContext(ctx, result, method, args...)
|
||||
}
|
||||
```
|
||||
|
||||
### Use Cases
|
||||
- Request ID/Trace ID injection
|
||||
- Per-request metadata
|
||||
- Dynamic header injection
|
||||
- Request correlation
|
||||
|
||||
---
|
||||
|
||||
## Strategy 5: Message-Level Interception (Advanced)
|
||||
|
||||
### Approach
|
||||
Wrap ServerCodec interface to intercept messages at codec level.
|
||||
|
||||
### Advantages
|
||||
- Transport-agnostic (works for HTTP, WS, IPC)
|
||||
- Full access to jsonrpcMessage
|
||||
- Can transform requests/responses
|
||||
- Enables comprehensive logging
|
||||
|
||||
### Challenges
|
||||
- Requires deeper integration
|
||||
- Complex state management
|
||||
- Need to handle all codec types
|
||||
- May affect performance
|
||||
|
||||
### Implementation Pattern
|
||||
```go
|
||||
// Wrapper for any ServerCodec
|
||||
type InterceptingCodec struct {
|
||||
codec rpc.ServerCodec
|
||||
interceptor MessageInterceptor
|
||||
}
|
||||
|
||||
type MessageInterceptor interface {
|
||||
OnReadMessage(msg *jsonrpcMessage) error
|
||||
OnWriteMessage(msg *jsonrpcMessage) error
|
||||
}
|
||||
|
||||
func (ic *InterceptingCodec) readBatch() ([]*jsonrpcMessage, bool, error) {
|
||||
msgs, batch, err := ic.codec.readBatch()
|
||||
if err == nil && ic.interceptor != nil {
|
||||
for _, msg := range msgs {
|
||||
if ierr := ic.interceptor.OnReadMessage(msg); ierr != nil {
|
||||
return nil, false, ierr
|
||||
}
|
||||
}
|
||||
}
|
||||
return msgs, batch, err
|
||||
}
|
||||
|
||||
func (ic *InterceptingCodec) writeJSON(ctx context.Context, v interface{}, isError bool) error {
|
||||
// Would need to intercept at this level
|
||||
if msg, ok := v.(*jsonrpcMessage); ok && ic.interceptor != nil {
|
||||
if ierr := ic.interceptor.OnWriteMessage(msg); ierr != nil {
|
||||
return ierr
|
||||
}
|
||||
}
|
||||
return ic.codec.writeJSON(ctx, v, isError)
|
||||
}
|
||||
|
||||
// Would need to implement other ServerCodec methods...
|
||||
```
|
||||
|
||||
### Use Cases
|
||||
- Request/response logging with full message body
|
||||
- Message transformation/validation
|
||||
- Performance metrics
|
||||
- Rate limiting at RPC level
|
||||
|
||||
---
|
||||
|
||||
## Strategy 6: Dispatch Channel Interception (Advanced)
|
||||
|
||||
### Approach
|
||||
Intercept at the channel layer in the Client's dispatch loop.
|
||||
|
||||
### Advantages
|
||||
- Access to internal state (request IDs, handlers)
|
||||
- Can correlate requests and responses
|
||||
- Pure Go concurrency primitives
|
||||
|
||||
### Challenges
|
||||
- Very tightly coupled to implementation
|
||||
- Breaks encapsulation
|
||||
- Complex to implement correctly
|
||||
- Difficult to maintain across versions
|
||||
|
||||
### Not Recommended
|
||||
This approach is too invasive and fragile. Prefer Strategies 1-5.
|
||||
|
||||
---
|
||||
|
||||
## Strategy 7: WebSocket-Specific Handlers
|
||||
|
||||
### Approach
|
||||
For WebSocket connections, create wrapper around websocket.Dialer to customize connection behavior.
|
||||
|
||||
### Advantages
|
||||
- WS-specific features possible
|
||||
- Connection-level control
|
||||
- Can inspect handshake
|
||||
|
||||
### Implementation Pattern
|
||||
```go
|
||||
type DialerWithMiddleware struct {
|
||||
base *websocket.Dialer
|
||||
middlewares []DialerMiddleware
|
||||
}
|
||||
|
||||
type DialerMiddleware interface {
|
||||
BeforeDial(ctx context.Context, url string) error
|
||||
AfterDial(conn *websocket.Conn) error
|
||||
}
|
||||
|
||||
func (dwm *DialerWithMiddleware) DialContext(ctx context.Context,
|
||||
urlStr string, requestHeader http.Header) (*websocket.Conn, *http.Response, error) {
|
||||
|
||||
// Before hooks
|
||||
for _, m := range dwm.middlewares {
|
||||
if err := m.BeforeDial(ctx, urlStr); err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
}
|
||||
|
||||
// Dial
|
||||
conn, resp, err := dwm.base.DialContext(ctx, urlStr, requestHeader)
|
||||
|
||||
// After hooks
|
||||
if err == nil && conn != nil {
|
||||
for _, m := range dwm.middlewares {
|
||||
if hookErr := m.AfterDial(conn); hookErr != nil {
|
||||
conn.Close()
|
||||
return nil, resp, hookErr
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return conn, resp, err
|
||||
}
|
||||
|
||||
// Usage
|
||||
dialer := &DialerWithMiddleware{
|
||||
base: &websocket.Dialer{...},
|
||||
middlewares: []DialerMiddleware{
|
||||
&WSLoggingMiddleware{},
|
||||
},
|
||||
}
|
||||
|
||||
rpcClient, _ := rpc.DialOptions(ctx, "ws://localhost:8545",
|
||||
rpc.WithWebsocketDialer(*dialer.base), // Note: can't pass wrapper directly
|
||||
)
|
||||
```
|
||||
|
||||
### Limitation
|
||||
WithWebsocketDialer expects a websocket.Dialer directly, not a wrapper, so this approach has limited applicability without modifying client_opt.go.
|
||||
|
||||
---
|
||||
|
||||
## Strategy 8: Subscription Wrapper
|
||||
|
||||
### Approach
|
||||
Wrap the ClientSubscription returned from Subscribe to intercept events.
|
||||
|
||||
### Advantages
|
||||
- Subscription-specific handling
|
||||
- Non-invasive
|
||||
- Works with existing client
|
||||
|
||||
### Implementation Pattern
|
||||
```go
|
||||
type SubscriptionMiddleware struct {
|
||||
sub *rpc.ClientSubscription
|
||||
ch interface{}
|
||||
middlewares []SubscriptionMiddleware
|
||||
}
|
||||
|
||||
type SubscriptionMiddleware interface {
|
||||
OnEvent(ev interface{}) error
|
||||
OnError(err error) error
|
||||
}
|
||||
|
||||
func WrapSubscription(sub *rpc.ClientSubscription,
|
||||
ch interface{}, middlewares ...SubscriptionMiddleware) *SubscriptionMiddleware {
|
||||
|
||||
return &SubscriptionMiddleware{
|
||||
sub: sub,
|
||||
ch: ch,
|
||||
middlewares: middlewares,
|
||||
}
|
||||
}
|
||||
|
||||
// Would need to read from sub.C and apply middlewares
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Recommended Strategy Selection
|
||||
|
||||
### For HTTP-Only Applications
|
||||
1. Use Strategy 2 (HTTP RoundTripper) for transport-level middleware
|
||||
2. Use Strategy 1 (Wrapper Client) for application-level logging/transformation
|
||||
3. Use HTTPAuth for simple authentication
|
||||
|
||||
### For WebSocket Applications
|
||||
1. Use Strategy 1 (Wrapper Client) for application-level concerns
|
||||
2. Use Strategy 2 if also supporting HTTP
|
||||
3. Use Context headers for per-request metadata
|
||||
|
||||
### For Comprehensive Tracing/Metrics
|
||||
1. Combine Strategy 1 (Wrapper Client) with Strategy 2 (RoundTripper)
|
||||
2. Use Strategy 4 (Context Headers) for correlation IDs
|
||||
3. Avoid Strategy 5 unless absolutely necessary
|
||||
|
||||
### For Advanced Use Cases
|
||||
1. Implement custom strategies based on application requirements
|
||||
2. Consider whether modifying go-ethereum is acceptable for your use case
|
||||
3. Always prefer non-invasive wrappers over modifying the library
|
||||
|
||||
---
|
||||
|
||||
## Implementation Checklist
|
||||
|
||||
When implementing middleware for go-ethereum RPC client:
|
||||
|
||||
- [ ] Identify which transports need to be supported (HTTP, WS, IPC, Stdio)
|
||||
- [ ] Determine middleware scope (connection-level, request-level, message-level)
|
||||
- [ ] Choose non-invasive approach when possible
|
||||
- [ ] Handle context cancellation properly
|
||||
- [ ] Implement error handling and propagation
|
||||
- [ ] Consider performance impact (avoid allocations in hot path)
|
||||
- [ ] Add tests for middleware behavior
|
||||
- [ ] Document expected behavior and limitations
|
||||
- [ ] Plan for go-ethereum version upgrades
|
||||
- [ ] Consider thread safety for concurrent calls
|
||||
- [ ] Handle subscription/notification middleware separately if needed
|
||||
- [ ] Implement proper logging without noise
|
||||
|
||||
---
|
||||
|
||||
## Anti-Patterns to Avoid
|
||||
|
||||
1. **Blocking in Middleware**: Don't block indefinitely; respect context timeouts
|
||||
2. **Global State**: Avoid global variables; use dependency injection
|
||||
3. **Ignoring Errors**: Always propagate errors from hooks
|
||||
4. **Transport Assumptions**: Don't assume HTTP if WS/IPC might be used
|
||||
5. **Tight Coupling**: Don't depend on private go-ethereum fields
|
||||
6. **Synchronous I/O**: Avoid synchronous network calls in hot path
|
||||
7. **Memory Leaks**: Always clean up goroutines and channels
|
||||
8. **Silent Failures**: Log all middleware errors, don't swallow them
|
||||
|
|
@ -90,6 +90,10 @@ type Client struct {
|
|||
batchItemLimit int
|
||||
batchResponseMaxSize int
|
||||
|
||||
// interceptors
|
||||
requestInterceptors []RequestInterceptor
|
||||
responseInterceptors []ResponseInterceptor
|
||||
|
||||
// writeConn is used for writing to the connection on the caller's goroutine. It should
|
||||
// only be accessed outside of dispatch, with the write lock held. The write lock is
|
||||
// taken by sending on reqInit and released by sending on reqSent.
|
||||
|
|
@ -248,6 +252,8 @@ func initClient(conn ServerCodec, services *serviceRegistry, cfg *clientConfig)
|
|||
idgen: cfg.idgen,
|
||||
batchItemLimit: cfg.batchItemLimit,
|
||||
batchResponseMaxSize: cfg.batchResponseLimit,
|
||||
requestInterceptors: cfg.requestInterceptors,
|
||||
responseInterceptors: cfg.responseInterceptors,
|
||||
writeConn: conn,
|
||||
close: make(chan struct{}),
|
||||
closing: make(chan struct{}),
|
||||
|
|
@ -339,6 +345,12 @@ func (c *Client) CallContext(ctx context.Context, result interface{}, method str
|
|||
if result != nil && reflect.TypeOf(result).Kind() != reflect.Ptr {
|
||||
return fmt.Errorf("call result parameter must be pointer or nil interface: %v", result)
|
||||
}
|
||||
|
||||
// Call request interceptors before sending.
|
||||
if err := c.callRequestInterceptors(ctx, method, args); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
msg, err := c.newMessage(method, args...)
|
||||
if err != nil {
|
||||
return err
|
||||
|
|
@ -354,25 +366,26 @@ func (c *Client) CallContext(ctx context.Context, result interface{}, method str
|
|||
err = c.send(ctx, op, msg)
|
||||
}
|
||||
if err != nil {
|
||||
return err
|
||||
return c.callResponseInterceptors(ctx, method, err)
|
||||
}
|
||||
|
||||
// dispatch has accepted the request and will close the channel when it quits.
|
||||
batchresp, err := op.wait(ctx, c)
|
||||
if err != nil {
|
||||
return err
|
||||
return c.callResponseInterceptors(ctx, method, err)
|
||||
}
|
||||
resp := batchresp[0]
|
||||
switch {
|
||||
case resp.Error != nil:
|
||||
return resp.Error
|
||||
return c.callResponseInterceptors(ctx, method, resp.Error)
|
||||
case len(resp.Result) == 0:
|
||||
return ErrNoResult
|
||||
return c.callResponseInterceptors(ctx, method, ErrNoResult)
|
||||
default:
|
||||
if result == nil {
|
||||
return nil
|
||||
return c.callResponseInterceptors(ctx, method, nil)
|
||||
}
|
||||
return json.Unmarshal(resp.Result, result)
|
||||
err = json.Unmarshal(resp.Result, result)
|
||||
return c.callResponseInterceptors(ctx, method, err)
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -398,6 +411,11 @@ func (c *Client) BatchCall(b []BatchElem) error {
|
|||
//
|
||||
// Note that batch calls may not be executed atomically on the server side.
|
||||
func (c *Client) BatchCallContext(ctx context.Context, b []BatchElem) error {
|
||||
// Call request interceptors for the batch (method="" for batch).
|
||||
if err := c.callRequestInterceptors(ctx, "", nil); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
var (
|
||||
msgs = make([]*jsonrpcMessage, len(b))
|
||||
byID = make(map[string]int, len(b))
|
||||
|
|
@ -423,12 +441,12 @@ func (c *Client) BatchCallContext(ctx context.Context, b []BatchElem) error {
|
|||
err = c.send(ctx, op, msgs)
|
||||
}
|
||||
if err != nil {
|
||||
return err
|
||||
return c.callResponseInterceptors(ctx, "", err)
|
||||
}
|
||||
|
||||
batchresp, err := op.wait(ctx, c)
|
||||
if err != nil {
|
||||
return err
|
||||
return c.callResponseInterceptors(ctx, "", err)
|
||||
}
|
||||
|
||||
// Wait for all responses to come back.
|
||||
|
|
@ -464,11 +482,18 @@ func (c *Client) BatchCallContext(ctx context.Context, b []BatchElem) error {
|
|||
elem.Error = ErrMissingBatchResponse
|
||||
}
|
||||
|
||||
return err
|
||||
// Call response interceptors for the batch (method="" for batch).
|
||||
// err here is the I/O error, not per-item errors (those are in BatchElem.Error).
|
||||
return c.callResponseInterceptors(ctx, "", err)
|
||||
}
|
||||
|
||||
// Notify sends a notification, i.e. a method call that doesn't expect a response.
|
||||
func (c *Client) Notify(ctx context.Context, method string, args ...interface{}) error {
|
||||
// Call request interceptors before sending notification.
|
||||
if err := c.callRequestInterceptors(ctx, method, args); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
op := new(requestOp)
|
||||
msg, err := c.newMessage(method, args...)
|
||||
if err != nil {
|
||||
|
|
@ -518,7 +543,14 @@ func (c *Client) Subscribe(ctx context.Context, namespace string, channel interf
|
|||
return nil, ErrNotificationsUnsupported
|
||||
}
|
||||
|
||||
msg, err := c.newMessage(namespace+subscribeMethodSuffix, args...)
|
||||
method := namespace + subscribeMethodSuffix
|
||||
|
||||
// Call request interceptors before sending subscription request.
|
||||
if err := c.callRequestInterceptors(ctx, method, args); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
msg, err := c.newMessage(method, args...)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
|
@ -531,12 +563,12 @@ func (c *Client) Subscribe(ctx context.Context, namespace string, channel interf
|
|||
// Send the subscription request.
|
||||
// The arrival and validity of the response is signaled on sub.quit.
|
||||
if err := c.send(ctx, op, msg); err != nil {
|
||||
return nil, err
|
||||
return nil, c.callResponseInterceptors(ctx, method, err)
|
||||
}
|
||||
if _, err := op.wait(ctx, c); err != nil {
|
||||
return nil, err
|
||||
return nil, c.callResponseInterceptors(ctx, method, err)
|
||||
}
|
||||
return op.sub, nil
|
||||
return op.sub, c.callResponseInterceptors(ctx, method, nil)
|
||||
}
|
||||
|
||||
// SupportsSubscriptions reports whether subscriptions are supported by the client
|
||||
|
|
@ -616,6 +648,26 @@ func (c *Client) reconnect(ctx context.Context) error {
|
|||
}
|
||||
}
|
||||
|
||||
// callRequestInterceptors calls all request interceptors in order.
|
||||
// Returns the first error encountered, or nil if all succeed.
|
||||
func (c *Client) callRequestInterceptors(ctx context.Context, method string, args []interface{}) error {
|
||||
for _, interceptor := range c.requestInterceptors {
|
||||
if err := interceptor(ctx, method, args); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// callResponseInterceptors calls all response interceptors in order.
|
||||
// Each interceptor receives the error from the previous one.
|
||||
func (c *Client) callResponseInterceptors(ctx context.Context, method string, err error) error {
|
||||
for _, interceptor := range c.responseInterceptors {
|
||||
err = interceptor(ctx, method, err)
|
||||
}
|
||||
return err
|
||||
}
|
||||
|
||||
// dispatch is the main loop of the client.
|
||||
// It sends read messages to waiting calls to Call and BatchCall
|
||||
// and subscription notifications to registered subscriptions.
|
||||
|
|
|
|||
|
|
@ -87,3 +87,54 @@ func subscribeBlocks(client *rpc.Client, subch chan Block) {
|
|||
// the connection.
|
||||
fmt.Println("connection lost: ", <-sub.Err())
|
||||
}
|
||||
|
||||
// This example demonstrates how to use request interceptors for rate limiting
|
||||
// and response interceptors for logging errors.
|
||||
func ExampleWithRequestInterceptor_rateLimiting() {
|
||||
// Create a simple rate limiter (allows 10 requests per second).
|
||||
// In production, you might use golang.org/x/time/rate or another package.
|
||||
limiter := make(chan struct{}, 10)
|
||||
for i := 0; i < 10; i++ {
|
||||
limiter <- struct{}{}
|
||||
}
|
||||
go func() {
|
||||
ticker := time.NewTicker(time.Second / 10)
|
||||
defer ticker.Stop()
|
||||
for range ticker.C {
|
||||
select {
|
||||
case limiter <- struct{}{}:
|
||||
default:
|
||||
}
|
||||
}
|
||||
}()
|
||||
|
||||
// Create client with rate limiting interceptor.
|
||||
client, err := rpc.DialOptions(
|
||||
context.Background(),
|
||||
"ws://127.0.0.1:8545",
|
||||
rpc.WithRequestInterceptor(func(ctx context.Context, method string, args []interface{}) error {
|
||||
// Wait for rate limit token (or until context is cancelled).
|
||||
select {
|
||||
case <-limiter:
|
||||
return nil
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
}
|
||||
}),
|
||||
rpc.WithResponseInterceptor(func(ctx context.Context, method string, err error) error {
|
||||
// Log any errors.
|
||||
if err != nil {
|
||||
fmt.Printf("RPC error for method %s: %v\n", method, err)
|
||||
}
|
||||
return err
|
||||
}),
|
||||
)
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
defer client.Close()
|
||||
|
||||
// All calls through this client will now be rate limited.
|
||||
var result string
|
||||
_ = client.CallContext(context.Background(), &result, "eth_blockNumber")
|
||||
}
|
||||
|
|
|
|||
|
|
@ -17,6 +17,7 @@
|
|||
package rpc
|
||||
|
||||
import (
|
||||
"context"
|
||||
"net/http"
|
||||
|
||||
"github.com/gorilla/websocket"
|
||||
|
|
@ -41,6 +42,10 @@ type clientConfig struct {
|
|||
idgen func() ID
|
||||
batchItemLimit int
|
||||
batchResponseLimit int
|
||||
|
||||
// Interceptors
|
||||
requestInterceptors []RequestInterceptor
|
||||
responseInterceptors []ResponseInterceptor
|
||||
}
|
||||
|
||||
func (cfg *clientConfig) initHeaders() {
|
||||
|
|
@ -142,3 +147,100 @@ func WithBatchResponseSizeLimit(sizeLimit int) ClientOption {
|
|||
cfg.batchResponseLimit = sizeLimit
|
||||
})
|
||||
}
|
||||
|
||||
// RequestInterceptor is called before sending RPC requests.
|
||||
//
|
||||
// The interceptor is invoked with the request context, method name, and arguments.
|
||||
// For batch requests, method is empty string and args is nil; the interceptor runs
|
||||
// once per batch, not per item.
|
||||
//
|
||||
// Request interceptors run in order. If an interceptor returns an error, the request
|
||||
// is not sent and the error is returned to the caller immediately.
|
||||
//
|
||||
// The context passed to the interceptor is the same context passed to CallContext.
|
||||
// Interceptors can use the context for rate limiting (e.g., limiter.Wait(ctx)) or
|
||||
// checking cancellation.
|
||||
//
|
||||
// IMPORTANT: Interceptors MUST NOT modify the args slice. Doing so results in
|
||||
// undefined behavior and may break retries or reconnections.
|
||||
type RequestInterceptor func(ctx context.Context, method string, args []interface{}) error
|
||||
|
||||
// ResponseInterceptor is called after receiving RPC responses.
|
||||
//
|
||||
// The interceptor is invoked with the request context, method name, and the final error
|
||||
// (which may be nil on success, or an I/O error, RPC error, or unmarshal error).
|
||||
//
|
||||
// For batch requests, method is empty string and the interceptor runs once per batch.
|
||||
// The error represents the transport-level error (usually nil if the batch request
|
||||
// succeeded). Per-item RPC errors within the batch are not passed to interceptors;
|
||||
// they remain in BatchElem.Error and should be checked by the caller.
|
||||
//
|
||||
// Response interceptors run in order. Each interceptor receives the error returned by
|
||||
// the previous interceptor (or the original error for the first interceptor).
|
||||
// The error returned by the last interceptor is returned to the caller.
|
||||
//
|
||||
// Interceptors can suppress errors by returning nil, wrap errors for additional context,
|
||||
// or return a different error entirely.
|
||||
type ResponseInterceptor func(ctx context.Context, method string, err error) error
|
||||
|
||||
// WithRequestInterceptor adds a request interceptor to the client.
|
||||
//
|
||||
// Request interceptors are called before sending RPC requests. Multiple interceptors
|
||||
// can be added and will run in the order they were added. If any interceptor returns
|
||||
// an error, the request is not sent.
|
||||
//
|
||||
// Example - rate limiting:
|
||||
//
|
||||
// limiter := rate.NewLimiter(rate.Every(time.Second), 10)
|
||||
// client, _ := rpc.DialOptions(ctx, url,
|
||||
// rpc.WithRequestInterceptor(func(ctx context.Context, method string, args []interface{}) error {
|
||||
// return limiter.Wait(ctx)
|
||||
// }),
|
||||
// )
|
||||
//
|
||||
// Example - logging:
|
||||
//
|
||||
// client, _ := rpc.DialOptions(ctx, url,
|
||||
// rpc.WithRequestInterceptor(func(ctx context.Context, method string, args []interface{}) error {
|
||||
// log.Printf("RPC call: %s", method)
|
||||
// return nil
|
||||
// }),
|
||||
// )
|
||||
func WithRequestInterceptor(interceptor RequestInterceptor) ClientOption {
|
||||
return optionFunc(func(cfg *clientConfig) {
|
||||
cfg.requestInterceptors = append(cfg.requestInterceptors, interceptor)
|
||||
})
|
||||
}
|
||||
|
||||
// WithResponseInterceptor adds a response interceptor to the client.
|
||||
//
|
||||
// Response interceptors are called after receiving RPC responses. Multiple interceptors
|
||||
// can be added and will run in the order they were added. Each interceptor receives
|
||||
// the error from the previous interceptor.
|
||||
//
|
||||
// Example - error logging:
|
||||
//
|
||||
// client, _ := rpc.DialOptions(ctx, url,
|
||||
// rpc.WithResponseInterceptor(func(ctx context.Context, method string, err error) error {
|
||||
// if err != nil {
|
||||
// log.Printf("RPC error for %s: %v", method, err)
|
||||
// }
|
||||
// return err
|
||||
// }),
|
||||
// )
|
||||
//
|
||||
// For batch requests, if you need per-item error observability, check BatchElem.Error
|
||||
// after the call returns:
|
||||
//
|
||||
// batch := []rpc.BatchElem{...}
|
||||
// err := client.BatchCallContext(ctx, batch)
|
||||
// for i, elem := range batch {
|
||||
// if elem.Error != nil {
|
||||
// log.Printf("Batch[%d] %s failed: %v", i, elem.Method, elem.Error)
|
||||
// }
|
||||
// }
|
||||
func WithResponseInterceptor(interceptor ResponseInterceptor) ClientOption {
|
||||
return optionFunc(func(cfg *clientConfig) {
|
||||
cfg.responseInterceptors = append(cfg.responseInterceptors, interceptor)
|
||||
})
|
||||
}
|
||||
|
|
|
|||
|
|
@ -14,28 +14,414 @@
|
|||
// You should have received a copy of the GNU Lesser General Public License
|
||||
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
|
||||
|
||||
package rpc_test
|
||||
package rpc
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"time"
|
||||
|
||||
"github.com/ethereum/go-ethereum/rpc"
|
||||
"net/http/httptest"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// This example configures a HTTP-based RPC client with two options - one setting the
|
||||
// overall request timeout, the other adding a custom HTTP header to all requests.
|
||||
func ExampleDialOptions() {
|
||||
tokenHeader := rpc.WithHeader("x-token", "foo")
|
||||
httpClient := rpc.WithHTTPClient(&http.Client{
|
||||
Timeout: 10 * time.Second,
|
||||
})
|
||||
func TestRequestInterceptor(t *testing.T) {
|
||||
// Setup a test server
|
||||
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
w.Header().Set("content-type", "application/json")
|
||||
fmt.Fprintln(w, `{"jsonrpc":"2.0","id":1,"result":"0x1"}`)
|
||||
}))
|
||||
defer server.Close()
|
||||
|
||||
ctx := context.Background()
|
||||
c, err := rpc.DialOptions(ctx, "http://rpc.example.com", httpClient, tokenHeader)
|
||||
// Test that request interceptor is called
|
||||
var called bool
|
||||
var capturedMethod string
|
||||
client, err := DialOptions(context.Background(), server.URL,
|
||||
WithRequestInterceptor(func(ctx context.Context, method string, args []interface{}) error {
|
||||
called = true
|
||||
capturedMethod = method
|
||||
return nil
|
||||
}),
|
||||
)
|
||||
if err != nil {
|
||||
panic(err)
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer client.Close()
|
||||
|
||||
var result string
|
||||
err = client.CallContext(context.Background(), &result, "test_method")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
if !called {
|
||||
t.Error("request interceptor was not called")
|
||||
}
|
||||
if capturedMethod != "test_method" {
|
||||
t.Errorf("interceptor got method %q, want %q", capturedMethod, "test_method")
|
||||
}
|
||||
}
|
||||
|
||||
func TestRequestInterceptorBlocks(t *testing.T) {
|
||||
// Setup a test server that should never be hit
|
||||
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
t.Error("server should not have been called")
|
||||
}))
|
||||
defer server.Close()
|
||||
|
||||
// Test that request interceptor can block the request
|
||||
blockErr := errors.New("blocked by interceptor")
|
||||
client, err := DialOptions(context.Background(), server.URL,
|
||||
WithRequestInterceptor(func(ctx context.Context, method string, args []interface{}) error {
|
||||
return blockErr
|
||||
}),
|
||||
)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer client.Close()
|
||||
|
||||
var result string
|
||||
err = client.CallContext(context.Background(), &result, "test_method")
|
||||
if err != blockErr {
|
||||
t.Errorf("got error %v, want %v", err, blockErr)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRequestInterceptorChaining(t *testing.T) {
|
||||
// Setup a test server
|
||||
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
w.Header().Set("content-type", "application/json")
|
||||
fmt.Fprintln(w, `{"jsonrpc":"2.0","id":1,"result":"0x1"}`)
|
||||
}))
|
||||
defer server.Close()
|
||||
|
||||
// Test that multiple interceptors run in order
|
||||
var order []int
|
||||
client, err := DialOptions(context.Background(), server.URL,
|
||||
WithRequestInterceptor(func(ctx context.Context, method string, args []interface{}) error {
|
||||
order = append(order, 1)
|
||||
return nil
|
||||
}),
|
||||
WithRequestInterceptor(func(ctx context.Context, method string, args []interface{}) error {
|
||||
order = append(order, 2)
|
||||
return nil
|
||||
}),
|
||||
WithRequestInterceptor(func(ctx context.Context, method string, args []interface{}) error {
|
||||
order = append(order, 3)
|
||||
return nil
|
||||
}),
|
||||
)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer client.Close()
|
||||
|
||||
var result string
|
||||
err = client.CallContext(context.Background(), &result, "test_method")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
if len(order) != 3 || order[0] != 1 || order[1] != 2 || order[2] != 3 {
|
||||
t.Errorf("interceptors ran in wrong order: %v", order)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRequestInterceptorShortCircuit(t *testing.T) {
|
||||
// Setup a test server that should never be hit
|
||||
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
t.Error("server should not have been called")
|
||||
}))
|
||||
defer server.Close()
|
||||
|
||||
// Test that first error stops the chain
|
||||
blockErr := errors.New("blocked")
|
||||
var thirdCalled bool
|
||||
client, err := DialOptions(context.Background(), server.URL,
|
||||
WithRequestInterceptor(func(ctx context.Context, method string, args []interface{}) error {
|
||||
return nil
|
||||
}),
|
||||
WithRequestInterceptor(func(ctx context.Context, method string, args []interface{}) error {
|
||||
return blockErr
|
||||
}),
|
||||
WithRequestInterceptor(func(ctx context.Context, method string, args []interface{}) error {
|
||||
thirdCalled = true
|
||||
return nil
|
||||
}),
|
||||
)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer client.Close()
|
||||
|
||||
var result string
|
||||
err = client.CallContext(context.Background(), &result, "test_method")
|
||||
if err != blockErr {
|
||||
t.Errorf("got error %v, want %v", err, blockErr)
|
||||
}
|
||||
if thirdCalled {
|
||||
t.Error("third interceptor should not have been called")
|
||||
}
|
||||
}
|
||||
|
||||
func TestResponseInterceptor(t *testing.T) {
|
||||
// Setup a test server
|
||||
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
w.Header().Set("content-type", "application/json")
|
||||
fmt.Fprintln(w, `{"jsonrpc":"2.0","id":1,"result":"0x1"}`)
|
||||
}))
|
||||
defer server.Close()
|
||||
|
||||
// Test that response interceptor is called with nil error on success
|
||||
var called bool
|
||||
var capturedMethod string
|
||||
var capturedErr error
|
||||
client, err := DialOptions(context.Background(), server.URL,
|
||||
WithResponseInterceptor(func(ctx context.Context, method string, err error) error {
|
||||
called = true
|
||||
capturedMethod = method
|
||||
capturedErr = err
|
||||
return err
|
||||
}),
|
||||
)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer client.Close()
|
||||
|
||||
var result string
|
||||
err = client.CallContext(context.Background(), &result, "test_method")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
if !called {
|
||||
t.Error("response interceptor was not called")
|
||||
}
|
||||
if capturedMethod != "test_method" {
|
||||
t.Errorf("interceptor got method %q, want %q", capturedMethod, "test_method")
|
||||
}
|
||||
if capturedErr != nil {
|
||||
t.Errorf("interceptor got error %v, want nil", capturedErr)
|
||||
}
|
||||
}
|
||||
|
||||
func TestResponseInterceptorWithError(t *testing.T) {
|
||||
// Setup a test server that returns an error
|
||||
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
w.Header().Set("content-type", "application/json")
|
||||
fmt.Fprintln(w, `{"jsonrpc":"2.0","id":1,"error":{"code":-32000,"message":"test error"}}`)
|
||||
}))
|
||||
defer server.Close()
|
||||
|
||||
// Test that response interceptor receives the error
|
||||
var capturedErr error
|
||||
client, err := DialOptions(context.Background(), server.URL,
|
||||
WithResponseInterceptor(func(ctx context.Context, method string, err error) error {
|
||||
capturedErr = err
|
||||
return err
|
||||
}),
|
||||
)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer client.Close()
|
||||
|
||||
var result string
|
||||
err = client.CallContext(context.Background(), &result, "test_method")
|
||||
if err == nil {
|
||||
t.Fatal("expected error")
|
||||
}
|
||||
|
||||
if capturedErr == nil {
|
||||
t.Error("interceptor should have received error")
|
||||
}
|
||||
if capturedErr.Error() != "test error" {
|
||||
t.Errorf("interceptor got error %q, want %q", capturedErr.Error(), "test error")
|
||||
}
|
||||
}
|
||||
|
||||
func TestResponseInterceptorCanModifyError(t *testing.T) {
|
||||
// Setup a test server that returns an error
|
||||
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
w.Header().Set("content-type", "application/json")
|
||||
fmt.Fprintln(w, `{"jsonrpc":"2.0","id":1,"error":{"code":-32000,"message":"original error"}}`)
|
||||
}))
|
||||
defer server.Close()
|
||||
|
||||
// Test that response interceptor can wrap the error
|
||||
wrappedErr := errors.New("wrapped error")
|
||||
client, err := DialOptions(context.Background(), server.URL,
|
||||
WithResponseInterceptor(func(ctx context.Context, method string, err error) error {
|
||||
if err != nil {
|
||||
return wrappedErr
|
||||
}
|
||||
return err
|
||||
}),
|
||||
)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer client.Close()
|
||||
|
||||
var result string
|
||||
err = client.CallContext(context.Background(), &result, "test_method")
|
||||
if err != wrappedErr {
|
||||
t.Errorf("got error %v, want %v", err, wrappedErr)
|
||||
}
|
||||
}
|
||||
|
||||
func TestResponseInterceptorChaining(t *testing.T) {
|
||||
// Setup a test server
|
||||
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
w.Header().Set("content-type", "application/json")
|
||||
fmt.Fprintln(w, `{"jsonrpc":"2.0","id":1,"error":{"code":-32000,"message":"original"}}`)
|
||||
}))
|
||||
defer server.Close()
|
||||
|
||||
// Test that multiple response interceptors run in order and chain errors
|
||||
client, err := DialOptions(context.Background(), server.URL,
|
||||
WithResponseInterceptor(func(ctx context.Context, method string, err error) error {
|
||||
if err != nil {
|
||||
return fmt.Errorf("first: %w", err)
|
||||
}
|
||||
return err
|
||||
}),
|
||||
WithResponseInterceptor(func(ctx context.Context, method string, err error) error {
|
||||
if err != nil {
|
||||
return fmt.Errorf("second: %w", err)
|
||||
}
|
||||
return err
|
||||
}),
|
||||
)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer client.Close()
|
||||
|
||||
var result string
|
||||
err = client.CallContext(context.Background(), &result, "test_method")
|
||||
if err == nil {
|
||||
t.Fatal("expected error")
|
||||
}
|
||||
|
||||
// Check that error was wrapped by both interceptors
|
||||
errMsg := err.Error()
|
||||
if errMsg != "second: first: original" {
|
||||
t.Errorf("got error %q, expected chained wrapping", errMsg)
|
||||
}
|
||||
}
|
||||
|
||||
func TestBatchCallWithInterceptors(t *testing.T) {
|
||||
// Setup a test server
|
||||
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
w.Header().Set("content-type", "application/json")
|
||||
fmt.Fprintln(w, `[{"jsonrpc":"2.0","id":1,"result":"0x1"},{"jsonrpc":"2.0","id":2,"result":"0x2"}]`)
|
||||
}))
|
||||
defer server.Close()
|
||||
|
||||
// Test that interceptors are called for batch requests
|
||||
var reqCalled, respCalled bool
|
||||
var reqMethod, respMethod string
|
||||
client, err := DialOptions(context.Background(), server.URL,
|
||||
WithRequestInterceptor(func(ctx context.Context, method string, args []interface{}) error {
|
||||
reqCalled = true
|
||||
reqMethod = method
|
||||
return nil
|
||||
}),
|
||||
WithResponseInterceptor(func(ctx context.Context, method string, err error) error {
|
||||
respCalled = true
|
||||
respMethod = method
|
||||
return err
|
||||
}),
|
||||
)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer client.Close()
|
||||
|
||||
batch := []BatchElem{
|
||||
{Method: "test_method1", Args: []interface{}{}, Result: new(string)},
|
||||
{Method: "test_method2", Args: []interface{}{}, Result: new(string)},
|
||||
}
|
||||
err = client.BatchCallContext(context.Background(), batch)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
if !reqCalled {
|
||||
t.Error("request interceptor was not called for batch")
|
||||
}
|
||||
if !respCalled {
|
||||
t.Error("response interceptor was not called for batch")
|
||||
}
|
||||
// For batch calls, method should be empty string
|
||||
if reqMethod != "" {
|
||||
t.Errorf("request interceptor got method %q, want empty string for batch", reqMethod)
|
||||
}
|
||||
if respMethod != "" {
|
||||
t.Errorf("response interceptor got method %q, want empty string for batch", respMethod)
|
||||
}
|
||||
}
|
||||
|
||||
func TestNotifyWithInterceptors(t *testing.T) {
|
||||
// Test that request interceptor can block notifications.
|
||||
// We don't actually send the notification since Notify is primarily
|
||||
// for persistent connections (WebSocket/IPC), not HTTP.
|
||||
blockErr := errors.New("blocked notification")
|
||||
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
t.Error("server should not have been called")
|
||||
}))
|
||||
defer server.Close()
|
||||
|
||||
client, err := DialOptions(context.Background(), server.URL,
|
||||
WithRequestInterceptor(func(ctx context.Context, method string, args []interface{}) error {
|
||||
if method == "test_notification" {
|
||||
return blockErr
|
||||
}
|
||||
return nil
|
||||
}),
|
||||
)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer client.Close()
|
||||
|
||||
err = client.Notify(context.Background(), "test_notification")
|
||||
if err != blockErr {
|
||||
t.Errorf("got error %v, want %v", err, blockErr)
|
||||
}
|
||||
}
|
||||
|
||||
func TestSubscribeWithInterceptors(t *testing.T) {
|
||||
// Test that request interceptor can block subscription requests.
|
||||
blockErr := errors.New("blocked subscribe")
|
||||
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
t.Error("server should not have been called")
|
||||
}))
|
||||
defer server.Close()
|
||||
|
||||
client, err := DialOptions(context.Background(), server.URL,
|
||||
WithRequestInterceptor(func(ctx context.Context, method string, args []interface{}) error {
|
||||
if method == "eth_subscribe" {
|
||||
return blockErr
|
||||
}
|
||||
return nil
|
||||
}),
|
||||
)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer client.Close()
|
||||
|
||||
ch := make(chan interface{})
|
||||
_, err = client.EthSubscribe(context.Background(), ch, "newHeads")
|
||||
|
||||
// Should get ErrNotificationsUnsupported for HTTP client first,
|
||||
// but if we had a WS client, the interceptor would block it.
|
||||
// For now, just verify HTTP correctly returns unsupported.
|
||||
if err != ErrNotificationsUnsupported {
|
||||
t.Errorf("got error %v, want %v", err, ErrNotificationsUnsupported)
|
||||
}
|
||||
c.Close()
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in a new issue