0
# Socket Configuration
1

2
Low-level socket configuration, timeout management, and network-level options for advanced connection control and performance tuning. These functions provide fine-grained control over the underlying network socket behavior.
3

4
## Capabilities
5

6
### Default Socket Options
7

8
Pre-configured socket options for optimal WebSocket performance and reliability.
9

10
```python { .api }
11
DEFAULT_SOCKET_OPTION: list
12
"""
13
Default socket configuration for WebSocket connections.
14

15
Includes:
16
- TCP_NODELAY: Disable Nagle's algorithm for low latency
17
- SO_KEEPALIVE: Enable TCP keepalive (if supported)
18
- TCP_KEEPIDLE: Keepalive idle time (if supported) 
19
- TCP_KEEPINTVL: Keepalive probe interval (if supported)
20
- TCP_KEEPCNT: Keepalive probe count (if supported)
21

22
These options optimize for WebSocket's real-time communication patterns.
23
"""
24
```
25

26
### Socket Options Container
27

28
Container class for managing socket and SSL configuration options.
29

30
```python { .api }
31
class sock_opt:
32
    def __init__(self, sockopt: list, sslopt: dict) -> None:
33
        """
34
        Initialize socket options container.
35

36
        Parameters:
37
        - sockopt: List of socket options for socket.setsockopt()
38
          Format: [(level, optname, value), ...]
39
          Example: [(socket.SOL_TCP, socket.TCP_NODELAY, 1)]
40
        - sslopt: Dictionary of SSL options for SSL context
41
          Example: {"cert_reqs": ssl.CERT_REQUIRED, "ca_certs": "/path/to/ca.pem"}
42

43
        Attributes:
44
        - sockopt: Socket options list
45
        - sslopt: SSL options dictionary  
46
        - timeout: Socket timeout value
47
        """
48
```
49

50
### Global Timeout Management
51

52
Configure default timeout behavior for all WebSocket connections.
53

54
```python { .api }
55
def setdefaulttimeout(timeout: Union[int, float, None]) -> None:
56
    """
57
    Set global default timeout for WebSocket connections.
58

59
    Parameters:
60
    - timeout: Default timeout in seconds (int/float) or None for blocking
61

62
    Note: This affects all new WebSocket connections that don't specify
63
    a timeout explicitly. Existing connections are not affected.
64
    """
65

66
def getdefaulttimeout() -> Union[int, float, None]:
67
    """
68
    Get current global default timeout setting.
69

70
    Returns:
71
    Union[int, float, None]: Current default timeout in seconds or None
72
    """
73
```
74

75
### Low-Level Socket Operations
76

77
Direct socket I/O operations with WebSocket-specific error handling and timeout support.
78

79
```python { .api }
80
def recv(sock: socket.socket, bufsize: int) -> bytes:
81
    """
82
    Receive data from socket with WebSocket-specific error handling.
83

84
    Parameters:
85
    - sock: Socket object to receive from
86
    - bufsize: Maximum number of bytes to receive
87

88
    Returns:
89
    bytes: Received data
90

91
    Raises:
92
    WebSocketConnectionClosedException: If socket is closed or connection lost
93
    WebSocketTimeoutException: If receive operation times out
94
    """
95

96
def recv_line(sock: socket.socket) -> bytes:
97
    """
98
    Receive single line from socket (reads until newline).
99

100
    Parameters:
101
    - sock: Socket object to receive from
102

103
    Returns:
104
    bytes: Received line including newline character
105

106
    Note: Used primarily for HTTP header parsing during handshake
107
    """
108

109
def send(sock: socket.socket, data: Union[bytes, str]) -> int:
110
    """
111
    Send data through socket with proper encoding and error handling.
112

113
    Parameters:
114
    - sock: Socket object to send through
115
    - data: Data to send (automatically UTF-8 encoded if string)
116

117
    Returns:
118
    int: Number of bytes sent
119

120
    Raises:
121
    WebSocketConnectionClosedException: If socket is closed
122
    WebSocketTimeoutException: If send operation times out
123
    """
124
```
125

