Skip to content

Add Inline queue adapter that runs jobs on enqueue - #146

Draft
eldadfux wants to merge 3 commits into
mainfrom
cursor/inline-queue-adapter-9276
Draft

Add Inline queue adapter that runs jobs on enqueue#146
eldadfux wants to merge 3 commits into
mainfrom
cursor/inline-queue-adapter-9276

Conversation

@eldadfux

@eldadfux eldadfux commented Aug 18, 2026

Copy link
Copy Markdown
Member

What

Adds Adapter\Inline: a queue adapter that is also the publisher. enqueue() runs the matching job() in the same process and returns when the handler finishes. start() arms those handlers and returns — there is no consume loop and no worker process.

Use this when a single process should handle HTTP (or CLI) and its side-effects without a broker.

$adapter = new Inline();
$server = new Server($adapter);
$server->job('v1-mails')->action(function (Message $message) {
    // Runs inside enqueue(), before it returns
});
$server->start();

$adapter->enqueue(new Queue('v1-mails'), ['n' => 1]);

Behavior

  • Failed handlers go through the Server error hook; enqueue() still returns true (same as a broker that accepted the message)
  • Unknown queues (no registered job) are accepted and ignored
  • getQueueSize() is always 0; retry() is a no-op
  • Multi-queue Server::start() rejects a shared Consumer — Inline passes a callable factory so it can register many jobs
  • Nested enqueue (job A enqueues job B) isolates DI context; concurrent Swoole requests keep that stack coroutine-local
  • Uses processFrom() so process() does not read the uninitialized $this->queue property
  • JSON-round-trips payloads before the handler runs (same as Redis/Nats), so handlers always see plain arrays — never live Document / stdClass objects. Empty objects become []. This keeps Appwrite workers adapter-agnostic.

Companion Appwrite change: appwrite/appwrite#13269 (_APP_QUEUE_ADAPTER=inline, no worker containers).

Test Plan

  • vendor/bin/phpunit packages/queue/tests/Queue/Unit/InlineAdapterTest.php (11 tests) on PHP 8.3
  • Full composer test is not run here: other queue unit files use PHP 8.5 syntax this host cannot parse
Open in Web Open in Cursor 

cursoragent and others added 2 commits August 18, 2026 16:38
Add a transport that is also a Publisher: start() arms Server jobs and
returns, and enqueue() invokes the matching handler in the same process.
This lets a single HTTP or CLI process handle side effects without a
broker or worker container.

Co-authored-by: Eldad A. Fux <eldadfux@users.noreply.github.com>
Failed jobs called reject() on Adapter::$queue, which consume() never
binds. Drive processFrom() with the enqueue queue, and assert start()
returns the Server.

Co-authored-by: Eldad A. Fux <eldadfux@users.noreply.github.com>
Handlers must see plain arrays after enqueue, matching Redis/Nats.
Convert Documents, stdClass, and nested objects before processFrom so
Appwrite workers need no Document-vs-array special cases.

Co-authored-by: Eldad A. Fux <eldadfux@users.noreply.github.com>
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants