ReactPHP HTTP

ReactPHP生態系の非同期HTTPクライアント。イベント駆動型アーキテクチャによる高並行性HTTP通信を実現。ノンブロッキングI/O、ストリーミング、リアルタイム通信に最適化。従来のPHP制約を超えた高性能アプリケーション開発を支援。

HTTPクライアントPHP非同期イベント駆動ストリーミングReactPHP

GitHub概要

reactphp/http

Event-driven, streaming HTTP client and server implementation for ReactPHP.

リポジトリ:https://github.com/reactphp/http

ホームページ:https://reactphp.org/http/

スター775

ウォッチ38

フォーク167

作成日:2012年5月10日

言語:PHP

ライセンス:MIT License

トピックス

httphttp-clienthttp-serverhttpsphpreactphpresponse-streamstreaming-requeststreaming-response

スター履歴

データ取得日時: 2025/10/22 09:54

ライブラリ

ReactPHP HTTP

概要

ReactPHP HTTPは「PHPのための非同期HTTP クライアント・サーバーライブラリ」として開発された、ReactPHPエコシステムの中核を成すHTTPコンポーネントです。イベント駆動・非ブロッキングI/Oによる高性能なHTTP通信を実現し、従来のPHPでは困難だった同時接続や長時間接続を効率的に処理。PSR-7メッセージ抽象化をベースとしたモダンなAPIデザインと、Promise ベースの非同期パターンにより、大量のHTTPリクエスト処理やリアルタイム通信が必要なアプリケーション開発において革新的なソリューションを提供します。

詳細

ReactPHP HTTP 2025年版は、PHP言語における非同期HTTP通信の先駆けとして10年以上の開発実績を持つ成熟したライブラリです。ReactPHPのイベントループをベースとした設計により、単一プロセスで数千の同時HTTP接続を効率的に処理可能。PSR-7 HTTP メッセージインターフェースを完全サポートし、標準的なHTTPリクエスト・レスポンス処理を提供。ストリーミング処理による大容量ファイルの効率的な送受信、WebSocketサポート、カスタムミドルウェア対応等の高度な機能を搭載。Promise/A+準拠の非同期パターンにより、従来のブロッキングI/Oでは実現困難な高性能Webアプリケーションの構築を可能にします。

主な特徴

イベント駆動アーキテクチャ: ReactPHPイベントループによる非ブロッキングI/O
PSR-7準拠: 標準的なHTTPメッセージインターフェース完全対応
Promise ベース: 非同期処理のための直感的なAPI設計
ストリーミング対応: 大容量データの効率的な送受信
高い同時接続性: 単一プロセスでの数千接続処理
拡張性: カスタムミドルウェアとプラグインサポート

メリット・デメリット

メリット

従来のPHPでは不可能だった真の非同期HTTP処理を実現
イベント駆動設計により極めて高いメモリ効率と処理性能
PSR-7準拠により他のPHPライブラリとの高い互換性
Promise パターンによる直感的で読みやすい非同期コード記述
ReactPHPエコシステムとの完全統合による拡張性
大量の同時リクエスト処理に適した優れたスケーラビリティ

デメリット

非同期プログラミングの学習コストが比較的高い
従来のPHP開発パターンとは大きく異なるパラダイム
デバッグとトラブルシューティングが複雑になる場合がある
すべてのPHPライブラリが非同期対応しているわけではない
小規模な同期処理にはオーバーヘッドとなる可能性
エラーハンドリングが Promise チェーンで複雑化する場合がある

参考ページ

書き方の例

インストールと基本セットアップ

# Composer でのインストール
composer require react/http react/socket

# 開発用の依存関係
composer require --dev react/http react/promise-timer

# ReactPHP エコシステム全体
composer require react/socket react/stream react/promise react/http

# プロジェクト初期化例
mkdir reactphp-http-project
cd reactphp-http-project
composer init
composer require react/http psr/http-message

基本的な HTTP クライアント機能

<?php
require 'vendor/autoload.php';

use React\EventLoop\Loop;
use React\Http\Browser;
use React\Promise\Promise;
use Psr\Http\Message\ResponseInterface;

