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', handler: print));

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

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.