Если таска завершается по ексепшену то попадаем в вечный цикл

даже если в методе fail передаем $requeue = false

"spiral/roadrunner": "v2.4.1",
"spiral/roadrunner-jobs": "v2.0.0"

Hello! I fill config for dispatching to RabbitMQ's exchange, but at startup always create queue with name "default". These messages are intended for other microservices, whoever wants, subscribes to the exchange. How to stop the queue from being created?

amqp:
  addr: amqp://guest:guest@rabbitmq:5672/

jobs:
  pool:
    num_workers: 1
    max_jobs: 0
    allocate_timeout: 600s
    destroy_timeout: 600s

  pipelines:
    aggregator-event:
      config:
        driver: amqp
        exchange: amqp.topic
        exchange_type: topic
        routing_key: aggregator-event.user.action
        durable: true

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);

i can't read tasks from $jobs->create created queue

worker code:

<?php

include realpath(__DIR__ . "/../bootstrap.php");

use Spiral\Goridge\RPC\RPC;
use Spiral\Goridge\StreamRelay;
use Spiral\RoadRunner;
use Spiral\RoadRunner\Environment;

ini_set('display_errors', 0);
error_reporting(0);

$env = Environment::fromGlobals();
$rpc = RPC::create($env->getRPCAddress());

$jobs = new RoadRunner\Jobs\Jobs($rpc);

$p = $jobs->create(new RoadRunner\Jobs\Queue\AMQPCreateInfo(
    'myqueue',
    1,
    1,
    'myqueue',
    'amqp.default1',
));
$p->resume();

$consumer = new RoadRunner\Jobs\Consumer();

while ($task = $consumer->waitTask()) {
    echo PHP_EOL . "taskk";

    $task->complete();
}

rr.yaml config:

rpc:
  listen: tcp://127.0.0.1:6001

server:
  command: "php workers/bot_queue_worker.php"
  relay: pipes

logs:
  level: debug

amqp:
  addr: amqp://guest:guest@rabbitmq:5672

jobs:
  consume: ["pipe1", "pipe2"]
  num_pollers: 16
  pipeline_size: 100000
  timeout: 1
  pool:
    num_workers: 10
    max_jobs: 0
    allocate_timeout: 60s
    destroy_timeout: 60s

RR2 in the v2.5.0 will bring NATS support: https://docs.nats.io/ support. Here's what needs to be done for the PHP client library:

1. Add new NATS const (or what is used in PHP to identify the jobs driver). ~2. Support the new protocol response type for all brokers. So, not only NATS would have the ability to respond.~
2. For the response protocol type, the data should be stored in the data section within the payload JSON key.

The response itself is fairly simple:

{
  "type": 2,
  "data": {
    "queue": "foo",
    "payload": "binary_payload"
  }
}

Where queue is used as a generic term to name the driver push abstraction. So, the queue is the synonym for the tube for the beanstalk or subject for the NATS. payload is a binary response that should be sent to the queue.

Limitations:

• NATS doesn't support the delay. The client library is responsible for rejecting a request which contains delay. The server part will double check that and in case of non-zero delay will return an error.

Tasks cloning and responding destructor side effects.

Currently it seems to be impossible to setup retries using the docs recommended way(via headers). We are working with \Spiral\RoadRunner\Jobs\Task\ReceivedTask objects. Once you call $task->withHeader('retries', $x) to refresh the retries left, a copy of your task is created. Now, take a look at __destruct():

public function __destruct()
{
    try {
        if ($this->completed === null) {
            $this->complete();
        }
    } catch (JobsException $e) {
        // Suppress shutdown exception
    }
}

And $this->complete() call:

public function complete(): void
{
    $this->respond(Type::SUCCESS);
}

Back to our code. Copy of a task means we will have 2 respond() calls, one from originalTask::destruct() and one from our retry logic. $task->fail('error message', $resend=true);

Even if you assign both task copies to delay the destruct() call of the original one: $taskCopy = $task->withHeader('attempts', (string) $retries);

$taskCopy->fail('error message', $resend=true); One of these will be called $task->complete() from __destruct() or $task->fail($x, $y) manually. And it will cause problems, breaking our resend flow.

Added headers to the Options and PreparedTask.

Parameter array $headers = [] in the PreparedTask similar to the same parameter in the other tasks:

1. https://github.com/spiral/roadrunner-jobs/blob/master/src/Task/QueuedTask.php#L37
2. https://github.com/spiral/roadrunner-jobs/blob/master/src/Task/ReceivedTask.php#L27

Hi,

This PR avoids an exception when a message sent into a kafka topic has an empty body. I also tested a message with an empty key (but with a valid body) and it works fine.

See: https://github.com/roadrunner-server/roadrunner/issues/1311

Thank you

closes: https://github.com/roadrunner-server/roadrunner/issues/1311

Since RoadRunner version 2.3 there is not an ability to get list of available plugins via RPC and every time when developers started using this package they bumped into a problem with Jobs::isAvailable method. With this PR the method was marked as a deprecated and now it will throw an exception with information about deprecation.

At the moment a queue sets priority for every pushed task to 10 and there is a problem with it. If the priority value for these tasks was explicitly set, default priority from RoadRunner config can not be used. If the priority value is 0, then will be used a default value from RoadRunner config.

| Q | A | ------------- | --- | Bugfix? | ❌ | Breaks BC? | ✔️ | New feature? | ❌

• Removed opis/closure dependency.
• Removed DefaultSerializer with opis/closure.
• Replaced DefaultSerializer to JsonSerializer

Hello, I see that this repo has the following version 2.6, 3.0. I suppose the latest version 3.0 but the latest commits were in 2.6. do you follow semantics of version?