// イベントループの初期化
$loop = Loop::get();

// HTTP ブラウザ（クライアント）の作成
$browser = new Browser($loop);

// 基本的な GET リクエスト
function simpleGetRequest($browser) {
    echo "=== Basic GET Request ===\n";
    
    $promise = $browser->get('https://api.example.com/users');
    
    $promise->then(
        function (ResponseInterface $response) {
            echo "Status: " . $response->getStatusCode() . "\n";
            echo "Headers: " . json_encode($response->getHeaders(), JSON_PRETTY_PRINT) . "\n";
            echo "Body: " . $response->getBody() . "\n\n";
            return $response;
        },
        function (Exception $error) {
            echo "Error: " . $error->getMessage() . "\n\n";
        }
    );
    
    return $promise;
}

// 複数の同時 GET リクエスト
function concurrentGetRequests($browser) {
    echo "=== Concurrent GET Requests ===\n";
    
    $urls = [
        'https://api.example.com/users',
        'https://api.example.com/posts', 
        'https://api.example.com/comments'
    ];
    
    $promises = [];
    foreach ($urls as $url) {
        $promises[] = $browser->get($url)->then(
            function (ResponseInterface $response) use ($url) {
                echo "Completed: $url (Status: " . $response->getStatusCode() . ")\n";
                return [
                    'url' => $url,
                    'status' => $response->getStatusCode(),
                    'body' => (string) $response->getBody()
                ];
            },
            function (Exception $error) use ($url) {
                echo "Failed: $url - " . $error->getMessage() . "\n";
                return null;
            }
        );
    }
    
    // すべてのリクエストの完了を待つ
    return \React\Promise\all($promises)->then(
        function ($results) {
            echo "All requests completed successfully!\n";
            echo "Results count: " . count(array_filter($results)) . "\n\n";
            return $results;
        }
    );
}

// タイムアウト付きリクエスト
function requestWithTimeout($browser) {
    echo "=== Request with Timeout ===\n";
    
    $promise = $browser->withTimeout(5.0)->get('https://api.example.com/slow-endpoint');
    
    return $promise->then(
        function (ResponseInterface $response) {
            echo "Request completed within timeout\n";
            echo "Status: " . $response->getStatusCode() . "\n\n";
            return $response;
        },
        function (Exception $error) {
            if ($error instanceof \React\Promise\Timer\TimeoutException) {
                echo "Request timed out after 5 seconds\n\n";
            } else {
                echo "Request failed: " . $error->getMessage() . "\n\n";
            }
        }
    );
}

// メイン実行
$browser = new Browser($loop);

// シーケンシャルな実行例
simpleGetRequest($browser)
    ->then(function() use ($browser) {
        return concurrentGetRequests($browser);
    })
    ->then(function() use ($browser) {
        return requestWithTimeout($browser);
    })
    ->then(function() use ($loop) {
        echo "All examples completed. Stopping event loop.\n";
        $loop->stop();
    });

// イベントループの開始
echo "Starting ReactPHP HTTP client examples...\n";
$loop->run();

POST/PUT/DELETE リクエストと認証

<?php
require 'vendor/autoload.php';

use React\EventLoop\Loop;
use React\Http\Browser;
use Psr\Http\Message\ResponseInterface;

$loop = Loop::get();
$browser = new Browser($loop);

// POST リクエスト（JSON データ送信）
function postJsonRequest($browser) {
    echo "=== POST JSON Request ===\n";
    
    $userData = [
        'name' => '田中太郎',
        'email' => '[email protected]',
        'age' => 30
    ];
    
    $headers = [
        'Content-Type' => 'application/json',
        'Authorization' => 'Bearer your-access-token',
        'User-Agent' => 'ReactPHP-HTTP-Client/1.0'
    ];
    
    $promise = $browser->post(
        'https://api.example.com/users',
        $headers,
        json_encode($userData)
    );
    
    return $promise->then(
        function (ResponseInterface $response) {
            echo "POST Status: " . $response->getStatusCode() . "\n";
            echo "Response: " . $response->getBody() . "\n\n";
            return $response;
        },
        function (Exception $error) {
            echo "POST Error: " . $error->getMessage() . "\n\n";
        }
    );
}

