Watch() not working with StreamBuilder

[JohnF17]

Hi, I just want to make it clear that, Before opening this issue I've looked at all the other related issues, tried their approach, checked out both the examples provided on watch, and also tried other methods, i'm still yet to come across a solution for the countless effort I've put forth onto making this work. Now i really need help with it.

Basic overview

• There is a collection in the db that's called private-messages, in that i have documents that have a field called group_id, which is basically the senders_id + _ + recipient_id.
• One of the other fields is the messages field which is an object (map) that holds the timestamp as key and the message as value
• The group_id is used to distinguish the conversation between two people
• The messages object (map) is used to collect the messages

  Here is what i want to achieve

• I want a stream that watches the document that equals the provided group_id and give/report changes when its messages object (map) is updated (could be an insert, update or delete).

• If it returns that specific document when any change happens to itself only that would be great👍(meaning i don't want it to watch over the whole collection and update whenever an unrelated document is updated. It only has to watch and return that specific document whenever that document itself changes, hope its not complicated😅)

  What I have tried

  Stream<Map<String, dynamic>?> getPrivateMessagesStream(
  String senderId,
  String recipientId,
  ) async* {
  final messages = (await getConnection()).collection('private-messages');
  
  StreamController<Map<String, dynamic>?> streamController =
      StreamController();
  StreamSubscription<dynamic>? streamSubscription;
  
  var pipeline = AggregationPipelineBuilder()
      .addStage(
        Match(
          where
              .eq('fullDocument.group_id', '${senderId}_$recipientId')
              .map['\$query'],
        ),
      )
      .addStage(
        Match(where.eq('operationType', 'update').map['\$query']),
      );
  var stream = messages.watch(pipeline,
      changeStreamOptions: ChangeStreamOptions(fullDocument: 'updateLookup'));
  
  streamSubscription = stream.listen((changeEvent) {
    print('listening to changes');
    Map<String, dynamic>? fullDocument = changeEvent.fullDocument;
    streamController.add(fullDocument);
  }, onDone: () {
    print('done');
    streamController.close();
    streamSubscription?.cancel();
  }, onError: (e) {
    print('error $e');
    streamController.close();
    streamSubscription?.cancel();
  });
  
  Timer.periodic(const Duration(seconds: 5), (t) async {
    print('changing data');
    await messages.updateOne(where.eq('group_id', '${senderId}_$recipientId'),
        ModifierBuilder().set('messages',  {'2023-1-1':'hello'}),
        writeConcern: WriteConcern.majority);
  });
  
  yield* streamController.stream;
  }

  Explanation

  The method getPrivateMessagesStream() is inside a singleton class called dbConnection, i'm calling this method inside a streamBuilder to show the messages.

This Below is Just For demonstration (extracting messages from documents etc.. has been implemented properly)

StreamBuilder(
  stream: DBConnection.getInstance()
      .getPrivateMessagesStream('63d79cb3412bf151d5436a85',
          '63d79cbf412bf151d5436a86'),
  builder: (context, snapshot) {
    if (snapshot.hasData) {
      return Center(child: Text(snapshot.data.toString()));
    }
    if (snapshot.hasError) {
      return const Center(child: Text('no data'));
    }
    return const Center(child: Text('loading'));
  },
)

And so here is the problem, i don't see any changes, it always displays loading in the streamBuilder, and the print statement listening to changes that is inside the listen method doesn't get called, even though i have put an updating timer inside the method getPrivateMessagesStream() that changes the data regularly (just for testing btw) and that itself prints changing data, (and funny enough i do see the changes happen in the db btw, no connection issues there), i just don't seem to get whats happening here, I really need help with this.

This is the point but Please do feel free to comment and correct any wrong doings here and/or in the db method i'm using Small Disclaimer

• The getConnection() method is a singleton so I'm not wasting memory and resources 😅
• The streamController and streamSubscription are a bit of a stretch and might need another method to be properly canceled and closed so feel free to comment or suggest another method

Additional info

As of now (1/30/2023), I'm using the latest monog_db package, i'm using atlas shared free, latest flutter stable and other CRUD operations work well.

[giorgiofran]

Hi, I'n not a flutter expert, so I'd have to spend some time in creating a test-case. Anyway, at a fisrt glance, it is not clear to me why you have to create a new stream. The object returned by watch() is a strem itself.

[JohnF17]

First of all, thank you for your reply and secondly

so I'd have to spend some time in creating a test-case.

You shall not take your time for this, i did some refactoring and few code changes 😁 the issue is now fixed, thanks to this phrase below

The object returned by watch() is a stream itself.

I have absolutely no idea how I let that fly by me😅, maybe its because I was too focused on closing the stream, forgetting StreamBuilders do that by default when they're removed off of the widget tree.

Anyhow, if it helps others, here's what i changed to the method getPrivateMessagesStream()

  Stream getPrivateMessagesStream({
    required String senderId,
    required String recipientId,
  }) async* {
    DbCollection? messages;
    String groupId = _getProperGroupId(senderId, recipientId);

    try {
      messages = (await getConnection()).collection('private-messages');
    } catch (e) {
      rethrow;
    }

    var pipeline = AggregationPipelineBuilder()
        .addStage(
          Match(where.eq('fullDocument._id', groupId).map['\$query']),
        )
        .addStage(
          Match(where.eq('operationType', 'update').map['\$query']),
        );

    // Just for testing purposes to listen to changes
    // Timer.periodic(
    //   const Duration(seconds: 5),
    //   (t) async {
    //     sendPrivateMessage(
    //       message: 'Hello ${t.tick}',
    //       recipientId: recipientId,
    //       senderId: senderId,
    //       reactionId: 3,
    //     );
    //   },
    // );

    yield* messages.watch(
      pipeline,
      changeStreamOptions: ChangeStreamOptions(fullDocument: 'updateLookup'),
    );
  }

And used it in a StreamBuilder like this

StreamBuilder(
  stream: DBConnection.getInstance().getPrivateMessagesStream(
    senderId: '63d79cb3412bf151d5436a85', // Sender
    recipientId: '63d79cbf412bf151d5436a86', // Recipient
  ),
  builder: (context, snapshot) {
    if (snapshot.hasData) {
      return Center(
        child: Text(snapshot.data.fullDocument.toString()),
      );
    }
    if (snapshot.hasError) {
      return Center(
          child: Text('no data, ${snapshot.error}'));
    }
    return const Center(child: Text('loading'));
  },
),

This issue is basically a closed issue and I will close it soon, but if you may, could you answer the following questions for me?

1. Currently, the StreamBuilder shows data only when an update occurs else shows loading (meaning if there are no updates, not even the initial data show up), how do I make it show the current available data first whether or not an update occurs and then let it do its normal job?
2. Establishing a connection to the database at first takes about 30 seconds, is that normal or is it probably connection issues?
3. What other operationTypes are there besides update, insert and delete (is there like an all, or a different type like operationType)?

Thank you in Advance!

[giorgiofran]

Establishing a connection to the database at first takes about 30 seconds, is that normal or is it probably connection issues?

Normally is by far faster. But I have been already told that sometimes it takes a lot. Are you connecting to an Atlas free cluster?

What other operationTypes are there besides update, insert and delete (is there like an all, or a different type like operationType)?

see here

[JohnF17]

Are you connecting to an Atlas free cluster?

Yes, I am indeed.

see here

I shall look in to it, thanks

[giorgiofran]

Currently, the StreamBuilder shows data only when an update occurs else shows loading (meaning if there are no updates, not even the initial data show up), how do I make it show the current available data first whether or not an update occurs and then let it do its normal job?

StremBuilder has an initialData property.

[JohnF17]

StremBuilder has an initialData property.

I was hoping to find a different way, but i guess i'll just use that, probably get data from cache (data i'll save locally).

Thanks and have a good one! 🙂