126
## Usage Examples
127

128
### Custom Socket Options
129

130
```python
131
import websocket
132
import socket
133

134
# Define custom socket options
135
custom_sockopt = [
136
    # Disable Nagle's algorithm for low latency
137
    (socket.SOL_TCP, socket.TCP_NODELAY, 1),
138
    # Set socket buffer sizes
139
    (socket.SOL_SOCKET, socket.SO_RCVBUF, 65536),
140
    (socket.SOL_SOCKET, socket.SO_SNDBUF, 65536),
141
    # Enable keepalive with custom settings
142
    (socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1),
143
]
144

145
# Add platform-specific keepalive options
146
if hasattr(socket, 'TCP_KEEPIDLE'):
147
    custom_sockopt.append((socket.SOL_TCP, socket.TCP_KEEPIDLE, 60))
148
if hasattr(socket, 'TCP_KEEPINTVL'):
149
    custom_sockopt.append((socket.SOL_TCP, socket.TCP_KEEPINTVL, 30))
150
if hasattr(socket, 'TCP_KEEPCNT'):
151
    custom_sockopt.append((socket.SOL_TCP, socket.TCP_KEEPCNT, 3))
152

153
# Use custom socket options with WebSocket
154
ws = websocket.create_connection(
155
    "ws://echo.websocket.events/",
156
    sockopt=custom_sockopt
157
)
158

159
print("Connected with custom socket options")
160
ws.send("Test message")
161
response = ws.recv()
162
print(f"Response: {response}")
163
ws.close()
164
```
165

166
### SSL Configuration
167

168
```python
169
import websocket
170
import ssl
171

172
# Comprehensive SSL configuration
173
ssl_options = {
174
    # Certificate verification
175
    "cert_reqs": ssl.CERT_REQUIRED,
176
    "ca_certs": "/etc/ssl/certs/ca-certificates.crt",
177
    
178
    # Client certificate authentication
179
    "certfile": "/path/to/client.crt",
180
    "keyfile": "/path/to/client.key",
181
    
182
    # Protocol and cipher selection
183
    "ssl_version": ssl.PROTOCOL_TLS,
184
    "ciphers": "ECDHE+AESGCM:ECDHE+CHACHA20:DHE+AESGCM:DHE+CHACHA20:!aNULL:!MD5:!DSS",
185
    
186
    # Hostname verification
187
    "check_hostname": True,
188
    
189
    # Security options
190
    "do_handshake_on_connect": True,
191
    "suppress_ragged_eofs": True,
192
}
193

194
try:
195
    ws = websocket.create_connection(
196
        "wss://secure.example.com/websocket",
197
        sslopt=ssl_options
198
    )
199
    print("Secure connection established")
200
    print(f"SSL cipher: {ws.sock.cipher()}")
201
    ws.close()
202
    
203
except ssl.SSLError as e:
204
    print(f"SSL error: {e}")
205
except Exception as e:
206
    print(f"Connection error: {e}")
207
```
208

209
### Timeout Configuration
210

211
```python
212
import websocket
213
import time
214

215
def test_timeouts():
216
    # Set global default timeout
217
    websocket.setdefaulttimeout(10.0)
218
    print(f"Default timeout: {websocket.getdefaulttimeout()}")
219
    
220
    # Test connection with default timeout
221
    try:
222
        ws = websocket.create_connection("ws://echo.websocket.events/")
223
        print(f"Connection timeout: {ws.gettimeout()}")
224
        
225
        # Change timeout for this connection
226
        ws.settimeout(5.0)
227
        print(f"Updated timeout: {ws.gettimeout()}")
228
        
229
        ws.send("Test")
230
        response = ws.recv()
231
        print(f"Response: {response}")
232
        ws.close()
233
        
234
    except websocket.WebSocketTimeoutException:
235
        print("Operation timed out")
236
    
237
    # Test with explicit timeout override
238
    try:
239
        ws = websocket.create_connection(
240
            "ws://echo.websocket.events/",
241
            timeout=2.0  # Override default
242
        )
243
        print(f"Override timeout: {ws.gettimeout()}")
244
        ws.close()
245
        
246
    except Exception as e:
247
        print(f"Error: {e}")
248
    
249
    # Reset to no timeout
250
    websocket.setdefaulttimeout(None)
251
    print(f"Reset timeout: {websocket.getdefaulttimeout()}")
252

253
test_timeouts()
254
```
255

