StreamController from dart:async

main.dart

import 'dart:async';

void addLessThanFive(StreamController controller, int value) {
  if (value < 5) {
    controller.sink.add(value);
  } else {
    controller.sink.addError(StateError('$value is not less than 5'));
  }
}

void main() {
  final controller = StreamController();
  
  addLessThanFive(controller, 1);
  addLessThanFive(controller, 2);
  addLessThanFive(controller, 3);
  addLessThanFive(controller, 4);
  addLessThanFive(controller, 5);
  
  controller.stream.listen( (value) {
    print(value);
  });
}

output

1
2
3
4
Uncaught Error: Bad state: 5 is not less than 5

This will raise an error,

Stream data including

• Simple values
• Events
• Objects
• Collections
• Maps
• Errors

Data In -> StreamSink -> StreamContoller -> Data Out

Stream controller handling with Error, and on Done.
Stream Event Types

import 'dart:async';

void addLessThanFive(StreamController controller, int value) {
  if (value < 5) {
    controller.sink.add(value);
  } else {
    controller.sink.addError(StateError('$value is not less than 5'));
  }
}

void main() {
  final controller = StreamController();
  
  addLessThanFive(controller, 1);
  addLessThanFive(controller, 2);
  addLessThanFive(controller, 3);
  addLessThanFive(controller, 4);
  addLessThanFive(controller, 5);
  
  // close the StreamController
  controller.close();
  
  controller.stream.listen( (value) {
    print(value);
  }, onError: (error) {
    print(error);
  }, onDone: () {
    print('Done');
  });
}

Output

1
2
3
4
Bad state: 5 is not less than 5
Done

Remember:

After you close the Stream using controller.close(), you can not add event after closing!

Always close the StreamController if no more going on.

  controller.stream.listen( (value) {
    print(value);
  }, onError: (error) {
    print(error);
  }, onDone: () {
    print('Done');
  });

Notice
if the error is the last component of the StreamController(sink) when you listen, you will not see Done printed out. Because the last component (error) is operated by onError, there's no reason to go to onDone.
Be notice!

In pure Dart, we use StreamController to handle Stream object

In Flutter framework, we use StreamBuilder as a widget to handle Stream object. We call each item in the Stream as snapshot. And with snapshot inside of the StreamBuilder, we can judge snapshot.connectionState and pass snapshot.data out. snapshot.data is an item stored in the Stream object.

Demo codes

file auth.dart

class Auth implements AuthBase {
  final _firebaseAuth = FirebaseAuth.instance;

  @override
  Stream<User> authStateChanges() => _firebaseAuth.authStateChanges();

  @override
  User get currentUser => _firebaseAuth.currentUser;

  @override
  Future<User> signInAnonymously() async {
    final userCredential = await _firebaseAuth.signInAnonymously();
    return userCredential.user;
  }

  @override
  Future<void> signOut() async {
    _firebaseAuth.signOut();
  }
}

File landing_page.dart

class LandingPage extends StatelessWidget {
  const LandingPage({Key key, @required this.auth}) : super(key: key);
  final AuthBase auth;

  @override
  Widget build(BuildContext context) {
    return StreamBuilder<User>(
      stream: auth.authStateChanges(),
      //initialData: "adsf",
      builder: (context, snapshot) {
        if (snapshot.connectionState == ConnectionState.active) {
          final User user = snapshot.data;
          if (user == null) {
            return SignInPage(auth: auth);
          }
          return HomePage(auth: auth);
        }
        return Scaffold(
          body: Center(
            child: CircularProgressIndicator(),
          ),
        );
      },
    );
  }
}

So on different pages/widgets, we can call and judge variables asynchronously easily.

StreamBuilder solves this question in the past

With StreamBuilder, life is easier.

So on different pages/widgets, we can call and judge variables asynchronously easily.

Here in the codes, StreamBuilder builds a producer in the class Auth - authStateChanges() which returns a User object, and in the class LandingPage has a consumer StreamBuilder<User>, which checks snapshot.connectionState infinitely.

If the User object is changed, the class LandingPage will redirect to other widget with variables. In this case, the variable is auth object -- SignInPage(auth: auth) or HomePage(auth: auth).