stem_redis

Redis Streams broker, result backend, and scheduler utilities for the Stem task runtime.

Install

dart pub add stem_redis

Add the core runtime if you haven't already:

dart pub add stem

Usage

Direct enqueue

import 'package:stem/stem.dart';
import 'package:stem_redis/stem_redis.dart';

Future<void> main() async {
  final registry = SimpleTaskRegistry()
    ..register(
      FunctionTaskHandler(
        name: 'demo.hello',
        entrypoint: (context, args) async {
          print('Hello ${(args['name'] as String?) ?? 'world'}');
        },
      ),
    );

  final broker = await RedisStreamsBroker.connect('redis://localhost:6379');
  final backend = await RedisResultBackend.connect('redis://localhost:6379/1');

  final stem = Stem(broker: broker, registry: registry, backend: backend);
  await stem.enqueue('demo.hello', args: {'name': 'Stem'});
}

Typed TaskDefinition

import 'package:stem/stem.dart';
import 'package:stem_redis/stem_redis.dart';

final helloTask = TaskDefinition<HelloArgs, void>(
  name: 'demo.hello',
  encodeArgs: (args) => {'name': args.name},
  metadata: TaskMetadata(description: 'Redis backed hello world'),
);

class HelloArgs {
  const HelloArgs({required this.name});
  final String name;
}

Future<void> main() async {
  final registry = SimpleTaskRegistry()
    ..register(
      FunctionTaskHandler<void>(
        name: helloTask.name,
        entrypoint: (context, args) async {
          print('Hello ${(args['name'] as String?) ?? 'world'}');
        },
        metadata: helloTask.metadata,
      ),
    );

  final broker = await RedisStreamsBroker.connect('redis://localhost:6379');
  final backend = await RedisResultBackend.connect('redis://localhost:6379/1');

  final stem = Stem(broker: broker, registry: registry, backend: backend);
  await stem.enqueueCall(helloTask(const HelloArgs(name: 'Stem')));
}

Distributed rate limiting

RedisRateLimiter shares a token bucket across worker processes. The RateLimit capacity is the number of permits available per interval; each acquire consumes one permit.

final limiter = await RedisRateLimiter.connect(
  'rediss://localhost:6380/0',
  namespace: 'billing-worker',
  tls: const TlsConfig(
    caCertificateFile: '/etc/ssl/certs/redis-ca.pem',
  ),
);

final workerConfig = StemWorkerConfig(rateLimiter: limiter);

Redis server time is used for refill calculations, and the Lua script performs refill and acquisition atomically. A denied acquisition includes retryAfter for the worker's retry scheduling. Close the limiter with the worker's other resources.

Tests

Integration suites require the dockerised Redis/Postgres stack provided by the CLI package:

source ../../stem_cli/_init_test_env
dart test

The redis contract tests will automatically skip if STEM_TEST_REDIS_URL is not set.