How to convert Amplify.DataStore.observeQuery into stream with async* that can be used in Bloc

Question

I'm trying to convert Amplify.DataStore.observeQuery into a stream in repository class.
The expected return type is Stream<List<MyObject>>.

Answer 1

In the repository class you can write:

Stream<List<MyObject>> watchMessages() async* {
yield* Amplify.DataStore.observeQuery(MyObject.classType,
            where: MyObject.ID.eq(myId))
        .map((snapshot) => snapshot.items));
  }
}

or if you have to convert objects to domain then:

Stream<List<MyObject>> watchMessages() async* {
yield* Amplify.DataStore.observeQuery(MyObject.classType,
            where: MyObject.ID.eq(myId))
        .map((snapshot) => snapshot.items.map((e) => e.toDomain()).toList());
  }
}

Then in the bloc you can emit states on every stream input:

MyBloc(this._repository) : super(MyState.initial()) {
    on<_WatchMessages>((_onWatchMessages));
  }

void _onWatchMessages(_WatchMessages event, Emitter<MyState> emit) async {
    await emit.forEach(_repository.watchMessages(),
        onData: ((List<MyObject> messages) {
      return state.copyWith(messages: List.from(messages));
    }));
  }