diff --git a/CHANGELOG.md b/CHANGELOG.md
index bf7899e..0ed01c9 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,3 +1,8 @@
+## 3.8.1
+
+- Fix an issue where closing the `SseConnection` stream would result in
+  an error.
+
 ## 3.8.0
 
 - Add `onConnected` to replace `onOpen`.
diff --git a/lib/client/sse_client.dart b/lib/client/sse_client.dart
index 22aeb41..2958bed 100644
--- a/lib/client/sse_client.dart
+++ b/lib/client/sse_client.dart
@@ -82,10 +82,9 @@
 
   void close() {
     _eventSource.close();
-    // If the _outgoingController doesn't have a listener that means the
-    // initial connection was never established. Add a listener so close
+    // If the initial connection was never established. Add a listener so close
     // adds a done event to [sink].
-    if (!_outgoingController.hasListener) _outgoingController.stream.drain();
+    if (!_onConnected.isCompleted) _outgoingController.stream.drain();
     _incomingController.close();
     _outgoingController.close();
   }
diff --git a/pubspec.yaml b/pubspec.yaml
index 8f15aff..9a44e4c 100644
--- a/pubspec.yaml
+++ b/pubspec.yaml
@@ -1,5 +1,5 @@
 name: sse
-version: 3.8.0
+version: 3.8.1
 homepage: https://github.com/dart-lang/sse
 description: >-
   Provides client and server functionality for setting up bi-directional
diff --git a/test/web/index.dart.js b/test/web/index.dart.js
index 034c69b..614df07 100644
--- a/test/web/index.dart.js
+++ b/test/web/index.dart.js
@@ -7170,13 +7170,14 @@
       W._EventStreamSubscription$(_this._eventSource, "error", t2._as(new M.SseClient_closure1(_this)), false, t4);
     },
     close$0: function(_) {
-      var t1;
-      this._eventSource.close();
-      t1 = this._outgoingController;
-      if ((t1._state & 1) === 0)
+      var t1, _this = this;
+      _this._eventSource.close();
+      if (_this._onConnected.future._state === 0) {
+        t1 = _this._outgoingController;
         new P._ControllerStream(t1, H._instanceType(t1)._eval$1("_ControllerStream<1>")).listen$2$cancelOnError(null, true).asFuture$1$1(null, type$.dynamic);
-      this._incomingController.close$0(0);
-      t1.close$0(0);
+      }
+      _this._incomingController.close$0(0);
+      _this._outgoingController.close$0(0);
     },
     _onIncomingControlMessage$1: function(message) {
       var data = new P._AcceptStructuredCloneDart2Js([], []).convertNativeToDart_AcceptStructuredClone$2$mustCopy(type$.legacy_MessageEvent._as(type$.legacy_Event._as(message)).data, true);