# kafkalib

A small PHP wrapper around the [php-rdkafka](https://github.com/arnaud-lb/php-rdkafka)
extension for producing and consuming Kafka messages.

## Requirements

- PHP 8.1+ (for the default classes). **For PHP 7.0–7.3 use the `php70/`
  variant instead — see [PHP version support](#php-version-support).**
- The `rdkafka` PHP extension (`pecl install rdkafka` — backed by librdkafka)
- A running Kafka broker

## Install the extension (later)

```bash
sudo apt-get install -y librdkafka-dev   # or: brew install librdkafka
sudo pecl install rdkafka
echo "extension=rdkafka.so" | sudo tee /etc/php/8.4/cli/conf.d/20-rdkafka.ini
php -m | grep rdkafka   # verify
```

## Files

| File               | Purpose                                                        |
|--------------------|---------------------------------------------------------------|
| `KafkaConfig.php`  | Shared broker connection + base librdkafka config             |
| `Producer.php`     | Publish messages to a topic                                   |
| `Consumer.php`     | High-level consumer (group-based) with per-message commit     |
| `autoload.php`     | Bootstrap that requires the three classes (no namespace)       |
| `examples/`        | Runnable produce / multi-group consume / worker scripts        |
| `tests/`           | PHPUnit unit + integration tests                              |
| `php70/`           | PHP 7.0-compatible copies of the classes + examples            |

## Quick start

```php
require_once __DIR__ . '/lib/php/include/kafkalib/autoload.php';

$config = new KafkaConfig('127.0.0.1:9092');

// 1) Produce
(new Producer($config))->produce('orders', ['order_id' => 1001], key: 'order-1001');

// 2) Consume
$consumer = new Consumer($config, 'email-service', ['orders']);
$consumer->consume(function (array $msg) {
    // $msg = ['topic','partition','offset','key','payload','headers']
    handleOrder($msg['payload']);
});
```

## One message → many consumers, different actions

This is a native Kafka behaviour driven by `group.id`:

- **Same `group.id`** → consumers share the work (partitions split between them). Use this to scale one job.
- **Different `group.id`** → each group gets its **own independent copy** of every message, with its own offsets.

So publishing one message to `orders` and running three consumers with
different group ids lets each perform a different action on that same message:

```bash
php examples/consume_multi_group.php email-service     # sends an email
php examples/consume_multi_group.php analytics-service # records a metric
php examples/consume_multi_group.php audit-service     # writes an audit log

php examples/produce.php                               # publish ONE message
```

All three groups receive it. See `tests/IntegrationRoundTripTest.php`
(`testSameMessageDeliveredToMultipleGroups`) for the automated proof.

## PHP version support

The default classes (`KafkaConfig.php`, `Producer.php`, `Consumer.php`) require
**PHP 8.1+** — they use typed properties, `mixed`, `?type` and `void`.

For hosts on **PHP 7.0–7.3**, load the `php70/` variant instead. It contains the
same three classes and the example scripts, rewritten so they parse on PHP 7.0+
(types moved to docblocks, no arrow functions, positional args instead of named
args). Behaviour is identical.

```php
// PHP 8.1+
require_once __DIR__ . '/lib/php/include/kafkalib/autoload.php';

// PHP 7.0 – 7.3
require_once __DIR__ . '/lib/php/include/kafkalib/php70/autoload.php';
```

| PHP version | Use            |
|-------------|----------------|
| 7.0 – 7.3   | `php70/`       |
| 7.4 / 8.0   | either works   |
| 8.1+        | default (root) |

> ⚠️ Load only **one** variant per process — both define the same class names
> (`KafkaConfig` / `Producer` / `Consumer`, no namespace) and would collide.

Runnable 7.0 examples live in `php70/examples/` (`produce.php`,
`consume_multi_group.php`, `worker.php`) and mirror the 8.x ones.

## Running the tests

```bash
cd lib/php/include/kafkalib

# Unit tests only (no broker needed; rdkafka-dependent tests self-skip)
vendor/bin/phpunit

# Include end-to-end tests against a real broker
KAFKA_BROKERS=127.0.0.1:9092 vendor/bin/phpunit
```

## Design notes

- **Manual offset commits** (`enable.auto.commit=false`): the consumer commits
  only *after* your handler returns without throwing, so a crash mid-processing
  re-delivers the message rather than silently dropping it (at-least-once).
- **Payloads**: strings are sent as-is; anything else is `json_encode`d.
- **Graceful shutdown**: `Consumer::stop()` breaks the loop after the current
  message; `close()` leaves the group cleanly so partitions rebalance fast.