Prevent onDone error (dart-lang/sse#40)

diff --git a/pkgs/sse/CHANGELOG.md b/pkgs/sse/CHANGELOG.md
index bf7899e..0ed01c9 100644
--- a/pkgs/sse/CHANGELOG.md
+++ b/pkgs/sse/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/pkgs/sse/lib/client/sse_client.dart b/pkgs/sse/lib/client/sse_client.dart
index 22aeb41..2958bed 100644
--- a/pkgs/sse/lib/client/sse_client.dart
+++ b/pkgs/sse/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/pkgs/sse/pubspec.yaml b/pkgs/sse/pubspec.yaml
index 8f15aff..9a44e4c 100644
--- a/pkgs/sse/pubspec.yaml
+++ b/pkgs/sse/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/pkgs/sse/test/web/index.dart.js b/pkgs/sse/test/web/index.dart.js
index 034c69b..614df07 100644
--- a/pkgs/sse/test/web/index.dart.js
+++ b/pkgs/sse/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);