Refine connection exception for internal control flow

composer.json

@@ -22,6 +22,7 @@
   ],
   "require": {
     "php": ">=8.1",
+    "ext-ffi": "*",
     "ext-sockets": "*",
     "ext-openssl": "*",
     "psr/http-message": "*",

docs/EXCEPTION_HANDLING.md

@@ -0,0 +1,214 @@
+# Stream Exception Handling Guide
+
+## Overview
+
+The Stream module uses a clear exception hierarchy that separates internal control-flow exceptions from application-level exceptions that user code should handle.
+
+## Exception Hierarchy
+
+```
+RuntimeException
+├── StreamException (base for user-catchable exceptions)
+│   └── TransportException (recoverable transport errors)
+│       ├── TransportTimeoutException (timeout errors)
+│       ├── ConnectionTimeoutException (connection timeouts)
+│       └── WriteClosedException (write to closed stream)
+└── ConnectionException (internal control-flow - DO NOT CATCH)
+```
+
+## Key Principles
+
+### 1. Internal Control-Flow Exceptions (DO NOT CATCH)
+
+**ConnectionException** is an internal exception used exclusively by the reactor for connection termination. It implements the `AbortConnection` marker interface.
+
+- **Purpose**: Signal immediate connection termination to the reactor
+- **Usage**: Only thrown internally when connection becomes unusable
+- **Handling**: Only caught by reactor's exception boundary, never by user code
+
+```php
+// ❌ NEVER do this
+try {
+    $stream->read(1024);
+} catch (ConnectionException $e) {
+    // This breaks the reactor's control flow!
+}
+
+// ✅ Use connection lifecycle events instead
+$stream->onClose(function (CloseEvent $event) {
+    echo "Connection closed: {$event->reason->value}";
+});
+```
+
+### 2. Application-Level Exceptions (Safe to Catch)
+
+**TransportException** and its subclasses represent recoverable errors that application code can handle.
+
+```php
+// ✅ Safe to catch and handle
+try {
+    $stream->write($data);
+} catch (WriteClosedException $e) {
+    // Write side is closed, but connection might still be readable
+    echo "Cannot write: {$e->getMessage()}";
+} catch (TransportException $e) {
+    // Other transport-level errors
+    echo "Transport error: {$e->getMessage()}";
+}
+```
+
+## Connection Lifecycle Events
+
+Instead of catching exceptions, use lifecycle events to handle connection state changes:
+
+### onClose(CloseEvent)
+Triggered once when the connection terminates.
+
+```php
+$stream->onClose(function (CloseEvent $event) {
+    echo "Closed: {$event->reason->value} by {$event->initiator}";
+    // Cleanup resources, update connection pools, etc.
+});
+```
+
+**CloseEvent** provides:
+- `reason`: ConnectionAbortReason enum (PEER_CLOSED, RESET, TIMEOUT, etc.)
+- `initiator`: 'peer' | 'local' | 'system'
+- `message`: Optional descriptive message
+- `lastError`: Optional underlying exception
+- `timestamp`: When the close occurred
+
+### onReadableEnd()
+Triggered when the read side closes (EOF) but connection may still be writable.
+
+```php
+$stream->onReadableEnd(function () use ($stream) {
+    echo "Read side closed - no more data from peer";
+    // Can still write final response, then close
+    $stream->write("HTTP/1.1 200 OK\r\n\r\nGoodbye");
+    $stream->close();
+});
+```
+
+### onWritableEnd()
+Triggered when the write side closes but connection may still be readable.
+
+```php
+$stream->onWritableEnd(function () use ($stream) {
+    echo "Write side closed - cannot send more data";
+    // Can still read remaining data from peer
+});
+```
+
+## Half-Close Support
+
+Half-close allows one side of a connection to be closed while the other remains open. This is useful for protocols like HTTP where the client sends a complete request, then the server sends a complete response.
+
+### Configuration
+
+```php
+$stream = new Stream($resource);
+$stream->supportsHalfClose = true; // Default: true
+```
+
+### Behavior
+
+When `supportsHalfClose = true`:
+- `read()` returning EOF triggers `onReadableEnd()` if registered, otherwise throws `ConnectionException`
+- `write()` getting EPIPE triggers `onWritableEnd()` if registered, otherwise throws `ConnectionException`
+
+When `supportsHalfClose = false`:
+- EOF or EPIPE immediately throws `ConnectionException` for reactor termination
+
+## Error Classification
+
+### Fatal Errors (→ ConnectionException)
+These errors indicate the connection is no longer usable:
+- Peer closed connection (EOF without half-close support)
+- Connection reset by peer (ECONNRESET)
+- TLS fatal alerts
+- Broken pipe (EPIPE without half-close support)
+
+### Recoverable Errors (→ TransportException)
+These errors can be handled by application logic:
+- Connection timeouts (can retry)
+- Write to closed stream (can detect and handle)
+- Protocol-level errors (application can decide response)
+- Temporary resource unavailability
+
+## Migration from Old API
+
+### Before
+```php
+try {
+    $data = $stream->read(1024);
+} catch (ConnectionException $e) {
+    if ($e->getCode() === ConnectionException::CONNECTION_CLOSED) {
+        // Handle close
+    }
+}
+```
+
+### After
+```php
+// Use events for lifecycle management
+$stream->onClose(function (CloseEvent $event) {
+    if ($event->reason === ConnectionAbortReason::PEER_CLOSED) {
+        // Handle close
+    }
+});
+
+$stream->onReadableEnd(function () {
+    // Handle EOF/half-close
+});
+
+// Only catch recoverable exceptions
+try {
+    $data = $stream->read(1024);
+} catch (TransportException $e) {
+    // Handle recoverable errors only
+}
+```
+
+## Best Practices
+
+1. **Never catch ConnectionException** - use lifecycle events instead
+2. **Register onClose for cleanup** - guaranteed to be called once per connection
+3. **Use onReadableEnd/onWritableEnd** for half-close protocols
+4. **Catch TransportException** for recoverable error handling
+5. **Let the reactor handle fatal errors** - it will clean up and emit events
+
+## Common Patterns
+
+### HTTP Server
+```php
+$stream->onReadableEnd(function () use ($stream, $response) {
+    // Client finished sending request, send response
+    $stream->write($response);
+    $stream->close();
+});
+
+$stream->onClose(function (CloseEvent $event) use ($connectionPool) {
+    $connectionPool->remove($stream);
+});
+```
+
+### Database Client
+```php
+$stream->onClose(function (CloseEvent $event) use ($pendingQueries) {
+    // Fail all pending queries
+    foreach ($pendingQueries as $query) {
+        $query->fail(new TransportException("Connection lost: {$event->reason->value}"));
+    }
+});
+```
+
+### WebSocket
+```php
+$stream->onClose(function (CloseEvent $event) use ($subscriptions) {
+    // Clean up subscriptions
+    foreach ($subscriptions as $sub) {
+        $sub->cancel();
+    }
+});
+```

examples/enhanced-stream-usage.php

@@ -0,0 +1,67 @@
+<?php declare(strict_types=1);
+
+include __DIR__ . '/../vendor/autoload.php';
+
+use Ripple\Socket;
+use Ripple\Stream\CloseEvent;
+use Ripple\Stream\Exception\TransportException;
+use Ripple\Stream\Exception\WriteClosedException;
+use Ripple\Utils\Output;
+
+use function Co\wait;
+
+/**
+ * Example demonstrating proper usage of the enhanced Stream API
+ *
+ * This example shows:
+ * 1. How to handle connection lifecycle events (onClose, onReadableEnd)
+ * 2. Proper exception handling (catch TransportException, not ConnectionException)
+ * 3. Half-close support for HTTP-like protocols
+ */
+
+try {
+    $stream = Socket::connect('tcp://httpbin.org:80');
+    $stream->setBlocking(false);
+
+    // Register connection lifecycle events
+    $stream->onClose(function (CloseEvent $event) {
+        Output::info("Connection closed: {$event->reason->value} by {$event->initiator}");
+    });
+
+    $stream->onReadableEnd(function () use ($stream) {
+        Output::info("Read side closed - server finished sending response");
+        // We can still write if needed, but in HTTP we typically close now
+        $stream->close();
+    });
+
+    // Send HTTP request
+    $request = "GET /get HTTP/1.1\r\nHost: httpbin.org\r\nConnection: close\r\n\r\n";
+    $stream->write($request);
+
+    // Read response
+    $stream->onReadable(function () use ($stream) {
+        try {
+            $data = $stream->read(1024);
+            if ($data !== '') {
+                echo $data;
+            }
+            // Note: When server closes connection, onReadableEnd will be triggered
+            // instead of an exception, allowing graceful handling
+        } catch (WriteClosedException $e) {
+            // This is a recoverable exception - safe to catch
+            Output::warning("Attempted to write to closed connection: {$e->getMessage()}");
+        } catch (TransportException $e) {
+            // This is a recoverable exception - safe to catch
+            Output::error("Transport error: {$e->getMessage()}");
+            $stream->close();
+        }
+        // Note: We NEVER catch ConnectionException - that's handled internally by the reactor
+    });
+
+} catch (TransportException $e) {
+    // Connection establishment failed - this is recoverable
+    Output::error("Failed to connect: {$e->getMessage()}");
+    exit(1);
+}
+
+wait();

examples/socket-client.php

@@ -3,7 +3,7 @@
 include __DIR__ . '/../vendor/autoload.php';
 
 use Ripple\Socket;
-use Ripple\Stream\Exception\ConnectionException;
+use Ripple\Stream\Exception\TransportException;
 use Ripple\Utils\Output;
 
 use function Co\wait;
@@ -14,7 +14,7 @@
     #  Enable SSL
     // $connection->enableSSL();
     $connection->setBlocking(false);
-} catch (ConnectionException $e) {
+} catch (TransportException $e) {
     Output::warning($e->getMessage());
     exit(1);
 }