// フォームデータの POST
function postFormRequest($browser) {
    echo "=== POST Form Request ===\n";
    
    $formData = http_build_query([
        'username' => 'testuser',
        'password' => 'testpass123',
        'remember' => '1'
    ]);
    
    $headers = [
        'Content-Type' => 'application/x-www-form-urlencoded',
        'Content-Length' => strlen($formData)
    ];
    
    return $browser->post('https://api.example.com/login', $headers, $formData)
        ->then(
            function (ResponseInterface $response) {
                echo "Form POST Status: " . $response->getStatusCode() . "\n";
                
                // Cookie の取得例
                if ($response->hasHeader('Set-Cookie')) {
                    $cookies = $response->getHeader('Set-Cookie');
                    echo "Received Cookies: " . implode(', ', $cookies) . "\n";
                }
                
                echo "Login Response: " . $response->getBody() . "\n\n";
                return $response;
            },
            function (Exception $error) {
                echo "Form POST Error: " . $error->getMessage() . "\n\n";
            }
        );
}

// PUT リクエスト（データ更新）
function putRequest($browser) {
    echo "=== PUT Request ===\n";
    
    $updateData = [
        'name' => '田中太郎（更新）',
        'email' => '[email protected]',
        'age' => 31
    ];
    
    $headers = [
        'Content-Type' => 'application/json',
        'Authorization' => 'Bearer your-access-token'
    ];
    
    return $browser->put(
        'https://api.example.com/users/123',
        $headers,
        json_encode($updateData)
    )->then(
        function (ResponseInterface $response) {
            echo "PUT Status: " . $response->getStatusCode() . "\n";
            echo "Updated User: " . $response->getBody() . "\n\n";
            return $response;
        },
        function (Exception $error) {
            echo "PUT Error: " . $error->getMessage() . "\n\n";
        }
    );
}

// DELETE リクエスト
function deleteRequest($browser) {
    echo "=== DELETE Request ===\n";
    
    $headers = [
        'Authorization' => 'Bearer your-access-token'
    ];
    
    return $browser->delete('https://api.example.com/users/123', $headers)
        ->then(
            function (ResponseInterface $response) {
                $status = $response->getStatusCode();
                echo "DELETE Status: $status\n";
                
                if ($status === 204) {
                    echo "User successfully deleted\n\n";
                } else {
                    echo "Delete Response: " . $response->getBody() . "\n\n";
                }
                
                return $response;
            },
            function (Exception $error) {
                echo "DELETE Error: " . $error->getMessage() . "\n\n";
            }
        );
}

// 認証付きリクエストチェーン
function authenticatedRequestChain($browser) {
    echo "=== Authenticated Request Chain ===\n";
    
    // 1. ログイン
    $loginData = ['username' => 'admin', 'password' => 'admin123'];
    
    return $browser->post(
        'https://api.example.com/auth/login',
        ['Content-Type' => 'application/json'],
        json_encode($loginData)
    )->then(
        function (ResponseInterface $response) use ($browser) {
            $loginResponse = json_decode($response->getBody(), true);
            $token = $loginResponse['access_token'] ?? null;
            
            if (!$token) {
                throw new Exception('Login failed: No access token received');
            }
            
            echo "Login successful, token received\n";
            
            // 2. 認証が必要なリソースにアクセス
            return $browser->get(
                'https://api.example.com/admin/users',
                ['Authorization' => "Bearer $token"]
            );
        }
    )->then(
        function (ResponseInterface $response) {
            echo "Protected resource access successful\n";
            echo "Admin Users: " . $response->getBody() . "\n\n";
            return $response;
        }
    )->otherwise(
        function (Exception $error) {
            echo "Authentication chain failed: " . $error->getMessage() . "\n\n";
        }
    );
}

