@@ -47,6 +47,7 @@ type Conn struct {
47
47
// read limit for a message in bytes.
48
48
msgReadLimit xsync.Int64
49
49
50
+ wg sync.WaitGroup
50
51
closingMu sync.Mutex
51
52
isReadClosed xsync.Int64
52
53
closeOnce sync.Once
@@ -223,6 +224,7 @@ func (c *Conn) write(ctx context.Context, typ MessageType, p []byte) error {
223
224
// or the connection is closed.
224
225
// It thus performs the full WebSocket close handshake.
225
226
func (c * Conn ) Close (code StatusCode , reason string ) error {
227
+ defer c .wg .Wait ()
226
228
err := c .exportedClose (code , reason )
227
229
if err != nil {
228
230
return fmt .Errorf ("failed to close WebSocket: %w" , err )
@@ -236,6 +238,7 @@ func (c *Conn) Close(code StatusCode, reason string) error {
236
238
// note: No different from Close(StatusGoingAway, "") in WASM as there is no way to close
237
239
// a WebSocket without the close handshake.
238
240
func (c * Conn ) CloseNow () error {
241
+ defer c .wg .Wait ()
239
242
return c .Close (StatusGoingAway , "" )
240
243
}
241
244
@@ -388,10 +391,15 @@ func (c *Conn) CloseRead(ctx context.Context) context.Context {
388
391
c .isReadClosed .Store (1 )
389
392
390
393
ctx , cancel := context .WithCancel (ctx )
394
+ c .wg .Add (1 )
391
395
go func () {
396
+ defer c .CloseNow ()
397
+ defer c .wg .Done ()
392
398
defer cancel ()
393
- c .read (ctx )
394
- c .Close (StatusPolicyViolation , "unexpected data message" )
399
+ _ , _ , err := c .read (ctx )
400
+ if err != nil {
401
+ c .Close (StatusPolicyViolation , "unexpected data message" )
402
+ }
395
403
}()
396
404
return ctx
397
405
}
@@ -580,3 +588,11 @@ func (m *mu) unlock() {
580
588
type noCopy struct {}
581
589
582
590
func (* noCopy ) Lock () {}
591
+
592
+ func (c * Conn ) wgGo (fn func ()) {
593
+ c .wg .Add (1 )
594
+ go func () {
595
+ defer c .wg .Done ()
596
+ fn ()
597
+ }()
598
+ }
0 commit comments