Socket Consumer

The Socket consumer provides a low-level transport implementation that communicates with the RudderStack Data Plane using raw TCP sockets via PHP's pfsockopen lib/Consumer/Socket.php70-76 It is designed for environments where the curl extension might be unavailable or where fine-grained control over the HTTP request construction and response parsing is required.

Implementation Overview

The Socket class extends QueueConsumer and implements the flushBatch method to transmit event data lib/Consumer/Socket.php7-9 Unlike the LibCurl consumer, which relies on a high-level library, the Socket consumer manually constructs HTTP/1.1 POST requests and parses the resulting byte stream lib/Consumer/Socket.php96-134

Connection Lifecycle

1. Socket Creation: Uses pfsockopen to establish a persistent connection lib/Consumer/Socket.php70 It defaults to a 5-second timeout lib/Consumer/Socket.php24
2. Protocol Selection: Supports both ssl and tls protocols based on the tls option lib/Consumer/Socket.php64
3. Request Construction: Manually builds the HTTP wire format, including headers for Authorization, Content-Type, and User-Agent lib/Consumer/Socket.php96-108
4. Data Transmission: Writes the payload to the socket using fwrite in a loop to ensure all bytes are sent lib/Consumer/Socket.php173-181
5. Response Handling: Reads the raw response using fread and passes it to parseResponse to extract the HTTP status code lib/Consumer/Socket.php187-188

Data Flow: Message to Socket

The following diagram illustrates how an array of messages is transformed into a raw HTTP request and sent over the wire.

Socket Request Flow

Sources: lib/Consumer/Socket.php34-51 lib/Consumer/Socket.php58-85 lib/Consumer/Socket.php94-151 lib/Consumer/Socket.php161-220

Key Features

Manual HTTP Response Parsing

Since it does not use a high-level HTTP client, the consumer implements parseResponse to manually split the HTTP response into headers and body lib/Consumer/Socket.php187 This allows the SDK to identify specific status codes like 429 (Rate Limited) or 5xx (Server Error) to trigger retry logic lib/Consumer/Socket.php197-200

AnonymousId Header Encoding

To assist the RudderStack server in maintaining event ordering, the consumer extracts the userId or anonymousId from the first message in a single-event batch lib/Consumer/Socket.php111-117 This value is base64-encoded and sent as a custom AnonymousId HTTP header lib/Consumer/Socket.php120-122

Exponential Backoff

The makeRequest function implements a retry loop with exponential backoff lib/Consumer/Socket.php168-171

• Initial Wait: 100ms lib/Consumer/Socket.php169
• Retry Conditions: Retries occur if the status code is 0, 429, or between 500-600 lib/Consumer/Socket.php200
• Backoff Strategy: The wait time doubles with each retry ($backoff *= 2) until it reaches the maximum_backoff_duration lib/Consumer/Socket.php201-215

Payload Size Enforcement

Before transmission, the consumer verifies that the total size of the constructed HTTP request (headers + body) does not exceed 512KB (specifically checked against 500 * 1024 bytes) lib/Consumer/Socket.php137-143 If the limit is exceeded, an error is logged and the request is aborted to prevent rejection by the Data Plane.

Class Interaction Map

This diagram maps the internal private methods of the Socket class to their functional roles in the network stack.

Socket Internal Logic Map

Sources: lib/Consumer/Socket.php7-10 lib/Consumer/Socket.php34-51 lib/Consumer/Socket.php70-76 lib/Consumer/Socket.php127

Configuration Options

The Socket consumer accepts an $options array during instantiation lib/Consumer/Socket.php21:

Option	Type	Default	Description
timeout	int	5	Connection timeout in seconds for pfsockopen lib/Consumer/Socket.php23-25
tls	bool	false	If true, uses tls:// protocol; otherwise uses ssl:// lib/Consumer/Socket.php27-29 lib/Consumer/Socket.php64
error_handler	callable	null	Custom function to handle socket or HTTP errors lib/Consumer/Socket.php18
debug	bool	false	If enabled, errors are passed to the error handler for 4xx responses lib/Consumer/Socket.php19 lib/Consumer/Socket.php207-209

Sources: lib/Consumer/Socket.php12-32 lib/Consumer/Socket.php64-67