Repository navigation
v2.4.0
What's Changed
- Adds support for kafka and boltdb pipelines by @butschster and @msmakouz in #22
Create kafka pipeline
use Spiral\RoadRunner\Jobs\Queue\KafkaCreateInfo;
use Spiral\RoadRunner\Jobs\Jobs;
$jobs = new Jobs(RPC::create('tcp://127.0.0.1:6001'));
$kafkaCreateInfo = new KafkaCreateInfo(
name: 'kafka-queue',
topic: 'subscription',
maxOpenRequests: 1000
);
$queue = $jobs->create($kafkaCreateInfo);Push job into kafka pipeline
use Spiral\RoadRunner\Jobs\KafkaOptions;
use Spiral\RoadRunner\Jobs\Queue\Kafka\PartitionOffset;
$queue->dispatch(
$queue->create(
'job:name',
['foo' => 'bar'],
new KafkaOptions(
topic: 'some-topic',
metadata: 'foo-bar',
offset: PartitionOffset::OFFSET_NEWEST
)
)
);Changing options in a prepared task:
$queue = $jobs->create($kafkaCreateInfo);
$task = $queue->create(
'job:name',
['foo' => 'bar'],
new KafkaOptions(
topic: 'some-topic',
metadata: 'foo-bar',
offset: PartitionOffset::OFFSET_NEWEST
)
);
$queue->dispatch($task->withOptions(new KafkaOptions(
topic: 'other',
metadata: 'foo-bar',
offset: PartitionOffset::OFFSET_NEWEST
)));
$queue->dispatch($task->withOptions($task->getOptions()->withTopic('other')));
$queue->dispatch($task->withDelay(10));Create boltdb pipeline
use Spiral\RoadRunner\Jobs\Queue\BoltdbCreateInfo;
use Spiral\RoadRunner\Jobs\Jobs;
$jobs = new Jobs(RPC::create('tcp://127.0.0.1:6001'));
$boltdbCreateInfo = new BoltdbCreateInfo(
name: 'boltdb-queue',
file: 'rr.db'
);
$queue = $jobs->create($boltdbCreateInfo);Full Changelog: v2.3.2...v2.4.0