256
### Performance Tuning
257

258
```python
259
import websocket
260
import socket
261

262
def create_high_performance_connection(url):
263
    """Create WebSocket connection optimized for high throughput."""
264
    
265
    # High-performance socket options
266
    perf_sockopt = [
267
        # Disable Nagle's algorithm
268
        (socket.SOL_TCP, socket.TCP_NODELAY, 1),
269
        
270
        # Large socket buffers for high throughput
271
        (socket.SOL_SOCKET, socket.SO_RCVBUF, 1024 * 1024),  # 1MB receive buffer
272
        (socket.SOL_SOCKET, socket.SO_SNDBUF, 1024 * 1024),  # 1MB send buffer
273
        
274
        # Enable keepalive
275
        (socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1),
276
    ]
277
    
278
    # Add TCP_USER_TIMEOUT if available (Linux)
279
    if hasattr(socket, 'TCP_USER_TIMEOUT'):
280
        perf_sockopt.append((socket.SOL_TCP, socket.TCP_USER_TIMEOUT, 30000))  # 30s
281
    
282
    # Create connection with performance options
283
    ws = websocket.create_connection(
284
        url,
285
        sockopt=perf_sockopt,
286
        skip_utf8_validation=True,  # Skip validation for binary data
287
        enable_multithread=True     # Enable thread safety
288
    )
289
    
290
    return ws
291

292
# Usage
293
ws = create_high_performance_connection("ws://echo.websocket.events/")
294
print("High-performance connection established")
295

296
# Test large message throughput
297
large_message = "x" * 10000
298
start_time = time.time()
299

300
for i in range(100):
301
    ws.send(large_message)
302
    response = ws.recv()
303

304
elapsed = time.time() - start_time
305
print(f"Sent/received 100 large messages in {elapsed:.2f}s")
306
ws.close()
307
```
308

309
### Low-Level Socket Access
310

311
```python
312
import websocket
313
import select
314
import socket
315

316
def manual_socket_handling():
317
    """Demonstrate direct socket access for advanced use cases."""
318
    
319
    ws = websocket.create_connection("ws://echo.websocket.events/")
320
    
321
    # Get underlying socket
322
    raw_socket = ws.sock
323
    print(f"Socket type: {type(raw_socket)}")
324
    print(f"Socket family: {raw_socket.family}")
325
    print(f"Socket type: {raw_socket.type}")
326
    
327
    # Use socket for select operations
328
    ws.send("async test")
329
    
330
    # Wait for data with select
331
    ready = select.select([raw_socket], [], [], 5.0)
332
    if ready[0]:
333
        print("Data available for reading")
334
        response = ws.recv()
335
        print(f"Response: {response}")
336
    else:
337
        print("No data received within timeout")
338
    
339
    # Get socket statistics (platform dependent)
340
    try:
341
        sockopt_info = {
342
            "SO_RCVBUF": raw_socket.getsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF),
343
            "SO_SNDBUF": raw_socket.getsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF),
344
            "SO_KEEPALIVE": raw_socket.getsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE),
345
        }
346
        
347
        if hasattr(socket, 'TCP_NODELAY'):
348
            sockopt_info["TCP_NODELAY"] = raw_socket.getsockopt(socket.SOL_TCP, socket.TCP_NODELAY)
349
            
350
        print("Socket options:")
351
        for opt, value in sockopt_info.items():
352
            print(f"  {opt}: {value}")
353
            
354
    except Exception as e:
355
        print(f"Could not read socket options: {e}")
356
    
357
    ws.close()
358

359
manual_socket_handling()
360
```
361

362
### Socket Error Handling
363

