Workers in Other Languages

The Milvaion scheduler runs on .NET, but workers don't have to. Besides the .NET worker SDK, Milvaion has worker SDKs for Node.js, Go, Python, Java and PHP.

A worker built with one of them works like a .NET worker:

• It registers itself and appears on the dashboard.
• It receives the jobs scheduled for it.
• It streams its logs live.
• Failed jobs are retried with exponential backoff, then moved to the dead letter queue.
• Jobs can be cancelled from the dashboard and are stopped by timeouts.
• On shutdown it lets running jobs finish and returns the rest to the queue.

The Workers page shows each worker's language next to its name, so a mixed fleet stays readable at a glance. Expand a worker to see its runtime and SDK version.

All of them speak the same worker protocol. That document describes the RabbitMQ queues and message formats, so a worker in any other language can be written against it too.

How a Job Ends​

The rules are the same in every language:

Delivery is at least once. A job whose worker crashes runs again, so make handlers idempotent. Every SDK exposes the occurrence id, which stays the same across retries, as an idempotency key.

Node.js​

npm install @milvasoft/milvaion-worker

import { MilvaionWorker, PermanentJobError } from '@milvasoft/milvaion-worker';

const worker = new MilvaionWorker({
  workerId: 'mail-worker',
  rabbitmq: { url: 'amqp://guest:guest@localhost:5672' },
  redis: { url: 'redis://localhost:6379' },
});

worker.addJob<{ to: string }>('SendEmailJob', async ctx => {
  if (!ctx.data?.to)
    throw new PermanentJobError('Job data needs "to".');

  ctx.log(`Sending to ${ctx.data.to}`);
  await sendMail(ctx.data.to, { signal: ctx.signal }); // aborted on cancel, timeout or shutdown

  return { sent: true };
});

await worker.start();
process.once('SIGTERM', () => worker.stop().then(() => process.exit(0)));

Go​

go get github.com/Milvasoft/milvaion/sdks/go/milvaion

worker, _ := milvaion.NewWorker(milvaion.Options{
	WorkerID:    "mail-worker",
	RabbitMQURL: "amqp://guest:guest@localhost:5672/",
	RedisURL:    "redis://localhost:6379",
})

worker.Handle("SendEmailJob", func(ctx context.Context, job *milvaion.Job) (any, error) {
	var data struct{ To string `json:"to"` }
	if err := job.Bind(&data); err != nil {
		return nil, err
	}
	if data.To == "" {
		return nil, milvaion.Permanentf(`job data needs "to"`)
	}

	job.Logf("Sending to %s", data.To)
	if err := sendMail(ctx, data.To); err != nil { // ctx is cancelled on cancel, timeout or shutdown
		return nil, err
	}
	return map[string]bool{"sent": true}, nil
})

ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
defer stop()
worker.Run(ctx) // returns after a graceful shutdown

Python​

pip install milvaion-worker

from milvaion import JobContext, PermanentJobError, Worker

worker = Worker(worker_id="mail-worker",
                rabbitmq_url="amqp://guest:guest@localhost:5672/",
                redis_url="redis://localhost:6379")


@worker.job("SendEmailJob")
async def send_email(ctx: JobContext) -> dict:
    if not (ctx.data or {}).get("to"):
        raise PermanentJobError('Job data needs "to".')

    ctx.log(f"Sending to {ctx.data['to']}")
    await send_mail(ctx.data["to"])  # cancelled on cancel, timeout or shutdown

    return {"sent": True}


worker.run()  # SIGINT/SIGTERM shut down gracefully

A handler can be async def or a plain function. Plain functions run in a thread and should check ctx.raise_if_aborted() in long loops.

Java​

<dependency>
  <groupId>com.milvasoft</groupId>
  <artifactId>milvaion-worker</artifactId>
  <version>1.0.0</version>
</dependency>

With Gradle: implementation("com.milvasoft:milvaion-worker:1.0.0").

MilvaionWorker worker = new MilvaionWorker(new WorkerOptions()
        .workerId("mail-worker")
        .rabbitMqUri("amqp://guest:guest@localhost:5672/")
        .redisUri("redis://localhost:6379"));

worker.job("SendEmailJob", ctx -> {
    EmailData data = ctx.data(EmailData.class);

    if (data == null || data.to() == null)
        throw new PermanentJobException("Job data needs \"to\".");

    ctx.log("Sending to " + data.to());
    mailer.send(data); // the thread is interrupted on cancel, timeout or shutdown

    return Map.of("sent", true);
});

worker.run(); // SIGTERM or Ctrl+C shut down gracefully

PHP​

composer require milvasoft/milvaion-worker

$worker = new Worker(new WorkerOptions(
    workerId: 'mail-worker',
    rabbitMqUrl: 'amqp://guest:guest@localhost:5672/',
    redisUrl: 'redis://localhost:6379',
));

$worker->job('SendEmailJob', function (JobContext $ctx): array {
    $data = $ctx->data() ?? [];

    if (empty($data['to'])) {
        throw new PermanentJobException('Job data needs "to".');
    }

    $ctx->log("Sending to {$data['to']}");
    $mailer->send($data);

    return ['sent' => true];
});

$worker->run(); // SIGTERM or Ctrl+C shut down gracefully

A PHP process runs one job at a time; run more processes for parallelism. With ext-pcntl (standard on Linux), each job runs in a forked child that the worker kills on cancel or timeout, so heartbeats keep going during long jobs.

Configuration​

Every SDK takes its options in code, or reads them from the same environment variables:

Per job, you can set:

• the number of parallel executions
• the timeout
• retry count and base delay
• a JSON Schema for the job data, which the dashboard turns into the job data form

The job name you register is the name you select in the dashboard when creating the job. It is matched case-insensitively.