// メイン実行
postJsonRequest($browser)
    ->then(function() use ($browser) {
        return postFormRequest($browser);
    })
    ->then(function() use ($browser) {
        return putRequest($browser);
    })
    ->then(function() use ($browser) {
        return deleteRequest($browser);
    })
    ->then(function() use ($browser) {
        return authenticatedRequestChain($browser);
    })
    ->then(function() use ($loop) {
        echo "All POST/PUT/DELETE examples completed.\n";
        $loop->stop();
    });

$loop->run();

ストリーミングとファイル処理

<?php
require 'vendor/autoload.php';

use React\EventLoop\Loop;
use React\Http\Browser;
use React\Stream\WritableResourceStream;
use React\Stream\ReadableResourceStream;
use Psr\Http\Message\ResponseInterface;

$loop = Loop::get();
$browser = new Browser($loop);

// ストリーミングレスポンスの処理
function streamingResponse($browser) {
    echo "=== Streaming Response ===\n";
    
    return $browser->requestStreaming('GET', 'https://api.example.com/large-dataset')
        ->then(
            function (ResponseInterface $response) {
                echo "Streaming response started (Status: " . $response->getStatusCode() . ")\n";
                
                $body = $response->getBody();
                
                // ストリームからデータを読み取り
                $dataReceived = 0;
                $body->on('data', function ($chunk) use (&$dataReceived) {
                    $dataReceived += strlen($chunk);
                    echo "Received chunk: " . strlen($chunk) . " bytes (Total: $dataReceived bytes)\n";
                    
                    // ここでチャンクデータを処理
                    // 例: データベースへの保存、ファイルへの書き込み等
                });
                
                $body->on('end', function () use (&$dataReceived) {
                    echo "Streaming completed. Total received: $dataReceived bytes\n\n";
                });
                
                $body->on('error', function (Exception $error) {
                    echo "Stream error: " . $error->getMessage() . "\n\n";
                });
                
                return $response;
            },
            function (Exception $error) {
                echo "Streaming request failed: " . $error->getMessage() . "\n\n";
            }
        );
}

// ファイルダウンロード
function downloadFile($browser, $url, $localPath) {
    echo "=== File Download ===\n";
    echo "Downloading: $url\n";
    
    $file = fopen($localPath, 'w');
    if (!$file) {
        echo "Cannot open file for writing: $localPath\n\n";
        return \React\Promise\resolve();
    }
    
    $fileStream = new WritableResourceStream($file, $browser->getLoop());
    
    return $browser->requestStreaming('GET', $url)
        ->then(
            function (ResponseInterface $response) use ($fileStream, $localPath) {
                echo "Download started (Status: " . $response->getStatusCode() . ")\n";
                
                if ($response->hasHeader('Content-Length')) {
                    $contentLength = $response->getHeaderLine('Content-Length');
                    echo "File size: $contentLength bytes\n";
                }
                
                $body = $response->getBody();
                $downloaded = 0;
                
                $body->on('data', function ($chunk) use ($fileStream, &$downloaded) {
                    $fileStream->write($chunk);
                    $downloaded += strlen($chunk);
                    
                    if ($downloaded % 10240 === 0) { // 10KB毎に進捗表示
                        echo "Downloaded: $downloaded bytes\n";
                    }
                });
                
                $body->on('end', function () use ($fileStream, $localPath, &$downloaded) {
                    $fileStream->end();
                    echo "Download completed: $localPath ($downloaded bytes)\n\n";
                });
                
                $body->on('error', function (Exception $error) use ($fileStream) {
                    $fileStream->close();
                    echo "Download error: " . $error->getMessage() . "\n\n";
                });
                
                return $response;
            }
        );
}

