diff --git a/CHANGELOG.md b/CHANGELOG.md
index 0aa832a..f7b9027 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,3 +1,8 @@
+## 3.4.0
+
+- Remove `onClose` from `SseConnection` and ensure the corresponding
+  `sink.close` correctly fires.
+
 ## 3.3.0
 
 - Add an `onClose` event to the `SseConnection`. This allows consumers to
diff --git a/lib/src/server/sse_handler.dart b/lib/src/server/sse_handler.dart
index e3f6cc7..c3c80c8 100644
--- a/lib/src/server/sse_handler.dart
+++ b/lib/src/server/sse_handler.dart
@@ -41,10 +41,9 @@
 
   final _closedCompleter = Completer<void>();
 
-  /// Completes when the [SseConnection] closes.
-  ///
-  /// This is guaranteed to fire unlike `this.sink.close`;
-  Future<void> get onClose => _closedCompleter.future;
+  /// Wraps the `_outgoingController.stream` to buffer events to enable keep
+  /// alive.
+  StreamQueue _outgoingStreamQueue;
 
   /// Creates an [SseConnection] for the supplied [_sink].
   ///
@@ -55,14 +54,15 @@
   /// If [keepAlive] is not supplied, the connection will be closed immediately
   /// after a disconnect.
   SseConnection(this._sink, {Duration keepAlive}) : _keepAlive = keepAlive {
+    _outgoingStreamQueue = StreamQueue(_outgoingController.stream);
     unawaited(_setUpListener());
     _outgoingController.onCancel = _close;
     _incomingController.onCancel = _close;
   }
 
   Future<void> _setUpListener() async {
-    var outgoingStreamQueue = StreamQueue(_outgoingController.stream);
-    while (await outgoingStreamQueue.hasNext) {
+    while (
+        !_outgoingController.isClosed && await _outgoingStreamQueue.hasNext) {
       // If we're in a KeepAlive timeout, there's nowhere to send messages so
       // wait a short period and check again.
       if (isInKeepAlivePeriod) {
@@ -72,7 +72,7 @@
 
       // Peek the data so we don't remove it from the stream if we're unable to
       // send it.
-      final data = await outgoingStreamQueue.peek;
+      final data = await _outgoingStreamQueue.peek;
 
       // Ignore outgoing messages since the connection may have closed while
       // waiting for the keep alive.
@@ -82,7 +82,7 @@
         // JSON encode the message to escape new lines.
         _sink.add('data: ${json.encode(data)}\n');
         _sink.add('\n');
-        await outgoingStreamQueue.next; // Consume from stream if no errors.
+        await _outgoingStreamQueue.next; // Consume from stream if no errors.
       } catch (StateError) {
         if (_keepAlive == null || _closedCompleter.isCompleted) {
           rethrow;
@@ -127,7 +127,10 @@
     if (!_closedCompleter.isCompleted) {
       _closedCompleter.complete();
       _sink.close();
-      if (!_outgoingController.isClosed) _outgoingController.close();
+      if (!_outgoingController.isClosed) {
+        _outgoingStreamQueue.cancel(immediate: true);
+        _outgoingController.close();
+      }
       if (!_incomingController.isClosed) _incomingController.close();
     }
   }
diff --git a/pubspec.yaml b/pubspec.yaml
index 194277d..612c417 100644
--- a/pubspec.yaml
+++ b/pubspec.yaml
@@ -1,5 +1,5 @@
 name: sse
-version: 3.3.0
+version: 3.4.0
 homepage: https://github.com/dart-lang/sse
 description: >-
   Provides client and server functionality for setting up bi-directional