Fixed processor dispose race condition (#135)

echistyakov · web-flow · commit 80656998673a · 2024-12-26T20:26:39.000+08:00
diff --git a/internal/socket/duplex.go b/internal/socket/duplex.go
@@ -256,11 +256,13 @@ func (dc *DuplexConnection) RequestResponse(req payload.Payload) (res mono.Mono)
 
 	onFinally := func(s reactor.SignalType, d reactor.Disposable) {
 		common.TryRelease(handler.cache)
-		d.Dispose()
 		if s == reactor.SignalTypeCancel {
 			dc.sendFrame(framing.NewWriteableCancelFrame(sid))
 		}
+		// Unregister handler w/sink (processor).
 		dc.unregister(sid)
+		// Dispose sink (processor).
+		d.Dispose()
 	}
 
 	m, s, _ := mono.NewProcessor(dc.reqSche, onFinally)

Original file line number	Diff line number	Diff line change
`@@ -256,11 +256,13 @@ func (dc *DuplexConnection) RequestResponse(req payload.Payload) (res mono.Mono)`
`256`	`256`
`257`	`257`	`onFinally := func(s reactor.SignalType, d reactor.Disposable) {`
`258`	`258`	`common.TryRelease(handler.cache)`
`259`		`- d.Dispose()`
`260`	`259`	`if s == reactor.SignalTypeCancel {`
`261`	`260`	`dc.sendFrame(framing.NewWriteableCancelFrame(sid))`
`262`	`261`	`}`
	`262`	`+ // Unregister handler w/sink (processor).`
`263`	`263`	`dc.unregister(sid)`
	`264`	`+ // Dispose sink (processor).`
	`265`	`+ d.Dispose()`
`264`	`266`	`}`
`265`	`267`
`266`	`268`	`m, s, _ := mono.NewProcessor(dc.reqSche, onFinally)`