diff --git a/lib/src/data_stream/stream_writer.dart b/lib/src/data_stream/stream_writer.dart
index 94d4c476..499500df 100644
--- a/lib/src/data_stream/stream_writer.dart
+++ b/lib/src/data_stream/stream_writer.dart
@@ -11,7 +11,7 @@ import '../utils.dart';
 
 class BaseStreamWriter<T, InfoType extends BaseStreamInfo> {
   final StreamWriter<T> writableStream;
-  Function()? onClose;
+  Future<void> Function()? onClose;
 
   final InfoType info;
 
@@ -23,7 +23,7 @@ class BaseStreamWriter<T, InfoType extends BaseStreamInfo> {
 
   Future<void> close() async {
     await writableStream.close();
-    onClose?.call();
+    await onClose?.call();
   }
 }
 
diff --git a/lib/src/managers/event.dart b/lib/src/managers/event.dart
index 36500d02..96ae45bb 100644
--- a/lib/src/managers/event.dart
+++ b/lib/src/managers/event.dart
@@ -185,7 +185,7 @@ abstract class EventsListenable<T> extends Disposable {
       // cast to E
       await then(event);
       // cancel after 1 event
-      cancelFunc?.call();
+      await cancelFunc?.call();
     });
     return cancelFunc;
   }
diff --git a/lib/src/participant/local.dart b/lib/src/participant/local.dart
index 4d7cca30..0d48bf51 100644
--- a/lib/src/participant/local.dart
+++ b/lib/src/participant/local.dart
@@ -1218,13 +1218,7 @@ extension DataStreamParticipantMethods on LocalParticipant {
 
     final cancelFun = room.engine.events.once<EngineClosingEvent>((_) => onEngineClose);
 
-    final writer = TextStreamWriter(
-      writableStream: writableStream,
-      info: info,
-      onClose: () {
-        cancelFun?.call();
-      },
-    );
+    final writer = TextStreamWriter(writableStream: writableStream, info: info, onClose: cancelFun);
 
     return writer;
   }
@@ -1320,9 +1314,7 @@ extension DataStreamParticipantMethods on LocalParticipant {
     final byteWriter = ByteStreamWriter(
       writableStream: writableStream,
       info: info,
-      onClose: () {
-        cancelFun?.call();
-      },
+      onClose: cancelFun,
     );
 
     return byteWriter;
diff --git a/lib/src/preconnect/pre_connect_audio_buffer.dart b/lib/src/preconnect/pre_connect_audio_buffer.dart
index 3cb3ec48..03b79b42 100644
--- a/lib/src/preconnect/pre_connect_audio_buffer.dart
+++ b/lib/src/preconnect/pre_connect_audio_buffer.dart
@@ -223,7 +223,7 @@ class PreConnectAudioBuffer {
   Future<void> reset() async {
     await stopRecording();
     _timeoutTimer?.cancel();
-    _participantStateListener?.call();
+    await _participantStateListener?.call();
     _participantStateListener = null;
     _buffer.clear();
 
diff --git a/lib/src/types/other.dart b/lib/src/types/other.dart
index f69737b8..f5c1dd7b 100644
--- a/lib/src/types/other.dart
+++ b/lib/src/types/other.dart
@@ -19,7 +19,7 @@ import 'package:flutter/material.dart';
 import '../extensions.dart';
 import '../participant/participant.dart';
 
-typedef CancelListenFunc = Function();
+typedef CancelListenFunc = Future<void> Function();
 
 /// Protocol version to use when connecting to server.
 /// Usually it's not recommended to change this.
diff --git a/test/core/room_e2e_test.dart b/test/core/room_e2e_test.dart
index e4813247..9dbd70f4 100644
--- a/test/core/room_e2e_test.dart
+++ b/test/core/room_e2e_test.dart
@@ -87,7 +87,7 @@ void main() {
       await room.events.waitFor<ParticipantConnectedEvent>(duration: const Duration(seconds: 1));
 
       // Clean up listener
-      cancel();
+      await cancel();
 
       // Verify participant had tracks when connected event was emitted
       expect(participantHadTracksOnConnect, isTrue,