examples/socket-tunnel.php

@@ -2,7 +2,7 @@
 
 include __DIR__ . '/../vendor/autoload.php';
 
-use Ripple\Stream\Exception\ConnectionException;
+use Ripple\Stream\Exception\TransportException;
 use Ripple\Tunnel\Socks5;
 use Ripple\Utils\Output;
 
@@ -26,7 +26,7 @@
         }
         echo $data;
     });
-} catch (ConnectionException $e) {
+} catch (TransportException $e) {
     Output::warning($e->getMessage());
     exit(1);
 }
@@ -53,7 +53,7 @@
         }
         echo $data;
     });
-} catch (ConnectionException $e) {
+} catch (TransportException $e) {
     Output::warning($e->getMessage());
     exit(1);
 }

src/Channel/Channel.php

@@ -17,7 +17,8 @@
 use Ripple\File\Lock;
 use Ripple\Kernel;
 use Ripple\Stream;
-use Ripple\Stream\Exception\ConnectionException;
+use Ripple\Stream\Exception\AbortConnection;
+use Ripple\Stream\Exception\TransportException;
 use Ripple\Utils\Path;
 use Ripple\Utils\Serialization\Zx7e;
 use Throwable;
@@ -193,14 +194,25 @@ public function receive(bool $blocking = true): mixed
 
             if ($this->readLock->lock(blocking: false)) {
                 try {
-                    if ($this->stream->read(1) === chr(Channel::FRAME_HEADER)) {
+                    $header = $this->stream->read(1);
+                    if ($header === chr(Channel::FRAME_HEADER)) {
                         break;
+                    } elseif ($header === '') {
+                        // EOF - connection closed
+                        $this->readLock->unlock();
+                        throw new TransportException('Connection closed while reading frame header');
                     } else {
+                        // Invalid frame header
                         $this->stream->close();
-                        throw new ConnectionException('Failed to read frame header.');
+                        throw new TransportException('Invalid frame header received');
                     }
-                } catch (ConnectionException) {
+                } catch (AbortConnection $e) {
+                    // Internal control flow exception - re-throw to let reactor handle
                     $this->readLock->unlock();
+                    throw $e;
+                } catch (TransportException $e) {
+                    $this->readLock->unlock();
+                    throw $e;
                 }
             }
         }