Deprecation notice

This library has been deprecated in favour of php-kafka-lib.
It will still receive bugfixes until end of 10/2020, after that it will be marked as abandoned on packagist.
Please check MIGRATION.md for the migration path.

messaging-lib

Description

Generic php messaging library Supports:

• Kafka

This is a convenience wrapper for arnaud-lb/php-rdkafka
Avro support relies on flix-tech/avro-serde-php
To read more about the functions used in this lib, check out the documentation of the extension.

Requirements

• php: ^7.1
• ext-rdkafka: ^4.0.0

Installation

composer require jobcloud/messaging-lib "~4.0"

Usage

Producer

Kafka

Simple example

<?phpuseJobcloud\Messaging\Kafka\Message\KafkaProducerMessage;
use \Jobcloud\Messaging\Kafka\Producer\KafkaProducerBuilder;
$producer = KafkaProducerBuilder::create()
->withAdditionalBroker('localhost:9092')
->build();
$message = KafkaProducerMessage::create('test-topic', 0)
->withKey('asdf-asdf-asfd-asdf')
->withBody('some test message payload')
->withHeaders([ 'key' => 'value' ]);
$producer->produce($message);

Avro Producer

To create an avro prodcuer add the avro encoder.

<?phpuseFlixTech\AvroSerializer\Objects\RecordSerializer;
useJobcloud\Messaging\Kafka\Message\KafkaProducerMessage;
useJobcloud\Messaging\Kafka\Message\Encoder\AvroEncoder;
useJobcloud\Messaging\Kafka\Message\Registry\AvroSchemaRegistry;
use \Jobcloud\Messaging\Kafka\Producer\KafkaProducerBuilder;
use \Jobcloud\Messaging\Kafka\Message\KafkaAvroSchema;
useFlixTech\SchemaRegistryApi\Registry\CachedRegistry;
useFlixTech\SchemaRegistryApi\Registry\BlockingRegistry;
useFlixTech\SchemaRegistryApi\Registry\PromisingRegistry;
useFlixTech\SchemaRegistryApi\Registry\Cache\AvroObjectCacheAdapter;
useGuzzleHttp\Client;
$cachedRegistry = newCachedRegistry(
newBlockingRegistry(
newPromisingRegistry(
newClient(['base_uri' => 'jobcloud-kafka-schema-registry:9081'])
)
),
newAvroObjectCacheAdapter()
);
$registry = newAvroSchemaRegistry($cachedRegistry);
$recordSerializer = newRecordSerializer($cachedRegistry);
//if no version is defined, latest version will be used//if no schema definition is defined, the appropriate version will be fetched form the registry$registry->addSchemaMappingForTopic(
'test-topic',
newKafkaAvroSchema('schemaName'/*, int $version, AvroSchema $definition */)
);
$encoder = newAvroEncoder($registry, $recordSerializer);
$producer = KafkaProducerBuilder::create()
->withAdditionalBroker('kafka:9092')
->withEncoder($encoder)
->build();
$schemaName = 'testSchema';
$version = 1;
$message = KafkaProducerMessage::create('test-topic', 0)
->withKey('asdf-asdf-asfd-asdf')
->withBody(['name' => 'someName'])
->withHeaders([ 'key' => 'value' ]);
$producer->produce($message);

NOTE: To improve producer latency you can install the pcntl extension.
The messaging-lib already has code in place, similarly described here:
https://github.com/arnaud-lb/php-rdkafka#performance--low-latency-settings

Consumer

Kafka High Level

<?phpuse \Jobcloud\Messaging\Consumer\ConsumerException;
use \Jobcloud\Messaging\Kafka\Consumer\KafkaConsumerBuilder;
useJobcloud\Messaging\Kafka\Exception\KafkaConsumerEndOfPartitionException;
useJobcloud\Messaging\Kafka\Exception\KafkaConsumerTimeoutException;
$consumer = KafkaConsumerBuilder::create()
->withAdditionalConfig(
[
'compression.codec' => 'lz4',
'auto.commit.interval.ms' => 500
]
)
->withAdditionalBroker('kafka:9092')
->withConsumerGroup('testGroup')
->withTimeout(120 * 10000)
->withAdditionalSubscription('test-topic')
->build();
$consumer->subscribe();
while (true) {
try {
$message = $consumer->consume();
// your business logic$consumer->commit($message);
} catch (KafkaConsumerTimeoutException$e) {
//no messages were read in a given time
} catch (KafkaConsumerEndOfPartitionException$e) {
//only occurs if enable.partition.eof is true (default: false)
} catch (ConsumerException$e) {
// Failed
} }

