SDKs: install and connect

Crystal applications use the JoobQ shards directly (joobq, and joobq_pro for Pro). The other eight languages use a thin SDK in the joobqui repository (sdks/<language>/). None is published to a package registry yet: copy the SDK into your project and reference it by path.

Language	What to copy	How to reference it
Crystal	nothing: `shard.yml` dependencies `joobq` (MIT) and, for Pro, `joobq_pro` (private repository)	`require "joobq"` / `require "joobq_pro"`
Python	`sdks/python/joobq` (standard library only)	put its parent directory on `PYTHONPATH`, `from joobq import ...`
TypeScript	`sdks/typescript/src/index.ts` (Node built-ins)	import it by path
Ruby	`sdks/ruby/lib`	`require_relative ".../lib/joobq"`
Go	`sdks/go` (module `github.com/azutoolkit/joobq-sdk-go`)	`replace github.com/azutoolkit/joobq-sdk-go => ./path/to/copy`
Rust	`sdks/rust` (crate `joobq`)	`joobq = { path = "..." }`
Java	`sdks/java/src` (`io/joobq/JoobQClient.java`, JDK only)	compile it with your sources
PHP	`sdks/php/src/JoobQClient.php` (namespace `JoobQ`)	`require_once`
.NET	`sdks/dotnet/JoobQClient.cs` (namespace `JoobQ`)	copy it into the project folder

Reference the SDK

How a worker in each language references its SDK (the Crystal worker uses the joobq and joobq_pro shards):

crystal

require "joobq"
require "joobq_pro"

python

sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "..", "sdks", "python"))

from joobq import (  # noqa: E402
    JoobQClient, JoobQWorker, NonRetryableError, compensation, gate, input, job, output, retry_policy, update_input, workflow,
)

import {
  type ExecutionContext,
  type TaskMiddleware,
  JoobQClient,
  JoobQWorker,
  NonRetryableError,
  compensation,
  defineJob,
  gate,
  input,
  output,
  retryPolicy,
  updateInput,
  workflow,
} from '../../../sdks/typescript/src/index.ts'

ruby

require_relative "../../../sdks/ruby/lib/joobq"

require github.com/azutoolkit/joobq-sdk-go v0.1.0

replace github.com/azutoolkit/joobq-sdk-go => ../../../sdks/go

toml

joobq = { path = "../../../sdks/rust" }

build) rm -rf classes && javac -d classes ../../../sdks/java/src/main/java/io/joobq/JoobQClient.java Worker.java ;;

php

require_once __DIR__ . '/../../../sdks/php/src/JoobQClient.php';

xml

<ItemGroup>
  <Compile Include="../../../sdks/dotnet/JoobQClient.cs" Link="JoobQClient.cs" />
</ItemGroup>

Tested against a real server in every language shown.

Connect

An SDK needs the server's origin and the SDK token (JOOBQUI_SDK_TOKEN on the server). Every call carries Authorization: Bearer <token>. The examples read them from JOOBQ_GATEWAY_URL and JOOBQ_SDK_TOKEN:

crystal

else
  worker = JoobQPro::Workflows::Worker.new(activity_workers: 8)
  Conf.blueprints.each { |bp| worker.register(bp) }
  worker.start
  registry = JoobQ::WorkerRegistry.new(id: ENV["CONF_WORKER_ID"]? || "conf-crystal")
  registry.start
  spawn { JoobQ.forge(trap_signals: false) }
  stop = Channel(Nil).new
  {Signal::INT, Signal::TERM}.each { |sig| sig.trap { stop.send(nil) } }
  stop.receive
  worker.stop
  JoobQ.shutdown
  registry.stop

python

def main():
    client = JoobQClient(os.environ["JOOBQ_GATEWAY_URL"], os.environ["JOOBQ_SDK_TOKEN"])
    options = {"app_context": App(), "concurrency": 8, "service_name": f"conformance-{LANG}"}
    if "lease_ttl_ms" in inspect.signature(JoobQWorker.__init__).parameters:
        options["lease_ttl_ms"] = int(os.environ.get("CONF_LEASE_TTL_MS", "30000"))
    worker = JoobQWorker(client, os.environ["CONF_WORKER_ID"], [QUEUE, WF_QUEUE], **options)
    for j in (context_job, retry_job, dead_job, exhaust_job, sealed_job, produce_job, slow_job, runat_job,
              produce_runat_job, opt_job, produce_options_job, sched_job, register_sched_job, ratelimit_job,
              uniqueslow_job, produce_unique_job, reenqueue_unique_job, wfclient_job, wfupdate_job, mw_job, mw_skip_job):
        worker.register_job(j)
    for bp in blueprints():
        worker.register_workflow(bp)
    worker.use(m1).use(m2)  # first registered = outermost
    worker.start()

