Duplicate values with Drift and Riverpod family stream

[Kirsdarkenvaar]

I have the following stream in the Drift repository:

  Stream<String> watchCharacterLocation({required String characterId}) {
    log("Repo: watchCharacterLocation(): $characterId");
    final character = (database.select(database.characters)
          ..where((character) => character.id.equals(characterId)))
        .watchSingle();

    return character.map((character) {
      log("Repo: watchCharacterLocation() locationId: ${character.locationId}");
      return character.locationId;
    });
  }

I subscribe to it with this provider (not simplified because of the debug logging):

@riverpod
class CharacterLocationStream extends _$CharacterLocationStream {
  @override
  Stream<String> build({required String characterId}) async* {
    log("CharacterLocationStream Build: $characterId");
    state = const AsyncLoading();

    final activeCharacterRepository =
        ref.watch(activeCharacterRepositoryProvider);

    try {
      final stream = activeCharacterRepository
          .watchCharacterLocation(characterId: characterId)
          .distinct();

      await for (final value in stream) {
        log("New character location from stream ${stream.hashCode}: $characterId is in $value");
        state = AsyncData(value);
      }
    } catch (error) {
      state = AsyncError(error, StackTrace.current);
    }
  }
}

The first time this is run with character A, it works properly. If I then switch to character B, the old provider is disposed, a new provider is created, but now when character B's locationId changes, I get notified twice, once for character A and once for character B:

[log] Repo: watchCharacterLocation() locationId: 64293d76-af97-488f-bc01-0e48897fd7c9 // old character's location
[log] Repo: watchCharacterLocation() locationId: 054f316e-d4e6-4daf-8032-72eb26570bf3 // new character's location
[log] New character location from stream 633876365: 3243df85-ddef-46ee-8e2b-32f3f39ac30c is in 054f316e-d4e6-4daf-8032-72eb26570bf3

If I switch from character A to character A, the provider gets disposed and rebuilt with the same hashcode, but when character A's locationId changes I get notified twice.

Am I somehow supposed to stop listening to the Drift stream manually when the provider is disposed?

[dickermoshe]

This is more of a Riverpod question. There many things wrong with your code.

1. The build method never yields anything. if should be yield value instead of state = AsyncData(value);
2. You should be returning the stream object and let riverpod take care of error handling

@riverpod
class CharacterLocationStream extends _$CharacterLocationStream {
  @override
  Stream<String> build({required String characterId})  {
    final activeCharacterRepository =
        ref.watch(activeCharacterRepositoryProvider);
    return activeCharacterRepository
          .watchCharacterLocation(characterId: characterId)
          .distinct();
  }
}

Now Riverpod will cancel the stream when it's disposed

[dickermoshe]

If you would like to log data, use .map and add a log statement