fix(request_response): dead request with an empty response (#93)

Author: jjeffcaii

go.mod

@@ -6,7 +6,7 @@ require (
 	github.com/golang/mock v1.4.3
 	github.com/google/uuid v1.1.2
 	github.com/gorilla/websocket v1.4.2
-	github.com/jjeffcaii/reactor-go v0.4.3
+	github.com/jjeffcaii/reactor-go v0.4.4
 	github.com/pkg/errors v0.9.1
 	github.com/pkg/profile v1.5.0
 	github.com/stretchr/testify v1.6.1

go.sum

@@ -10,8 +10,8 @@ github.com/google/uuid v1.1.2 h1:EVhdT+1Kseyi1/pUmXKaFxYsDNy9RQYkMWRH68J/W7Y=
 github.com/google/uuid v1.1.2/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/gorilla/websocket v1.4.2 h1:+/TMaTYc4QFitKJxsQ7Yye35DkWvkdLcvGKqM+x0Ufc=
 github.com/gorilla/websocket v1.4.2/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
-github.com/jjeffcaii/reactor-go v0.4.3 h1:nh0epvnYQhzGDKc0PUImkONxx40j1Kh4V9MoNGNQ6Is=
-github.com/jjeffcaii/reactor-go v0.4.3/go.mod h1:2ZzeNFnQ2c55NHRh0KJ4k5yMvmrcpx1APzh7BKkRNQE=
+github.com/jjeffcaii/reactor-go v0.4.4 h1:s5zOwQ7hut6ICL/jucFoHiAwDx/ugOJpS/Ncp8Cc4Ko=
+github.com/jjeffcaii/reactor-go v0.4.4/go.mod h1:2ZzeNFnQ2c55NHRh0KJ4k5yMvmrcpx1APzh7BKkRNQE=
 github.com/panjf2000/ants/v2 v2.4.3 h1:wHghL17YKFanB62QjPQ9o+DuM4q7WrQ7zAhoX8+eBXU=
 github.com/panjf2000/ants/v2 v2.4.3/go.mod h1:f6F0NZVFsGCp5A7QW/Zj/m92atWwOkY0OIhFxRNFr4A=
 github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=

internal/socket/duplex.go

@@ -1033,7 +1033,6 @@ func (dc *DuplexConnection) sendPayload(
 	sending payload.Payload,
 	frameFlag core.FrameFlag,
 ) {
-
 	d := sending.Data()
 	m, _ := sending.Metadata()
 	size := framing.CalcPayloadFrameSize(d, m)

internal/socket/subscriber_request_response.go

@@ -3,6 +3,7 @@ package socket
 import (
 	"context"
 	"sync"
+	"sync/atomic"
 
 	"github.com/jjeffcaii/reactor-go"
 	"github.com/rsocket/rsocket-go/core"
@@ -12,20 +13,35 @@ import (
 	"github.com/rsocket/rsocket-go/rx"
 )
 
-var _requestResponseSubscriberPool = sync.Pool{
-	New: func() interface{} {
-		return new(requestResponseSubscriber)
-	},
+var globalRequestResponseSubscriberPool requestResponseSubscriberPool
+
+type requestResponseSubscriberPool struct {
+	inner sync.Pool
+}
+
+func (p *requestResponseSubscriberPool) get() *requestResponseSubscriber {
+	if exist, _ := p.inner.Get().(*requestResponseSubscriber); exist != nil {
+		return exist
+	}
+	return &requestResponseSubscriber{}
+}
+
+func (p *requestResponseSubscriberPool) put(s *requestResponseSubscriber) {
+	if s == nil {
+		return
+	}
+	p.inner.Put(s)
 }
 
 type requestResponseSubscriber struct {
 	dc        *DuplexConnection
 	sid       uint32
 	receiving fragmentation.HeaderAndPayload
+	sndCnt    int32
 }
 
 func borrowRequestResponseSubscriber(dc *DuplexConnection, sid uint32, receiving fragmentation.HeaderAndPayload) rx.Subscriber {
-	s := _requestResponseSubscriberPool.Get().(*requestResponseSubscriber)
+	s := globalRequestResponseSubscriberPool.get()
 	s.receiving = receiving
 	s.dc = dc
 	s.sid = sid
@@ -39,11 +55,13 @@ func returnRequestResponseSubscriber(s rx.Subscriber) {
 	}
 	actual.dc = nil
 	actual.receiving = nil
-	_requestResponseSubscriberPool.Put(actual)
+	actual.sndCnt = 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)
 }
 
 func (r *requestResponseSubscriber) OnError(err error) {
@@ -55,6 +73,9 @@ func (r *requestResponseSubscriber) OnError(err error) {
 }
 
 func (r *requestResponseSubscriber) OnComplete() {
+	if atomic.AddInt32(&r.sndCnt, 1) == 1 {
+		r.dc.sendPayload(r.sid, payload.Empty(), core.FlagComplete)
+	}
 	r.dc.unregister(r.sid)
 	r.finish()
 }

payload/payload.go

@@ -9,6 +9,8 @@ import (
 	"github.com/rsocket/rsocket-go/internal/common"
 )
 
+var _empty = New(nil, nil)
+
 type (
 	// Payload is a stream message (upstream or downstream).
 	// It contains data associated with a stream created by a previous request.
@@ -122,3 +124,8 @@ func Equal(a Payload, b Payload) bool {
 
 	return bytes.Equal(m1, m2)
 }
+
+// Empty returns an empty Payload.
+func Empty() Payload {
+	return _empty
+}

payload/payload_test.go

@@ -92,3 +92,8 @@ func TestNewFile(t *testing.T) {
 		payload.MustNewFile("/not/existing", nil)
 	}()
 }
+
+func TestEmpty(t *testing.T) {
+	p := payload.Empty()
+	assert.Nil(t, p.Data())
+}