// ファイルアップロード
function uploadFile($browser, $url, $filePath) {
    echo "=== File Upload ===\n";
    echo "Uploading: $filePath\n";
    
    if (!file_exists($filePath)) {
        echo "File not found: $filePath\n\n";
        return \React\Promise\resolve();
    }
    
    $fileSize = filesize($filePath);
    echo "File size: $fileSize bytes\n";
    
    $file = fopen($filePath, 'r');
    $fileStream = new ReadableResourceStream($file, $browser->getLoop());
    
    $headers = [
        'Content-Type' => 'application/octet-stream',
        'Content-Length' => $fileSize,
        'Authorization' => 'Bearer your-upload-token'
    ];
    
    return $browser->requestStreaming('POST', $url, $headers, $fileStream)
        ->then(
            function (ResponseInterface $response) use ($filePath) {
                echo "Upload response (Status: " . $response->getStatusCode() . ")\n";
                echo "Upload response body: " . $response->getBody() . "\n\n";
                return $response;
            },
            function (Exception $error) use ($filePath) {
                echo "Upload failed for $filePath: " . $error->getMessage() . "\n\n";
            }
        );
}

// 大容量データのチャンク処理
function processLargeDataInChunks($browser) {
    echo "=== Large Data Chunk Processing ===\n";
    
    $processedChunks = 0;
    $buffer = '';
    
    return $browser->requestStreaming('GET', 'https://api.example.com/large-json-array')
        ->then(
            function (ResponseInterface $response) use (&$processedChunks, &$buffer) {
                echo "Starting large data processing\n";
                
                $body = $response->getBody();
                
                $body->on('data', function ($chunk) use (&$processedChunks, &$buffer) {
                    $buffer .= $chunk;
                    
                    // 改行でデータを分割して処理
                    $lines = explode("\n", $buffer);
                    $buffer = array_pop($lines); // 最後の不完全な行はバッファに保持
                    
                    foreach ($lines as $line) {
                        if (trim($line)) {
                            // 各行を JSON として処理
                            $data = json_decode($line, true);
                            if ($data) {
                                $processedChunks++;
                                echo "Processed item $processedChunks: " . ($data['id'] ?? 'unknown') . "\n";
                                
                                // ここで実際のデータ処理を行う
                                // 例: データベースへの保存、変換処理等
                            }
                        }
                    }
                });
                
                $body->on('end', function () use (&$processedChunks, &$buffer) {
                    // 最後の残りデータを処理
                    if (trim($buffer)) {
                        $data = json_decode($buffer, true);
                        if ($data) {
                            $processedChunks++;
                            echo "Processed final item: " . ($data['id'] ?? 'unknown') . "\n";
                        }
                    }
                    
                    echo "Large data processing completed. Total items: $processedChunks\n\n";
                });
                
                return $response;
            }
        );
}

// メイン実行
streamingResponse($browser)
    ->then(function() use ($browser) {
        return downloadFile($browser, 'https://example.com/sample-file.zip', '/tmp/downloaded-file.zip');
    })
    ->then(function() use ($browser) {
        return uploadFile($browser, 'https://api.example.com/upload', '/tmp/test-upload.txt');
    })
    ->then(function() use ($browser) {
        return processLargeDataInChunks($browser);
    })
    ->then(function() use ($loop) {
        echo "All streaming examples completed.\n";
        $loop->stop();
    });

$loop->run();

エラーハンドリングとリトライ機能

<?php
require 'vendor/autoload.php';

use React\EventLoop\Loop;
use React\Http\Browser;
use React\Promise\Timer\TimeoutException;
use Psr\Http\Message\ResponseInterface;

$loop = Loop::get();
$browser = new Browser($loop);

// 包括的なエラーハンドリング
function comprehensiveErrorHandling($browser) {
    echo "=== Comprehensive Error Handling ===\n";
    
    return $browser->get('https://api.example.com/unreliable-endpoint')
        ->then(
            function (ResponseInterface $response) {
                $status = $response->getStatusCode();
                echo "Response received with status: $status\n";
                
                // HTTP ステータスコード別の処理
                if ($status >= 200 && $status < 300) {
                    echo "Success: Request completed successfully\n";
                    return $response;
                } elseif ($status >= 400 && $status < 500) {
                    throw new Exception("Client error: HTTP $status - " . $response->getReasonPhrase());
                } elseif ($status >= 500) {
                    throw new Exception("Server error: HTTP $status - " . $response->getReasonPhrase());
                } else {
                    throw new Exception("Unexpected status code: $status");
                }
            }
        )->otherwise(
            function (Exception $error) {
                echo "Error occurred: " . get_class($error) . "\n";
                echo "Error message: " . $error->getMessage() . "\n";
                
                // エラータイプ別の処理
                if ($error instanceof TimeoutException) {
                    echo "Error type: Request timeout\n";
                } elseif ($error instanceof \React\Socket\ConnectionException) {
                    echo "Error type: Connection error\n";
                } elseif ($error instanceof \InvalidArgumentException) {
                    echo "Error type: Invalid request parameters\n";
                } else {
                    echo "Error type: General error\n";
                }
                
                echo "Error handling completed\n\n";
                throw $error; // 必要に応じて再スロー
            }
        );
}

