registerStream<T extends Object> method

Stream<T> registerStream<T extends Object>(QueryStreamFetcher<T> fetcher)

Creates a new stream from a select statement expressed through fetcher.

Implementation

Stream<T> registerStream<T extends Object>(QueryStreamFetcher<T> fetcher) {
  final key = fetcher.key;

  if (key != null) {
    final cached = _activeKeyStreams[key];
    if (cached != null) {
      return cached._stream as Stream<T>;
    }
  }

  // no cached instance found, create a new stream and register it so later
  // requests with the same key can be cached.
  final stream = _StreamImpl<T>(fetcher, this);
  // todo this adds the stream to a map, where it will only be removed when
  // somebody listens to it and later calls .cancel(). Failing to do so will
  // cause a memory leak. Is there any way we can work around it? Perhaps a
  // weak reference with an Expando could help.
  _markAsOpened(stream);

  return stream._stream;
}