sohris / reactphp-mysql

0.1.1 2024-08-12 21:43 UTC

Requires

• php: ^8.0
• react/event-loop: ^1.5
• react/promise: ^3.2

Requires (Dev)

• phpunit/phpunit: ^11.3

MIT

• Rodrigo Danieli

Published on Packagist.org on 2024-08-12 21:44 UTC

README

Sohris ReactPHP MySQL is an asynchronous, non-blocking MySQL client and connection pool built on top of ReactPHP and ext-mysqli.

Designed following Clean Code, SOLID, and PSR-12 principles, it allows high-performance event-driven PHP applications to execute SQL queries sequentially on single connections or concurrently across an auto-balancing connection pool.

Features

• Asynchronous Execution: Non-blocking MySQL query execution integrated with the ReactPHP EventLoop.
• Parallel Connection Pool: Automatically distributes and queues queries across multiple parallel connections up to a configurable limit.
• Unified Client Contract: Both single connections (Connection) and pools (Pool) implement ClientInterface, making dependency injection effortless.
• Immutable Value Objects: Type-safe ConnectionConfig (with DSN URI support), Query (with automatic parameter escaping), and QueryResult.
• Testable Architecture: Decoupled low-level DriverInterface allows 100% deterministic unit testing without requiring a live MySQL instance.

Requirements

• PHP >= 8.1
• ext-mysqli
• react/event-loop ^1.5
• react/promise ^3.2

Installation

Install the package via Composer:

composer require sohris/reactphp-mysql

Configuration (ConnectionConfig)

You can configure database credentials using either named arguments on ConnectionConfig or a standard MySQL connection URI:

use Sohris\Mysql\Config\ConnectionConfig;

// 1. Using explicit constructor arguments
$config = new ConnectionConfig(
    host: '127.0.0.1',
    port: 3306,
    user: 'app_user',
    password: 'secret',
    database: 'my_database',
    timeout: 5
);

// 2. Using a MySQL DSN URI
$config = ConnectionConfig::fromUri('mysql://app_user:secret@127.0.0.1:3306/my_database?timeout=5');

Usage

1. Connection Pool (Recommended for Parallel Queries)

Pool manages multiple underlying connections and routes each query to the connection with the shortest queue. Connections are created lazily on demand up to maxSize.

<?php

declare(strict_types=1);

use Sohris\Mysql\Config\ConnectionConfig;
use Sohris\Mysql\Io\QueryResult;
use Sohris\Mysql\Pool;

require __DIR__ . '/vendor/autoload.php';

$config = ConnectionConfig::fromUri('mysql://user:pass@127.0.0.1:3306/mydb');

// Create a pool with up to 5 concurrent connections
$pool = new Pool($config, maxSize: 5);

$pool->query('SELECT id, name FROM users WHERE active = ? LIMIT 10', [true])
    ->then(
        function (QueryResult $result): void {
            foreach ($result->resultRows ?? [] as $row) {
                echo "{$row['id']}: {$row['name']}" . PHP_EOL;
            }
        },
        function (Throwable $exception): void {
            error_log('Query failed: ' . $exception->getMessage());
        }
    );

2. Single Connection (ConnectionFactory)

When you need a single sequential connection, use ConnectionFactory:

<?php

declare(strict_types=1);

use Sohris\Mysql\Config\ConnectionConfig;
use Sohris\Mysql\Connection\ConnectionFactory;
use Sohris\Mysql\Io\QueryResult;

require __DIR__ . '/vendor/autoload.php';

$factory = new ConnectionFactory();
$config = new ConnectionConfig(host: '127.0.0.1', user: 'root', database: 'mydb');

// Lazy connection (connects on first query)
$connection = $factory->createLazy($config);

// Or eager connection (connects immediately)
// $connection = $factory->create($config);

// Or isolated lazy connection (never shared in factory cache)
// $connection = $factory->createNewLazy($config);

$connection->query('INSERT INTO logs (message, created_at) VALUES (?, NOW())', ['App started'])
    ->then(function (QueryResult $result): void {
        echo "Inserted ID: {$result->insertId} (Affected: {$result->affectedRows})" . PHP_EOL;
    });

Parameter Binding & Results

Parameter Binding (Query)

Placeholders (?) are safely interpolated by Query supporting int, float, bool, string, null, Stringable, and array (for IN (?) clauses):

$pool->query(
    'SELECT * FROM orders WHERE status IN (?) AND total >= ? AND deleted_at IS ?',
    [['paid', 'shipped'], 99.95, null]
);

Inspecting QueryResult

Every resolved query promise returns an immutable QueryResult instance:

• $result->resultRows: ?array<int, array<string, mixed>> — Rows returned by SELECT, SHOW, etc. (null for INSERT/UPDATE/DELETE).
• $result->affectedRows: int — Number of affected rows.
• $result->insertId: ?int — Last auto-increment ID generated by an INSERT.
• $result->hasResultSet(): bool — Whether the query produced a row set.
• $result->first(): ?array<string, mixed> — Returns the first row or null.
• $result->count(): int — Number of rows in the result set.

Architecture Overview

src/
├── Config/
│   └── ConnectionConfig.php          # Immutable connection configuration & URI parser
├── Connection/
│   ├── Driver/
│   │   ├── DriverInterface.php        # Low-level driver contract for async MySQL
│   │   ├── DriverFactoryInterface.php # Factory contract for creating drivers
│   │   ├── MysqliDriver.php           # Native ext-mysqli async implementation
│   │   └── MysqliDriverFactory.php    # Default MysqliDriver factory
│   ├── AsyncPoller.php                # ReactPHP EventLoop periodic poller
│   ├── ClientInterface.php            # Unified interface for Connection and Pool
│   ├── Connection.php                 # Single connection & sequential query queue
│   └── ConnectionFactory.php          # Factory for eager, lazy, and isolated connections
├── Exception/
│   ├── MysqlException.php             # Base domain exception
│   ├── ConnectionException.php        # Connection-level errors
│   └── QueryException.php             # Query execution errors (includes SQL context)
├── Io/
│   ├── PendingQuery.php               # Internal queue item pairing Query + Deferred
│   ├── Query.php                      # SQL builder and parameter escaper
│   └── QueryResult.php                # Immutable query result value object
└── Pool.php                           # Load-balanced parallel connection pool

Testing

Run the unit test suite using PHPUnit:

composer test
# or directly:
vendor/bin/phpunit

License

This project is licensed under the MIT License - see the LICENSE file for details.