// リトライ機能付きリクエスト
function requestWithRetry($browser, $url, $maxRetries = 3, $backoffDelay = 1.0) {
    echo "=== Request with Retry Logic ===\n";
    echo "Attempting request: $url\n";
    
    $attempt = 0;
    
    $makeRequest = function() use ($browser, $url, $maxRetries, $backoffDelay, &$attempt, &$makeRequest) {
        $attempt++;
        echo "Attempt $attempt/$maxRetries\n";
        
        return $browser->withTimeout(10.0)->get($url)
            ->then(
                function (ResponseInterface $response) use ($attempt) {
                    $status = $response->getStatusCode();
                    echo "Request successful on attempt $attempt (Status: $status)\n\n";
                    return $response;
                }
            )->otherwise(
                function (Exception $error) use ($attempt, $maxRetries, $backoffDelay, $makeRequest, $url) {
                    echo "Attempt $attempt failed: " . $error->getMessage() . "\n";
                    
                    // リトライ可能なエラーかチェック
                    $shouldRetry = false;
                    
                    if ($error instanceof TimeoutException) {
                        $shouldRetry = true;
                        echo "Timeout error - retryable\n";
                    } elseif ($error instanceof \React\Socket\ConnectionException) {
                        $shouldRetry = true;
                        echo "Connection error - retryable\n";
                    } elseif (method_exists($error, 'getResponse')) {
                        $response = $error->getResponse();
                        if ($response && $response->getStatusCode() >= 500) {
                            $shouldRetry = true;
                            echo "Server error - retryable\n";
                        }
                    }
                    
                    if ($shouldRetry && $attempt < $maxRetries) {
                        $delay = $backoffDelay * pow(2, $attempt - 1); // Exponential backoff
                        echo "Retrying in $delay seconds...\n";
                        
                        return \React\Promise\Timer\sleep($delay, $browser->getLoop())
                            ->then($makeRequest);
                    } else {
                        echo "Max retries reached or non-retryable error\n\n";
                        throw $error;
                    }
                }
            );
    };
    
    return $makeRequest();
}

// サーキットブレーカーパターン
class CircuitBreaker {
    private $failureCount = 0;
    private $lastFailureTime = null;
    private $state = 'CLOSED'; // CLOSED, OPEN, HALF_OPEN
    private $maxFailures;
    private $recoveryTimeout;
    
    public function __construct($maxFailures = 5, $recoveryTimeout = 60) {
        $this->maxFailures = $maxFailures;
        $this->recoveryTimeout = $recoveryTimeout;
    }
    
    public function call($browser, $url) {
        echo "Circuit breaker state: {$this->state}\n";
        
        if ($this->state === 'OPEN') {
            if (time() - $this->lastFailureTime > $this->recoveryTimeout) {
                $this->state = 'HALF_OPEN';
                echo "Circuit breaker transitioning to HALF_OPEN\n";
            } else {
                echo "Circuit breaker is OPEN - request blocked\n\n";
                return \React\Promise\reject(new Exception('Circuit breaker is OPEN'));
            }
        }
        
        return $browser->get($url)
            ->then(
                function (ResponseInterface $response) {
                    // Success - reset failure count
                    $this->failureCount = 0;
                    if ($this->state === 'HALF_OPEN') {
                        $this->state = 'CLOSED';
                        echo "Circuit breaker recovered - state: CLOSED\n";
                    }
                    return $response;
                }
            )->otherwise(
                function (Exception $error) {
                    // Failure - increment counter
                    $this->failureCount++;
                    $this->lastFailureTime = time();
                    
                    echo "Request failed (failure count: {$this->failureCount})\n";
                    
                    if ($this->failureCount >= $this->maxFailures) {
                        $this->state = 'OPEN';
                        echo "Circuit breaker opened due to excessive failures\n";
                    }
                    
                    throw $error;
                }
            );
    }
}

