adapt oneshot mono api. (#76)

* adapt oneshot mono api.

examples/echo/echo.go

@@ -76,7 +76,7 @@ func responder() rsocket.RSocket {
 		}),
 		rsocket.RequestResponse(func(pl payload.Payload) mono.Mono {
 			// just echo
-			return mono.Just(pl)
+			return mono.JustOneshot(pl)
 
 			// Graceful with context API.
 			//return rx.NewMono(func(ctx context.Context, sink rx.MonoProducer) {

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.3.0
+	github.com/jjeffcaii/reactor-go v0.3.1
 	github.com/pkg/errors v0.9.1
 	github.com/stretchr/testify v1.6.1
 	github.com/urfave/cli/v2 v2.1.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.3.0 h1:aiWHiyzzy0Q0SeG2ESLhic8kkFwLA9r8JCEBUpT28h4=
-github.com/jjeffcaii/reactor-go v0.3.0/go.mod h1:w5NNtYKACJocob2sc18oPf1F0/NJBhWITZcfcEoyZOQ=
+github.com/jjeffcaii/reactor-go v0.3.1 h1:IQwuVdhEEsM5FJyraOxR9kLfumQkJ7v36fN7i9KBvyc=
+github.com/jjeffcaii/reactor-go v0.3.1/go.mod h1:w5NNtYKACJocob2sc18oPf1F0/NJBhWITZcfcEoyZOQ=
 github.com/panjf2000/ants/v2 v2.4.2 h1:kesjjo8JipN3vNNg1XaiXaeSs6xJweBTgenkBtsrHf8=
 github.com/panjf2000/ants/v2 v2.4.2/go.mod h1:f6F0NZVFsGCp5A7QW/Zj/m92atWwOkY0OIhFxRNFr4A=
 github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=

internal/socket/duplex.go

@@ -10,6 +10,7 @@ import (
 	"github.com/rsocket/rsocket-go/core"
 	"github.com/rsocket/rsocket-go/core/framing"
 	"github.com/rsocket/rsocket-go/core/transport"
+	"github.com/rsocket/rsocket-go/internal/bytesconv"
 	"github.com/rsocket/rsocket-go/internal/common"
 	"github.com/rsocket/rsocket-go/internal/fragmentation"
 	"github.com/rsocket/rsocket-go/internal/queue"
@@ -389,9 +390,7 @@ func (dc *DuplexConnection) RequestChannel(sending flux.Flux) (ret flux.Flux) {
 				rcv:          receiving,
 				result:       sendResult,
 			}
-			sending.
-				SubscribeOn(scheduler.Parallel()).
-				SubscribeWith(context.Background(), sub)
+			sending.SubscribeOn(scheduler.Parallel()).SubscribeWith(context.Background(), sub)
 		})
 	return ret
 }
