rx 0.0.3

Reactive Dart #

Pub Package Build Status Coverage Status GitHub Issues GitHub Forks GitHub Stars GitHub License

Reactive Extensions Library for Dart providing an API for asynchronous programming with observable streams.

Reactive is created in the style of ReactiveX, and does not depend on Dart Streams. At this point this code is quite experimental, and might miss features or have serious bugs.

This library is open source and well tested. Development happens on GitHub. Feel free to report issues or create a pull-request there. General questions are best asked on StackOverflow.

The package is hosted on dart packages. Up-to-date class documentation is created with every release.

Currently the Dart programming language (or my inability to use it correctly) is blocking a more pleasant API in a few places:

  • The lack of extension methods in Dart makes it awkward to access operator functions and constructor methods: dart-lang/language#40, dart-lang/language#41, dart-lang/language#42, dart-lang/language#177, dart-lang/language#309, dart-lang/language#8547.
  • For some reason type inference over chained operators does not work. For example observable.pipe2(filter(()) => ...), map(() => ...)) is unable to correctly infer the types in the second operator, while observable.pipe(filter() => ...).pipe(map(() => ...) does.
  • The lack of variable length arguments and generics requires duplicated functions and typedefs with a number suffix (see above).

License #

The MIT License, see LICENSE.

0.0.1 #

  • Initial version.

0.0.2 #

  • Operator and composition basics.

0.0.3 #

  • Subject and multicast basics.

example/example.dart

library rx.example.example;

import 'package:more/collection.dart';
import 'package:rx/operators.dart' as ops;
import 'package:rx/rx.dart' as rx;

rx.Observer<T> printObserver<T>(String name) => rx.Observer(
      next: (value) => print('$name.next($value)'),
      error: (error, [stackTrace]) => print('$name.error($error)'),
      complete: () => print('$name.complete()'),
    );

void main() {
  // create
  final create = rx.create((subscriber) {
    for (var i = 0; i < 3; i++) {
      subscriber.next(i);
    }
    subscriber.complete();
  });
  create.subscribe(printObserver('create'));

  // empty
  final empty = rx.empty();
  empty.subscribe(printObserver('empty'));

  // future
  final fromFuture = rx.fromFuture(Future.value(42));
  fromFuture.subscribe(printObserver('fromFuture'));

  final toFuture = rx.toFuture(rx.fromIterable([1, 2, 3]));
  toFuture.then((value) => print('toFuture.then($value)'));

  // just
  final just = rx.just(42);
  just.subscribe(printObserver('just'));

  // never
  final never = rx.never();
  never.subscribe(printObserver('never'));

  // stream
  final fromStream = rx.fromStream(Stream.fromIterable([1, 2, 3]));
  fromStream.subscribe(printObserver('fromStream'));

  final toStream = rx.toStream(rx.fromIterable([1, 2, 3]));
  toStream.listen((value) => print('toStream.value($value)'));

  // throw
  final throwError = rx.throwError(Exception('Hello World'));
  throwError.subscribe(printObserver('throw'));

  // double subscription
  final transformed = rx
      .fromIterable(IntegerRange(0, 100))
      .pipe(ops.filter((value) => value.isEven))
      .pipe(ops.map((value) => '{value * value}'))
      .pipe(ops.filter((value) => value.length < 3));
  transformed.subscribe(printObserver('one'));
  transformed.subscribe(printObserver('two'));

  // subject subscription
  final subject = rx
      .fromIterable(IntegerRange(0, 100, 25))
      .pipe(ops.publishReplay())
      .pipe(ops.refCount());
  subject.subscribe(printObserver('subject1'));
  subject.subscribe(printObserver('subject2'));

  // timer
  final obs = rx.timer(
      delay: const Duration(seconds: 2),
      period: const Duration(milliseconds: 500));
  final subs1 = obs.subscribe(printObserver('first'));
  final subs2 = obs.subscribe(printObserver('second'));
  rx
      .timer(delay: const Duration(seconds: 3))
      .subscribe(rx.Observer(complete: () => subs1.unsubscribe()));
  rx
      .timer(delay: const Duration(seconds: 5))
      .subscribe(rx.Observer(complete: () => subs2.unsubscribe()));
}

Use this package as a library

1. Depend on it

Add this to your package's pubspec.yaml file:


dependencies:
  rx: ^0.0.3

2. Install it

You can install packages from the command line:

with pub:


$ pub get

with Flutter:


$ flutter pub get

Alternatively, your editor might support pub get or flutter pub get. Check the docs for your editor to learn more.

3. Import it

Now in your Dart code, you can use:


import 'package:rx/rx.dart';
  
Popularity:
Describes how popular the package is relative to other packages. [more]
30
Health:
Code health derived from static analysis. [more]
100
Maintenance:
Reflects how tidy and up-to-date the package is. [more]
90
Overall:
Weighted score of the above. [more]
63
Learn more about scoring.

We analyzed this package on Aug 21, 2019, and provided a score, details, and suggestions below. Analysis was completed with status completed using:

  • Dart: 2.4.0
  • pana: 0.12.19

Platforms

Detected platforms: Flutter, web, other

No platform restriction found in primary library package:rx/rx.dart.

Maintenance suggestions

Package is pre-v0.1 release. (-10 points)

While nothing is inherently wrong with versions of 0.0.*, it might mean that the author is still experimenting with the general direction of the API.

Dependencies

Package Constraint Resolved Available
Direct dependencies
Dart SDK >=2.4.0 <3.0.0
collection ^1.14.0 1.14.12
meta ^1.1.0 1.1.7
more ^1.18.0 1.18.0
Dev dependencies
test ^1.6.0