Kafka Low Level

<?phpuse \Jobcloud\Messaging\Consumer\ConsumerException;
use \Jobcloud\Messaging\Kafka\Consumer\KafkaConsumerBuilder;
useJobcloud\Messaging\Kafka\Exception\KafkaConsumerEndOfPartitionException;
useJobcloud\Messaging\Kafka\Exception\KafkaConsumerTimeoutException;
$consumer = KafkaConsumerBuilder::create()
->withAdditionalConfig(
[
'compression.codec' => 'lz4',
'auto.commit.interval.ms' => 500
]
)
->withAdditionalBroker('kafka:9092')
->withConsumerGroup('testGroup')
->withTimeout(120 * 10000)
->withAdditionalSubscription('test-topic')
->withConsumerType(KafkaConsumerBuilder::CONSUMER_TYPE_LOW_LEVEL)
->build();
$consumer->subscribe();
while (true) {
try {
$message = $consumer->consume();
// your business logic$consumer->commit($message);
} catch (KafkaConsumerTimeoutException$e) {
//no messages were read in a given time
} catch (KafkaConsumerEndOfPartitionException$e) {
//only occurs if enable.partition.eof is true (default: false)
} catch (ConsumerException$e) {
// Failed
} }

Avro Consumer

To create an avro consumer add the avro decoder.

<?phpuseFlixTech\AvroSerializer\Objects\RecordSerializer;
useJobcloud\Messaging\Consumer\ConsumerException;
use \Jobcloud\Messaging\Kafka\Consumer\KafkaConsumerBuilder;
useJobcloud\Messaging\Kafka\Exception\KafkaConsumerEndOfPartitionException;
useJobcloud\Messaging\Kafka\Exception\KafkaConsumerTimeoutException;
useJobcloud\Messaging\Kafka\Message\Decoder\AvroDecoder;
useJobcloud\Messaging\Kafka\Message\KafkaAvroSchema;
useJobcloud\Messaging\Kafka\Message\Registry\AvroSchemaRegistry;
useFlixTech\SchemaRegistryApi\Registry\CachedRegistry;
useFlixTech\SchemaRegistryApi\Registry\BlockingRegistry;
useFlixTech\SchemaRegistryApi\Registry\PromisingRegistry;
useFlixTech\SchemaRegistryApi\Registry\Cache\AvroObjectCacheAdapter;
useGuzzleHttp\Client;
$cachedRegistry = newCachedRegistry(
newBlockingRegistry(
newPromisingRegistry(
newClient(['base_uri' => 'jobcloud-kafka-schema-registry:9081'])
)
),
newAvroObjectCacheAdapter()
);
$registry = newAvroSchemaRegistry($cachedRegistry);
$recordSerializer = newRecordSerializer($cachedRegistry);
//if no version is defined, latest version will be used//if no schema definition is defined, the appropriate version will be fetched form the registry$registry->addSchemaMappingForTopic(
'test-topic',
newKafkaAvroSchema('someSchema' , 9/* , AvroSchema $definition */)
);
$decoder = newAvroDecoder($registry, $recordSerializer);
$consumer = KafkaConsumerBuilder::create()
->withAdditionalConfig(
[
'compression.codec' => 'lz4',
'auto.commit.interval.ms' => 500
]
)
->withDecoder($decoder)
->withAdditionalBroker('kafka:9092')
->withConsumerGroup('testGroup')
->withTimeout(120 * 10000)
->withAdditionalSubscription('test-topic')
->build();
$consumer->subscribe();
while (true) {
try {
$message = $consumer->consume();
// your business logic$consumer->commit($message);
} catch (KafkaConsumerTimeoutException$e) {
//no messages were read in a given time
} catch (KafkaConsumerEndOfPartitionException$e) {
//only occurs if enable.partition.eof is true (default: false)
} catch (ConsumerException$e) {
// Failed
} }

ProducerPool

<?phpuse \Jobcloud\Messaging\Producer\ProducerPool;
use \Jobcloud\Messaging\Producer\ProducerInterface;
/** @var ProducerInterface $someKafkaProducer *//** @var ProducerInterface $someRabbitMQProducer */$pool = newProducerPool();
$pool
->addProducer($someKafkaProducer)
->addProducer($someRabbitMQProducer)
;
$message = KafkaMessage::create('test-topic', 0)
->withKey('asdf-asdf-asfd-asdf')
->withBody('some test content')
->withHeaders([ 'key' => 'value' ]);
$pool->produce($message);