Supabase Realtime stream cancel and initialize again

[iampopal]

Describe the bug

We are using Supabase realtime for a chat app, and we are facing an unusual behavior with stream and channels

1) We have a global cubit that listens to room status name RoomStatusCubit
 This RoomStatusCubit is initialized by listening to the room_status table:

initStream() {
  var realstream = from('room_status')
      .stream(primaryKey: ['id'])
      .eq('to_uid', supabase.uid)
      .order('at');
  var subscription = realStream!.listen(
    (event) {
      log('realtime 0 listen event: $event');
      emit(
        state.copyWith(
          rooms: RoomStatus.listFrom(event),
          status: RealtimeStatus.result,
        ),
      );
    },
    onError: (e, s) {
      log('realtime 1 error: $e \ns:$s');
      if (e.toString() == '' || e.toString() == '{}') return;
      emit(state.copyWith(status: RealtimeStatus.error, error: e));
    },
  );
}

The above cubit has worked well for a while but has an issue when 
 1) the internet disconnects and reconnects 
 2) when for a long time there is no internet 
3) when an error happens

 
 
To solve our above issue we have decided to close current channel and reinitalize stream 

 


Future cancelStream() async {
    try {
      await subscription?.cancel();
      subscription = null;
    } catch (e, s) {
      subscription = null;
      loges(e, s, type: 'realtime cancelStream');
    }
  }

  List<RealtimeChannel> get roomStatusChannels => supabase
      .getChannels()
      .where((e) => e.topic.contains('room_status'))
      .toList();

  Future cancelChannel() async {
    for (RealtimeChannel c in roomStatusChannels) {
      log('realtime cancel topic ${c.topic}');
      await supabase.removeChannel(c);
    }

    await cancelStream();
  }



 The issue is when we call cancelChannel() and then initStream() the stream is not working anyway. 

currently, we need to restart our app to solve this issue and we need a better solution for restarting superbase realtime channel to work correctly so users do not close and open their app.

To Reproduce Steps to reproduce the behavior: 1) Create an app with realtime stream in a cubit 2) Try to cancelChannel() and initStream()

Expected behavior Allow us to cancel a realtime stream and init stream again without any error and stream shall work like restarting an app

Version (please complete the following information): On Linux/macOS Please run dart pub deps | grep -E "supabase|gotrue|postgrest|storage_client|realtime_client|functions_client" in your project directory and paste the output here. │ └── supabase 1.10.0 │ ├── functions_client 1.3.2 │ ├── gotrue 1.11.2 │ ├── postgrest 1.5.0 │ ├── realtime_client 1.1.3 │ ├── storage_client 1.5.1 ├── supabase_flutter 1.10.12 │ ├── supabase...

Additional context We sometime also see heartbeat timeout issue and not understand how this is related and sometime see Bad state: Cannot add event after closing error event or cubit is not closed and working correctly

[iampopal]

Error 1: We are also getting an empty error while listening to the stream

in the following error-> there is no error and stock-> is the stocktrace of error.

[log] error->  
stock->:#0      SupabaseStreamBuilder._addException
package:supabase/src/supabase_stream_builder.dart:442
#1      SupabaseStreamBuilder._getStreamData.<anonymous closure>
package:supabase/src/supabase_stream_builder.dart:348
#2      RealtimeChannel.subscribe.<anonymous closure>
package:realtime_client/src/realtime_channel.dart:202
#3      RealtimeChannel.onError.<anonymous closure>
package:realtime_client/src/realtime_channel.dart:331
#4      RealtimeChannel.trigger
package:realtime_client/src/realtime_channel.dart:569
#5      RealtimeClient._triggerChanError
package:realtime_client/src/realtime_client.dart:408
#6      RealtimeClient._onConnClose
package:realtime_client/src/realtime_client.dart:389
#7      RealtimeClient.connect.<anonymous closure>
package:realtime_client/src/realtime_client.dart:147
#8      _RootZone.runGuarded (dart:async/zone.dart:1581:10)
#9      _BufferingStreamSubscription._sendDone.sendDone (dart:async/stream_impl.dart:392:13)

[iampopal]

Error 2: we are getting this error after a long time of our app is open:

[ERROR:flutter/runtime/dart_vm_initializer.cc(41)] Unhandled Exception: Bad state: Cannot add event after closing
#0      _StreamController.add (dart:async/stream_controller.dart:595:24)
#1      _CompleterSink.add
package:web_socket_channel/src/sink_completer.dart:90
#2      RealtimeClient.push.callback.<anonymous closure>
package:realtime_client/src/realtime_client.dart:267
#3      new RealtimeClient.<anonymous closure>
package:realtime_client/src/realtime_client.dart:107
#4      RealtimeClient.push.callback
package:realtime_client/src/realtime_client.dart:267
#5      RealtimeClient.push
package:realtime_client/src/realtime_client.dart:284
#6      Push.send
package:realtime_client/src/push.dart:61
#7      Push.resend
package:realtime_client/src/push.dart:52
#8      RealtimeChannel.rejoin
package:realtime_client/src/realtime_channel.dart:510
#9      RealtimeChannel.rejoinUntilConnected
package:realtime_client/src/realtime_channel.dart:181
#10     new RealtimeChannel.<anonymous closure>
package:realtime_client/src/realtime_channel.dart:134
#11     RetryTimer.scheduleTimeout.<anonymous closure>
package:realtime_client/src/retry_timer.dart:46
#12     Timer._createTimer.<anonymous closure> (dart:async-patch/timer_patch.dart:18:15)
#13     _Timer._runTimers (dart:isolate-patch/timer_impl.dart:398:19)
#14     _Timer._handleMessage (dart:isolate-patch/timer_impl.dart:429:5)
#15     _RawReceivePort._handleMessage (dart:isolate-patch/isolate_patch.dart:192:26)

[maxfornacon]

Your error 2 seems to be similar to the issue I descibed in #579. Could be related.

[iampopal]

Your error 2 seems to be similar to the issue I descibed in #579. Could be related.

Yea, looking like #579 is related to this, and we need to cancel the stream and reinitialize the feature so in future if anything else happens in future will be easily solvable.

[iampopal]

How's the progress of this going? we need to be able to close and re-initialize streams because the internet may come and go.