RxDart In Flutter

Reactive functional programming

  • Stream
  • Streams represent flux of data and events.
  • Streams, you can listen to data and event changes, and just as well, deal with what’s coming from the Stream with listeners.
  • StreamBuilder that builds itself based on the latest snapshot of interaction with a Stream, and when there’s a new flux of data the Widget reload to deal with the new data. 
  • Sink used to input.
  • Pipe one end is the entrance and the other end is the exit.

Two types of Stream

  1. Single-subscription Streams
  2. Broadcast Streams
  • Observable<T> 
  • It is extends Stream.
  • It combines all the Streams and StreamTransformers contained.
  • Observableof the Subject, in other words, the object which will be used to notify the Widgets when changes happen in the Stream.
  • Streams vs Observables
  • Observables work with any API and Stream as an input.
  • StreamController
  • It is captures the latest item that has been added to the controller, and emits that as the first item to any new listener.
  • Subject
  • Subject allows sending data, error and done events to the listener. The latest item that has been added to the subject will be sent to any new listeners of the subject. After that, any new events will be appropriately sent to the listeners. It is possible to provide a seed value that will be emitted if no items have been added to the subject.
  • BehaviorSubject 
  • Subject’s stream can be listened to multiple times.
  • Every new subcriber receives the last item.
  • PublishSubject
  • push textChanged-events
  • ReplaySubject
  • Items are added to the subject, the ReplaySubject will store them and when the stream is listened to, those recorded items will be emitted to the listener.

Screenshot :

RxDart In Flutter

1. main.dart

import 'package:flutter/material.dart';
import 'counter_bloc.dart';

void main() => runApp(new MyApp());

class MyApp extends StatelessWidget {
  @override
  Widget build(BuildContext context) {
    return new MaterialApp(
      title: 'Flutter Demo',
      theme: new ThemeData(
        primarySwatch: Colors.blue,
      ),
      home: new MyHomePage(title: 'Flutter Demo Home Page'),
    );
  }
}

class MyHomePage extends StatefulWidget {
  MyHomePage({Key key, this.title}) : super(key: key);

  final String title;

  @override
  _MyHomePageState createState() => new _MyHomePageState();
}

class _MyHomePageState extends State<MyHomePage> {
  CounterBloc _counterBloc = new CounterBloc(initialCount: 0); //dispose must

  @override
  Widget build(BuildContext context) {
    return new Scaffold(
        appBar: new AppBar(),
        body: new Center(
          child: new Column(
            mainAxisAlignment: MainAxisAlignment.center,
            children: <Widget>[
              new Text('RxDart in flutter'),
              new StreamBuilder(
                  stream: _counterBloc.counterObservable,
                  builder: (context, AsyncSnapshot<int> snapshot) {
                    return new Text('${snapshot.data}',
                        style: Theme.of(context).textTheme.display1);
                  })
            ],
          ),
        ),
        floatingActionButton: new Column(
            mainAxisAlignment: MainAxisAlignment.end,
            children: <Widget>[
              new Padding(
                  padding: EdgeInsets.only(bottom: 10),
                  child: new FloatingActionButton(
                    onPressed: _counterBloc.increment,
                    tooltip: 'Increment',
                    child: new Icon(Icons.add),
                  )),
              new FloatingActionButton(
                onPressed: _counterBloc.decrement,
                tooltip: 'Decrement',
                child: new Icon(Icons.remove),
              ),
            ]));
  }

  @override
  void dispose() {
    _counterBloc.dispose();
    super.dispose();
  }
}

 

 

2. counter_bloc.dart

import 'package:rxdart/rxdart.dart';

class CounterBloc {
  int initialCount = 0;
  BehaviorSubject<int> _subjectCounter;

  CounterBloc({this.initialCount}) {
    _subjectCounter = new BehaviorSubject<int>.seeded(this.initialCount);
  }

  Observable<int> get counterObservable => _subjectCounter.stream;

  void increment() {
    initialCount++;
    _subjectCounter.sink.add(initialCount);
  }

  void decrement() {
    initialCount--;
    _subjectCounter.sink.add(initialCount);
  }

  void dispose() {
    _subjectCounter.close();
  }
}

 

 

 

3.pubspec.yaml

name: welcome
description: A new Flutter application.

version: 1.0.0+1

environment:
  sdk: ">=2.1.0 <3.0.0"

dependencies:
  flutter:
    sdk: flutter

  cupertino_icons: ^0.1.2
  rxdart:

dev_dependencies:
  flutter_test:
    sdk: flutter

flutter:
  uses-material-design: true