Better way to handle a StreamBuilder where I also want to manually refresh the stream

I have this code below and it almost works I think, but it is not quite there. Essentially what I am trying to do it use a StreamBuilder to watch a database stream. This table may be updated by any means external to my app hence the stream. A user has the option to change the range of the query to use a different DateTimeRange though, and I want the stream to update in this case. I can get things working pretty easily using a setState and updating the whole widget but this widget shows a map from the google_maps_flutter package and it is not the prettiest thing to see watching it update each time.

This is my current attempt: I am having issues with errors such as Unhandled Exception: Bad state: Cannot add new events while doing an addStream trying to update the underlying stream of the StreamBuilder to use the new database query.

I did end up getting a working version that was very ugly by having two streams, one to detect refresh events, and one to detect the database changes. These two streams fed into another stream that would be connected to the StreamBuilder. This one had issues with the streams being called hundreds of times each refresh or database update which was not a pleasant UX.

There has to be some way that I am missing to do this.

TLDR; I want to have a stream to a database using a custom query where a user can change that query and also trigger a new stream emit, or the database changes can trigger a stream emit.I do not want to use setState as that will refresh the map I use.

  @override
  Widget build(BuildContext context) {
    return Scaffold(
      resizeToAvoidBottomInset: false,
      body: FutureBuilder(
        future: _fetchPrevDateRangeSettings(),
        builder: (_, ss) {
          if (ss.connectionState == ConnectionState.waiting) {
            return Center(child: CircularProgressIndicator());
          }

          _backOfficeStreamController = StreamController.broadcast(); // Create a broadcast StreamController

          // refresh will take the most up to date query parameters and refresh the stream
          _refreshStreamController.stream.listen((_) async {
            _backOfficeStreamController.close();
            _backOfficeStreamController.addStream(ref
                .read(claimsDatabaseProvider)
                .fetchMapDataForDateTimeRange2(
                _dateTimeRange, filterByInspectionDate));
          });

          // will listen for database changes
          _backOfficeStreamController.addStream(ref
              .watch(claimsDatabaseProvider)
              .fetchMapDataForDateTimeRange2(
                  _dateTimeRange, filterByInspectionDate));

          // listen to the stream, and update when the user refreshes, or db updates
          return StreamBuilder(
            stream: _backOfficeStreamController.stream,
            builder: (_, ss2) {
              if (ss2.connectionState == ConnectionState.waiting) {
                return Center(child: CircularProgressIndicator());
              }
              return Column(
                children: [
                  Expanded(
                    flex: 1,
                    child: _getMap(context, ss2.requireData),
                  ),
                  Expanded(
                    child: Column(
                      children: [
                        _getFilterWidget(ss2.requireData),
                        Divider(thickness: 1),
                        _getListWidget(ss2.requireData),
                      ],
                    ),
                  ),
                ],
              );
            },
          );
        },
      ),
    );
  }

I was not able to get a working solution from any of the comments or answers above, but I did get it working with using a StreamGroup as shown below:

I simply remove the old stream, and add the new one on a refresh action. I am sure there was a way to replace the stream using a stream controller, but I could not figure it out and using the stream group works while keeping the code pretty concise, even if it only ever holds one stream.

  void _refreshData() async {
    // Close and reopen the stream to avoid errors
    streamGroup.remove(backOfficeStream);
    backOfficeStream = ref
        .read(claimsDatabaseProvider)
        .fetchMapDataForDateTimeRange2(_dateTimeRange, filterByInspectionDate);
    streamGroup.add(backOfficeStream);
  }

  @override
  void dispose() async {
    _saveSettings();
    await streamGroup.remove(backOfficeStream);
    await streamGroup.close();
    super.dispose();
  }

@override
Widget build(BuildContext context) {
  return Scaffold(
    resizeToAvoidBottomInset: false,
    body: FutureBuilder(
      future: _fetchPrevDateRangeSettings(),
      builder: (_, ss) {
        if (ss.connectionState == ConnectionState.waiting) {
          return Center(child: CircularProgressIndicator());
        }

          // update the stream reference in the class
          backOfficeStream = ref
              .read(claimsDatabaseProvider)
              .fetchMapDataForDateTimeRange2(
                  _dateTimeRange, filterByInspectionDate);
          // add to stream group
          streamGroup.add(backOfficeStream);

        // Use the StreamController's stream for the StreamBuilder
        return StreamBuilder(
          stream: streamGroup.stream,
          builder: (_, ss2) {
            if (ss2.connectionState == ConnectionState.waiting) {
              return Center(child: CircularProgressIndicator());
            }
            return Column(
              children: [
                Expanded(
                  flex: 1,
                  child: _getMap(context, ss2.requireData),
                ),
                Expanded(
                  child: Column(
                    children: [
                      _getFilterWidget(ss2.requireData),
                      Divider(thickness: 1),
                      _getListWidget(ss2.requireData),
                    ],
                  ),
                ),
              ],
            );
          },
        );
      },
    ),
  );
}