const client = new JoobQClient(env('JOOBQ_GATEWAY_URL'), env('JOOBQ_SDK_TOKEN'))
const worker = new JoobQWorker<App>(client, {
  workerId: env('CONF_WORKER_ID'), taskQueues: [QUEUE, WF_QUEUE], app: new App(), appContext: new App(),
  concurrency: 8, serviceName: `conformance-${LANG}`, leaseTtlMs: Number(process.env.CONF_LEASE_TTL_MS ?? 30000),
} as any)
for (const j of jobs) worker.registerJob(j)
for (const w of workflows) worker.registerWorkflow(w)
worker.use(m1).use(m2) // first registered = outermost
await worker.start()

ruby

client = JoobQ::Client.new(gateway_url: ENV.fetch("JOOBQ_GATEWAY_URL"), token: ENV.fetch("JOOBQ_SDK_TOKEN"))
worker = JoobQ::Worker.new(client: client, worker_id: ENV.fetch("CONF_WORKER_ID"), task_queues: [QUEUE, WF_QUEUE],
                           app: App.new, concurrency: 8, service_name: "conformance-#{LANG}",
                           lease_ttl_ms: ENV.fetch("CONF_LEASE_TTL_MS", "30000").to_i)
[ContextJob, RetryJob, DeadJob, ExhaustJob, SealedJob, ProduceJob, RunatJob, ProduceRunatJob,
 OptJob, ProduceOptionsJob, SchedJob, RegisterSchedJob, RateLimitJob,
 UniqueSlowJob, ProduceUniqueJob, ReenqueueUniqueJob, SlowJob, WfClientJob, WfUpdateJob, MwJob, MwSkipJob].each { |j| worker.register_job(j) }
workflows.each { |w| worker.register_workflow(w) }
worker.use(&M1).use(&M2) # first registered = outermost
worker.start

client := joobq.NewClient(os.Getenv("JOOBQ_GATEWAY_URL"), os.Getenv("JOOBQ_SDK_TOKEN"))
client.ServiceName = "conformance-" + lang
worker := joobq.NewWorker[*App](client, os.Getenv("CONF_WORKER_ID"), []string{queue, wfQueue},
	&App{Marker: os.Getenv("CONF_APP_MARKER"), calls: map[string]int{}, traces: map[string][]string{}})
worker.Concurrency = 8
if ttl, err := strconv.ParseInt(os.Getenv("CONF_LEASE_TTL_MS"), 10, 64); err == nil {
	worker.LeaseTTLMs = ttl
}
for _, j := range jobs {
	worker.RegisterJob(j)
}
for _, w := range workflows {
	worker.RegisterWorkflow(w)
}
worker.Use(m1).Use(m2) // first registered = outermost
if err := worker.Run(context.Background()); err != nil {
	fmt.Fprintln(os.Stderr, err)
}

rust

let mut client = JoobQClient::new(var("JOOBQ_GATEWAY_URL"), var("JOOBQ_SDK_TOKEN"));
client.service_name = Some(format!("conformance-{}", lang));
let mut worker = Worker::new(client, var("CONF_WORKER_ID"), vec![queue.clone(), wf_queue.clone()],
    App { marker: var("CONF_APP_MARKER"), calls: Mutex::new(HashMap::new()), traces: Mutex::new(HashMap::new()) });
worker.concurrency = 8;
worker.lease_ttl_ms = var("CONF_LEASE_TTL_MS").parse().unwrap_or(30_000);
for j in jobs {
    worker = worker.register_job(j);
}
for w in workflows {
    worker = worker.register_workflow(w);
}
worker = worker.use_middleware(m1).use_middleware(m2); // first added = outermost
if let Err(e) = worker.run() {
    eprintln!("{}", e);
}

java

JoobQClient client = new JoobQClient(System.getenv("JOOBQ_GATEWAY_URL"), System.getenv("JOOBQ_SDK_TOKEN"));
client.serviceName = "conformance-" + LANG;
String ttl = System.getenv("CONF_LEASE_TTL_MS");
JoobQWorker<App> worker = new JoobQWorker<>(client, System.getenv("CONF_WORKER_ID"), List.of(QUEUE, WF_QUEUE), new App())
    .concurrency(8).leaseTtlMs(ttl == null ? 30_000 : Long.parseLong(ttl));
jobs.forEach(worker::register);
workflows.forEach(worker::register);
worker.use(M1).use(M2); // first registered = outermost
worker.run();

php