@@ -408,7 +407,7 @@ func (dc *DuplexConnection) onFrameRequestResponse(frame core.BufferedFrame) err
 func (dc *DuplexConnection) respondRequestResponse(receiving fragmentation.HeaderAndPayload) error {
 	sid := receiving.Header().StreamID()
 
-	// 1. execute socket handler
+	// execute socket handler
 	sending, err := func() (mono mono.Mono, err error) {
 		defer func() {
 			if e := recover(); e != nil {
@@ -419,22 +418,29 @@ func (dc *DuplexConnection) respondRequestResponse(receiving fragmentation.Heade
 		mono = dc.responder.RequestResponse(receiving)
 		return
 	}()
-	// 2. sending error with panic
+	// sending error with panic
 	if err != nil {
 		common.TryRelease(receiving)
 		dc.writeError(sid, err)
 		return nil
 	}
-	// 3. sending error with unsupported handler
+	// sending error with unsupported handler
 	if sending == nil {
 		common.TryRelease(receiving)
 		dc.writeError(sid, framing.NewWriteableErrorFrame(sid, core.ErrorCodeApplicationError, unsupportedRequestResponse))
 		return nil
 	}
 
-	// 4. async subscribe publisher
+	// async subscribe publisher
 	sub := borrowRequestResponseSubscriber(dc, sid, receiving)
-	sending.SubscribeOn(scheduler.Parallel()).SubscribeWith(context.Background(), sub)
+	if mono.IsSubscribeAsync(sending) {
+		sending.SubscribeWith(context.Background(), sub)
+	} else {
+		go func() {
+			sending.SubscribeWith(context.Background(), sub)
+		}()
+	}
+
 	return nil
 }
 
@@ -516,16 +522,19 @@ func (dc *DuplexConnection) respondRequestChannel(req fragmentation.HeaderAndPay
 	// Ensure registering message success before func end.
 	subscribed := make(chan struct{})
 
-	// Create subscriber
-	sub := respondChannelSubscriber{
-		sid:        sid,
-		n:          initRequestN,
-		dc:         dc,
-		rcv:        receivingProcessor,
-		subscribed: subscribed,
-		calls:      finallyRequests,
-	}
-	sending.SubscribeOn(scheduler.Parallel()).SubscribeWith(context.Background(), sub)
+	go func() {
+		// Create subscriber
+		sub := respondChannelSubscriber{
+			sid:        sid,
+			n:          initRequestN,
+			dc:         dc,
+			rcv:        receivingProcessor,
+			subscribed: subscribed,
+			calls:      finallyRequests,
+		}
+		sending.SubscribeWith(context.Background(), sub)
+	}()
+
 	<-subscribed
 
 	return nil
@@ -594,8 +603,11 @@ func (dc *DuplexConnection) respondRequestStream(receiving fragmentation.HeaderA
 		dc.writeError(sid, err)
 		return nil
 	}
+
 	// async subscribe publisher
-	sending.SubscribeOn(scheduler.Parallel()).SubscribeWith(context.Background(), borrowRequestStreamSubscriber(receiving, dc, sid, n))
+	sub := borrowRequestStreamSubscriber(receiving, dc, sid, n)
+	sending.SubscribeOn(scheduler.Parallel()).SubscribeWith(context.Background(), sub)
+
 	return nil
 }
 
@@ -610,7 +622,12 @@ func (dc *DuplexConnection) writeError(sid uint32, e error) {
 	case core.CustomError:
 		dc.sendFrame(framing.NewWriteableErrorFrame(sid, err.ErrorCode(), err.ErrorData()))
 	default:
-		dc.sendFrame(framing.NewWriteableErrorFrame(sid, core.ErrorCodeApplicationError, []byte(e.Error())))
+		errFrame := framing.NewWriteableErrorFrame(
+			sid,
+			core.ErrorCodeApplicationError,
+			bytesconv.StringToBytes(e.Error()),
+		)
+		dc.sendFrame(errFrame)
 	}
 }
 

payload/payload.go

@@ -6,6 +6,7 @@ import (
 	"time"
 
 	"github.com/rsocket/rsocket-go/core"
+	"github.com/rsocket/rsocket-go/internal/common"
 )
 
 type (
@@ -48,34 +49,26 @@ func Clone(payload Payload) Payload {
 	}
 	switch v := payload.(type) {
 	case *rawPayload:
-		var data []byte
-		if v.data != nil {
-			data = make([]byte, len(v.data))
-			copy(data, v.data)
+		data := common.CloneBytes(v.data)
+		metadata := common.CloneBytes(v.metadata)
+		return &rawPayload{
+			data:     data,
+			metadata: metadata,
 		}
-		var metadata []byte
-		if v.metadata != nil {
-			metadata = make([]byte, len(v.metadata))
-			copy(metadata, v.metadata)
-		}
-		return &rawPayload{data: data, metadata: metadata}
 	case *strPayload:
-		return &strPayload{data: v.data, metadata: v.metadata}
-	default:
-		ret := &rawPayload{}
-		if d := payload.Data(); len(d) > 0 {
-			clone := make([]byte, len(d))
-			copy(clone, d)
-			ret.data = clone
+		return &strPayload{
+			data:     v.data,
+			metadata: v.metadata,
 		}
-		if m, ok := payload.Metadata(); ok && len(m) > 0 {
-			clone := make([]byte, len(m))
-			copy(clone, m)
-			ret.metadata = clone
+	default:
+		data := common.CloneBytes(v.Data())
+		metadata, _ := v.Metadata()
+		metadata = common.CloneBytes(metadata)
+		return &rawPayload{
+			data:     data,
+			metadata: metadata,
 		}
-		return ret
 	}
-
 }
 
 // New create a new payload with bytes.

rx/mono/mono_bench_test.go

@@ -0,0 +1,27 @@
+package mono_test
+
+import (
+	"context"
+	"testing"
+
+	"github.com/rsocket/rsocket-go/payload"
+	"github.com/rsocket/rsocket-go/rx/mono"
+)
+
+var _fakePayload = payload.NewString("", "")
+
+func BenchmarkDefaultProxy(b *testing.B) {
+	b.RunParallel(func(pb *testing.PB) {
+		for pb.Next() {
+			mono.Just(_fakePayload).Subscribe(context.Background())
+		}
+	})
+}
+
+func BenchmarkOneshotProxy(b *testing.B) {
+	b.RunParallel(func(pb *testing.PB) {
+		for pb.Next() {
+			mono.JustOneshot(_fakePayload).Subscribe(context.Background())
+		}
+	})
+}

rx/mono/proxy.go → rx/mono/proxy_default.go

@@ -7,8 +7,6 @@ import (
 	"github.com/jjeffcaii/reactor-go"
 	"github.com/jjeffcaii/reactor-go/mono"
 	"github.com/jjeffcaii/reactor-go/scheduler"
-	"github.com/pkg/errors"
-	"github.com/rsocket/rsocket-go/internal/common"
 	"github.com/rsocket/rsocket-go/payload"
 	"github.com/rsocket/rsocket-go/rx"
 )
@@ -21,68 +19,28 @@ func newProxy(source mono.Mono) proxy {
 	return proxy{source}
 }
 
-func deepClone(any reactor.Any) (reactor.Any, error) {
-	src := any.(payload.Payload)
-	m, _ := src.Metadata()
-	return payload.New(common.CloneBytes(src.Data()), common.CloneBytes(m)), nil
-}
-
 func (p proxy) Raw() mono.Mono {
 	return p.Mono
 }
 
-func (p proxy) mustProcessor() mono.Processor {
-	m, ok := p.Mono.(mono.Processor)
-	if !ok {
-		panic(errors.Errorf("require processor but %v", p.Mono))
-	}
-	return m
-}
-
 func (p proxy) Success(v payload.Payload) {
-	p.mustProcessor().Success(v)
+	mustProcessor(p.Mono).Success(v)
 }
 
 func (p proxy) Error(e error) {
-	p.mustProcessor().Error(e)
+	mustProcessor(p.Mono).Error(e)
 }
 
 func (p proxy) ToChan(ctx context.Context) (<-chan payload.Payload, <-chan error) {
-	value := make(chan payload.Payload, 1)
-	err := make(chan error, 1)
-	p.subscribeWithChan(ctx, value, err, true)
-	return value, err
+	return toChan(ctx, p.Mono)
 }
 
 func (p proxy) SubscribeOn(sc scheduler.Scheduler) Mono {
 	return newProxy(p.Mono.SubscribeOn(sc))
 }
 
-func (p proxy) subscribeWithChan(ctx context.Context, valueChan chan<- payload.Payload, errChan chan<- error, autoClose bool) {
-	p.Mono.
-		DoFinally(func(s reactor.SignalType) {
-			if autoClose {
-				defer close(valueChan)
-				defer close(errChan)
-			}
-			if s == reactor.SignalTypeCancel {
-				errChan <- reactor.ErrSubscribeCancelled
-			}
-		}).
-		Subscribe(
-			ctx,
-			reactor.OnNext(func(v reactor.Any) error {
-				valueChan <- v.(payload.Payload)
-				return nil
-			}),
-			reactor.OnError(func(e error) {
-				errChan <- e
-			}),
-		)
-}
-
 func (p proxy) SubscribeWithChan(ctx context.Context, valueChan chan<- payload.Payload, errChan chan<- error) {
-	p.subscribeWithChan(ctx, valueChan, errChan, false)
+	subscribeWithChan(ctx, p.Mono, valueChan, errChan, false)
 }
 
 func (p proxy) Block(ctx context.Context) (pa payload.Payload, err error) {