function circuitBreakerExample($browser) {
    echo "=== Circuit Breaker Pattern ===\n";
    
    $circuitBreaker = new CircuitBreaker(3, 30);
    
    // 複数回のリクエストでサーキットブレーカーをテスト
    $promises = [];
    for ($i = 1; $i <= 7; $i++) {
        $promises[] = $circuitBreaker->call($browser, 'https://api.example.com/unstable-endpoint')
            ->then(
                function (ResponseInterface $response) use ($i) {
                    echo "Request $i succeeded\n";
                    return $response;
                },
                function (Exception $error) use ($i) {
                    echo "Request $i failed: " . $error->getMessage() . "\n";
                    return null; // Continue with other requests
                }
            );
    }
    
    return \React\Promise\all($promises)->then(
        function ($results) {
            echo "Circuit breaker test completed\n\n";
            return $results;
        }
    );
}

// メイン実行
comprehensiveErrorHandling($browser)
    ->otherwise(function() { return null; }) // エラーを捕捉して継続
    ->then(function() use ($browser) {
        return requestWithRetry($browser, 'https://api.example.com/sometimes-fails', 3, 1.0);
    })
    ->otherwise(function() { return null; }) // エラーを捕捉して継続
    ->then(function() use ($browser) {
        return circuitBreakerExample($browser);
    })
    ->then(function() use ($loop) {
        echo "All error handling examples completed.\n";
        $loop->stop();
    });

$loop->run();

HTTP サーバーの実装

<?php
require 'vendor/autoload.php';

use React\EventLoop\Loop;
use React\Http\HttpServer;
use React\Http\Message\Response;
use React\Socket\SocketServer;
use Psr\Http\Message\ServerRequestInterface;

$loop = Loop::get();

// HTTP サーバーの作成と設定
function createHttpServer($loop) {
    echo "=== ReactPHP HTTP Server ===\n";
    
    $server = new HttpServer($loop, function (ServerRequestInterface $request) {
        $method = $request->getMethod();
        $path = $request->getUri()->getPath();
        $query = $request->getQueryParams();
        
        echo "Request: $method $path\n";
        
        // ルーティング処理
        switch ($path) {
            case '/':
                return new Response(200, ['Content-Type' => 'application/json'], 
                    json_encode(['message' => 'Welcome to ReactPHP HTTP Server!', 'timestamp' => time()]));
                    
            case '/users':
                if ($method === 'GET') {
                    return handleGetUsers($query);
                } elseif ($method === 'POST') {
                    return handleCreateUser($request);
                }
                break;
                
            case '/health':
                return new Response(200, ['Content-Type' => 'application/json'], 
                    json_encode(['status' => 'healthy', 'uptime' => time()]));
                    
            case '/streaming':
                return handleStreamingResponse();
                
            default:
                if (preg_match('/^\/users\/(\d+)$/', $path, $matches)) {
                    $userId = $matches[1];
                    if ($method === 'GET') {
                        return handleGetUser($userId);
                    } elseif ($method === 'PUT') {
                        return handleUpdateUser($userId, $request);
                    } elseif ($method === 'DELETE') {
                        return handleDeleteUser($userId);
                    }
                }
                break;
        }
        
        // 404 Not Found
        return new Response(404, ['Content-Type' => 'application/json'], 
            json_encode(['error' => 'Not Found', 'path' => $path]));
    });
    
    return $server;
}