$client = new JoobQClient((string)getenv('JOOBQ_GATEWAY_URL'), (string)getenv('JOOBQ_SDK_TOKEN'));
$worker = new JoobQWorker($client, (string)getenv('CONF_WORKER_ID'), [QUEUE, WF_QUEUE], new App(),
    leaseTtlMs: (int)(getenv('CONF_LEASE_TTL_MS') ?: 30000), serviceName: 'conformance-' . LANG, concurrency: 4,
    appFactory: fn () => new App());
foreach ([ContextJob::class, RetryJob::class, DeadJob::class, ExhaustJob::class, SealedJob::class, ProduceJob::class, RunAtJob::class, ProduceRunAtJob::class, OptJob::class, ProduceOptionsJob::class, SchedJob::class, RegisterSchedJob::class, RateLimitJob::class, UniqueSlowJob::class, ProduceUniqueJob::class, ReenqueueUniqueJob::class, SlowJob::class, WfClientJob::class, WfUpdateJob::class, MwJob::class, MwSkipJob::class] as $job) {
    $worker->registerJob($job);
}
foreach ($workflows as $w) {
    $worker->registerWorkflow($w);
}
$worker->use('m1')->use('m2'); // first registered = outermost
$worker->run();

csharp

var client = new JoobQClient(Env("JOOBQ_GATEWAY_URL"), Env("JOOBQ_SDK_TOKEN")) { ServiceName = $"conformance-{Lang}" };
var worker = new JoobQWorker<App>(client, Env("CONF_WORKER_ID"), new[] { Queue, WfQueue }, new App())
{
    Concurrency = 8,
    LeaseTtlMs = long.TryParse(Env("CONF_LEASE_TTL_MS"), out var ttl) ? ttl : 30000,
};
foreach (var j in jobs) worker.Register(j);
foreach (var w in workflows) worker.Register(w);
worker.Use(M1).Use(M2); // first added = outermost
await worker.RunAsync();

Tested against a real server in every language shown.

Register each job and workflow before starting: register_job adds a job (its queue joins the worker's queues), register_workflow a workflow blueprint (its task queue joins them, and its steps' handlers are registered), and handle_job(job_type, handler) adds a bare job handler by name (Python names; the other SDKs have the same calls in their own style, as the tabs show). On start the worker registers its blueprints (POST /sdk/blueprints), then itself (POST /sdk/workers/register), and runs up to concurrency tasks at once. It registers again every 10 seconds: that is its heartbeat on the Workers page. At shutdown it deregisters (POST /sdk/workers/:id/deregister).

How a worker runs

No handler waits for HTTP: the SDK's transport makes every gateway call beside the handlers.

Drawing the diagram…

The poller long-polls POST /sdk/tasks/poll (timeout_ms 5000) for as many tasks as fit in concurrency + prefetch (prefetch defaults to max(1, concurrency / 2)). A poll waiting on an empty queue is woken through Redis pub/sub the moment a task lands. Long-poll is the only delivery: delivery_mode: "push" is refused with 422.
The acker sends outcomes with POST /sdk/tasks/ack, up to 100 per call, once 100 are queued or 5 ms after the first. The outcome queue holds 4 x concurrency; when it is full, handlers wait (back-pressure).
The heartbeater renews every lease the worker holds with one POST /sdk/tasks/heartbeat every third of the lease (lease_ttl_ms, 30 s by default).
A lease that is not renewed expires: the task is queued again with one retry used (a job with none left goes to the dead letter queue as LeaseExpired). Delivery is at least once, so make handlers idempotent.
The single-task routes POST /sdk/tasks/:token/heartbeat, POST /sdk/tasks/:token/complete and POST /sdk/tasks/:token/fail stay, for tools and for code outside a running worker's transport; a handler's own ctx.heartbeat() uses the first for a step with heartbeat_ms.
When the gateway cannot be reached or answers 5xx, the SDK waits 250 ms, doubling up to 5 s, and goes on.
Shutdown: SIGTERM or SIGINT (or stop()) stops polling, runs what it holds, sends the outcomes and deregisters, within 30 s by default.

Python, Ruby and Rust use threads, Go goroutines, Java threads, .NET tasks, TypeScript async loops. PHP forks: with ext-pcntl the worker is a parent that makes the gateway calls and a pool of concurrency processes that run one task at a time (open connections lazily or in appFactory, since each process has its own copy of the app); without ext-pcntl it runs one task at a time.

Every SDK follows the same rules, and every language is tested against a real server.

SDKs: install and connect ​

Reference the SDK ​

Connect ​

How a worker runs ​

SDKs: install and connect

Reference the SDK

Connect

How a worker runs