364
```python
365
import websocket
366
import socket
367
import errno
368

369
def robust_socket_operations():
370
    """Demonstrate robust socket error handling."""
371
    
372
    try:
373
        ws = websocket.create_connection("ws://echo.websocket.events/")
374
        
375
        # Test various socket error conditions
376
        test_cases = [
377
            {"name": "Normal operation", "data": "Hello"},
378
            {"name": "Large message", "data": "x" * 65536},
379
            {"name": "Binary data", "data": b"\x00\x01\x02\x03"},
380
        ]
381
        
382
        for test in test_cases:
383
            try:
384
                print(f"Testing: {test['name']}")
385
                
386
                if isinstance(test['data'], bytes):
387
                    ws.send_bytes(test['data'])
388
                else:
389
                    ws.send(test['data'])
390
                    
391
                response = ws.recv()
392
                print(f"  Success: received {len(response)} bytes")
393
                
394
            except websocket.WebSocketConnectionClosedException:
395
                print(f"  Connection closed during {test['name']}")
396
                break
397
                
398
            except websocket.WebSocketTimeoutException:
399
                print(f"  Timeout during {test['name']}")
400
                
401
            except socket.error as e:
402
                if e.errno == errno.ECONNRESET:
403
                    print(f"  Connection reset during {test['name']}")
404
                elif e.errno == errno.EPIPE:
405
                    print(f"  Broken pipe during {test['name']}")
406
                else:
407
                    print(f"  Socket error during {test['name']}: {e}")
408
                    
409
        ws.close()
410
        
411
    except Exception as e:
412
        print(f"Failed to establish connection: {e}")
413

414
robust_socket_operations()
415
```
416

417
### Custom Socket Factory
418

419
```python
420
import websocket
421
import socket
422
import ssl
423

424
class CustomSocketFactory:
425
    """Factory for creating pre-configured sockets."""
426
    
427
    @staticmethod
428
    def create_optimized_socket():
429
        """Create socket with optimal settings for WebSocket."""
430
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
431
        
432
        # Apply optimizations
433
        sock.setsockopt(socket.SOL_TCP, socket.TCP_NODELAY, 1)
434
        sock.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)
435
        
436
        # Set large buffers
437
        sock.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 262144)
438
        sock.setsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF, 262144)
439
        
440
        return sock
441
    
442
    @staticmethod 
443
    def create_ssl_socket(hostname):
444
        """Create SSL socket with secure configuration."""
445
        context = ssl.create_default_context()
446
        context.check_hostname = True
447
        context.verify_mode = ssl.CERT_REQUIRED
448
        
449
        # Security hardening
450
        context.minimum_version = ssl.TLSVersion.TLSv1_2
451
        context.set_ciphers('ECDHE+AESGCM:ECDHE+CHACHA20:DHE+AESGCM')
452
        
453
        sock = CustomSocketFactory.create_optimized_socket()
454
        ssl_sock = context.wrap_socket(sock, server_hostname=hostname)
455
        
456
        return ssl_sock
457

458
def use_custom_socket():
459
    """Demonstrate using pre-configured socket with WebSocket."""
460
    
461
    # Create optimized socket
462
    custom_sock = CustomSocketFactory.create_optimized_socket()
463
    
464
    # Use with WebSocket connection
465
    ws = websocket.create_connection(
466
        "ws://echo.websocket.events/",
467
        socket=custom_sock
468
    )
469
    
470
    print("Connected with custom socket")
471
    ws.send("Custom socket test")
472
    response = ws.recv()
473
    print(f"Response: {response}")
474
    ws.close()
475

476
use_custom_socket()
477
```
478

479
## Types
480

481
```python { .api }
482
# Socket option tuple format
483
SocketOption = tuple[int, int, int]  # (level, optname, value)
484
SocketOptionList = list[SocketOption]
485

486
# SSL option dictionary keys (common options)
487
SSLOptions = dict[str, Union[str, int, bool]]
488

489
# Timeout types
490
TimeoutValue = Union[int, float, None]
491

492
# Socket operation return types
493
SocketBytes = bytes
494
SocketSendResult = int
495
```