// ユーザー一覧の取得
function handleGetUsers($query) {
    $page = $query['page'] ?? 1;
    $limit = $query['limit'] ?? 10;
    
    $users = [
        ['id' => 1, 'name' => '田中太郎', 'email' => '[email protected]'],
        ['id' => 2, 'name' => '佐藤花子', 'email' => '[email protected]'],
        ['id' => 3, 'name' => '鈴木一郎', 'email' => '[email protected]'],
    ];
    
    return new Response(200, ['Content-Type' => 'application/json'], json_encode([
        'users' => array_slice($users, ($page - 1) * $limit, $limit),
        'pagination' => [
            'page' => (int)$page,
            'limit' => (int)$limit,
            'total' => count($users)
        ]
    ]));
}

// 特定ユーザーの取得
function handleGetUser($userId) {
    $users = [
        1 => ['id' => 1, 'name' => '田中太郎', 'email' => '[email protected]', 'created_at' => '2025-01-01T00:00:00Z'],
        2 => ['id' => 2, 'name' => '佐藤花子', 'email' => '[email protected]', 'created_at' => '2025-01-02T00:00:00Z'],
        3 => ['id' => 3, 'name' => '鈴木一郎', 'email' => '[email protected]', 'created_at' => '2025-01-03T00:00:00Z'],
    ];
    
    if (isset($users[$userId])) {
        return new Response(200, ['Content-Type' => 'application/json'], 
            json_encode($users[$userId]));
    } else {
        return new Response(404, ['Content-Type' => 'application/json'], 
            json_encode(['error' => 'User not found', 'user_id' => $userId]));
    }
}

// ユーザーの作成
function handleCreateUser(ServerRequestInterface $request) {
    $body = (string) $request->getBody();
    $data = json_decode($body, true);
    
    if (!$data || !isset($data['name']) || !isset($data['email'])) {
        return new Response(400, ['Content-Type' => 'application/json'], 
            json_encode(['error' => 'Invalid request data', 'required' => ['name', 'email']]));
    }
    
    $newUser = [
        'id' => rand(1000, 9999),
        'name' => $data['name'],
        'email' => $data['email'],
        'created_at' => date('c')
    ];
    
    return new Response(201, ['Content-Type' => 'application/json'], 
        json_encode($newUser));
}

// ユーザーの更新
function handleUpdateUser($userId, ServerRequestInterface $request) {
    $body = (string) $request->getBody();
    $data = json_decode($body, true);
    
    if (!$data) {
        return new Response(400, ['Content-Type' => 'application/json'], 
            json_encode(['error' => 'Invalid JSON data']));
    }
    
    $updatedUser = [
        'id' => (int)$userId,
        'name' => $data['name'] ?? 'Unknown',
        'email' => $data['email'] ?? '[email protected]',
        'updated_at' => date('c')
    ];
    
    return new Response(200, ['Content-Type' => 'application/json'], 
        json_encode($updatedUser));
}

// ユーザーの削除
function handleDeleteUser($userId) {
    // 実際の削除処理はここで行う
    return new Response(204, [], ''); // No Content
}

// ストリーミングレスポンス
function handleStreamingResponse() {
    return new Response(200, ['Content-Type' => 'text/plain'], 
        \React\Stream\ReadableResourceStream::fromIterator([
            "Starting stream...\n",
            "Data chunk 1\n",
            "Data chunk 2\n", 
            "Data chunk 3\n",
            "Stream completed.\n"
        ]));
}

// サーバーの開始
function startServer($loop) {
    $server = createHttpServer($loop);
    
    $socket = new SocketServer('127.0.0.1:8080', $loop);
    $server->listen($socket);
    
    echo "HTTP Server started on http://127.0.0.1:8080\n";
    echo "Available endpoints:\n";
    echo "  GET  / - Welcome message\n";
    echo "  GET  /health - Health check\n";
    echo "  GET  /users - List users\n";
    echo "  POST /users - Create user\n";
    echo "  GET  /users/{id} - Get specific user\n";
    echo "  PUT  /users/{id} - Update user\n";
    echo "  DELETE /users/{id} - Delete user\n";
    echo "  GET  /streaming - Streaming response\n\n";
    echo "Press Ctrl+C to stop the server\n";
    
    return $server;
}

// サーバーの開始
startServer($loop);
$loop->run();