Fixed requestResponseSubscriber race condition (#137)

echistyakov · web-flow · commit 6a82fb19fc55 · 2024-12-26T20:18:48.000+08:00
diff --git a/internal/socket/subscriber_request_response.go b/internal/socket/subscriber_request_response.go
@@ -37,7 +37,7 @@ type requestResponseSubscriber struct {
 	dc        *DuplexConnection
 	sid       uint32
 	receiving fragmentation.HeaderAndPayload
-	sndCnt    int32
+	sndCnt    atomic.Int32
 }
 
 func borrowRequestResponseSubscriber(dc *DuplexConnection, sid uint32, receiving fragmentation.HeaderAndPayload) rx.Subscriber {
@@ -55,13 +55,13 @@ func returnRequestResponseSubscriber(s rx.Subscriber) {
 	}
 	actual.dc = nil
 	actual.receiving = nil
-	actual.sndCnt = 0
+	actual.sndCnt.Store(0)
 	globalRequestResponseSubscriberPool.put(actual)
 }
 
 func (r *requestResponseSubscriber) OnNext(next payload.Payload) {
 	r.dc.sendPayload(r.sid, next, core.FlagNext|core.FlagComplete)
-	atomic.AddInt32(&r.sndCnt, 1)
+	r.sndCnt.Add(1)
 }
 
 func (r *requestResponseSubscriber) OnError(err error) {
@@ -73,7 +73,7 @@ func (r *requestResponseSubscriber) OnError(err error) {
 }
 
 func (r *requestResponseSubscriber) OnComplete() {
-	if atomic.AddInt32(&r.sndCnt, 1) == 1 {
+	if r.sndCnt.Add(1) == 1 {
 		r.dc.sendPayload(r.sid, payload.Empty(), core.FlagComplete)
 	}
 	r.dc.unregister(r.sid)

Original file line number	Diff line number	Diff line change
`@@ -37,7 +37,7 @@ type requestResponseSubscriber struct {`
`37`	`37`	`dc *DuplexConnection`
`38`	`38`	`sid uint32`
`39`	`39`	`receiving fragmentation.HeaderAndPayload`
`40`		`- sndCnt int32`
	`40`	`+ sndCnt atomic.Int32`
`41`	`41`	`}`
`42`	`42`
`43`	`43`	`func borrowRequestResponseSubscriber(dc *DuplexConnection, sid uint32, receiving fragmentation.HeaderAndPayload) rx.Subscriber {`
`@@ -55,13 +55,13 @@ func returnRequestResponseSubscriber(s rx.Subscriber) {`
`55`	`55`	`}`
`56`	`56`	`actual.dc = nil`
`57`	`57`	`actual.receiving = nil`
`58`		`- actual.sndCnt = 0`
	`58`	`+ actual.sndCnt.Store(0)`
`59`	`59`	`globalRequestResponseSubscriberPool.put(actual)`
`60`	`60`	`}`
`61`	`61`
`62`	`62`	`func (r *requestResponseSubscriber) OnNext(next payload.Payload) {`
`63`	`63`	`r.dc.sendPayload(r.sid, next, core.FlagNext\|core.FlagComplete)`
`64`		`- atomic.AddInt32(&r.sndCnt, 1)`
	`64`	`+ r.sndCnt.Add(1)`
`65`	`65`	`}`
`66`	`66`
`67`	`67`	`func (r *requestResponseSubscriber) OnError(err error) {`
`@@ -73,7 +73,7 @@ func (r *requestResponseSubscriber) OnError(err error) {`
`73`	`73`	`}`
`74`	`74`
`75`	`75`	`func (r *requestResponseSubscriber) OnComplete() {`
`76`		`- if atomic.AddInt32(&r.sndCnt, 1) == 1 {`
	`76`	`+ if r.sndCnt.Add(1) == 1 {`
`77`	`77`	`r.dc.sendPayload(r.sid, payload.Empty(), core.FlagComplete)`
`78`	`78`	`}`
`79`	`79`	`r.dc.unregister(r.sid)`