Use bytebuff pool to reduce alloc memory. (#74)

Author: jjeffcaii

.travis.yml

@@ -16,7 +16,7 @@ install:
 
 script:
   - golangci-lint run ./...
-  - go test -race -coverprofile=coverage.txt -covermode=atomic ./...
+  - go test -coverprofile=coverage.txt -covermode=atomic ./...
 
 after_success:
   - bash <(curl -s https://codecov.io/bash)

client.go

@@ -62,7 +62,7 @@ type ClientBuilder interface {
 	// OnClose register handler when client socket closed.
 	OnClose(func(error)) ClientBuilder
 	// OnConnect register handler when client socket connected.
-	OnConnect(func(Client,error)) ClientBuilder
+	OnConnect(func(Client, error)) ClientBuilder
 	// Acceptor set acceptor for RSocket client.
 	Acceptor(acceptor ClientSocketAcceptor) ToClientStarter
 }
@@ -91,7 +91,7 @@ type clientBuilder struct {
 	setup          *socket.SetupInfo
 	acceptor       ClientSocketAcceptor
 	onCloses       []func(error)
-	onConnects     []func(Client,error)
+	onConnects     []func(Client, error)
 	connectTimeout time.Duration
 }
 
@@ -119,7 +119,7 @@ func (cb *clientBuilder) Fragment(mtu int) ClientBuilder {
 	return cb
 }
 
-func (cb *clientBuilder) OnConnect(fn func(Client,error)) ClientBuilder {
+func (cb *clientBuilder) OnConnect(fn func(Client, error)) ClientBuilder {
 	cb.onConnects = append(cb.onConnects, fn)
 	return cb
 }
@@ -215,11 +215,11 @@ func (cb *clientBuilder) Start(ctx context.Context) (client Client, err error) {
 
 	// trigger OnConnect
 	if len(cb.onConnects) > 0 {
-		var onConnects []func(Client,error)
+		var onConnects []func(Client, error)
 		onConnects, cb.onConnects = cb.onConnects, nil
 		go func() {
 			for _, onConnect := range onConnects {
-				onConnect(client,err)
+				onConnect(client, err)
 			}
 		}()
 	}

cmd/rsocket-cli/rsocket-cli.go

@@ -41,7 +41,7 @@ func main() {
 	app.UsageText = "rsocket-cli [global options] [URI]"
 	app.Name = "rsocket-cli"
 	app.Usage = "CLI for RSocket."
-	app.Version = "v0.6"
+	app.Version = "v0.7"
 	app.Flags = newFlags(conf)
 	app.ArgsUsage = "[URI]"
 	app.Action = func(c *cli.Context) (err error) {

cmd/rsocket-cli/runner.go

@@ -188,7 +188,7 @@ func (r *Runner) runServerMode(ctx context.Context) error {
 					r.showPayload(message)
 					return sendingPayloads
 				}))
-				options = append(options, rsocket.RequestChannel(func(messages rx.Publisher) flux.Flux {
+				options = append(options, rsocket.RequestChannel(func(messages flux.Flux) flux.Flux {
 					messages.Subscribe(ctx, rx.OnNext(func(input payload.Payload) error {
 						r.showPayload(input)
 						return nil

core/framing/frame.go

@@ -10,52 +10,83 @@ import (
 
 var errIncompleteFrame = errors.New("incomplete frame")
 
-type tinyFrame struct {
+type baseWriteableFrame struct {
 	header core.FrameHeader
 	done   chan struct{}
 }
 
-func (t *tinyFrame) Header() core.FrameHeader {
+// FromBytes creates frame from a byte slice.
+func FromBytes(b []byte) (f core.BufferedFrame, err error) {
+	if len(b) < core.FrameHeaderLen {
+		err = errIncompleteFrame
+		return
+	}
+	header := core.ParseFrameHeader(b[:core.FrameHeaderLen])
+	bb := common.BorrowByteBuff()
+	_, err = bb.Write(b[core.FrameHeaderLen:])
+	if err != nil {
+		common.ReturnByteBuff(bb)
+		return
+	}
+	raw := newBaseDefaultFrame(header, bb)
+	f, err = FromRawFrame(raw)
+	if err != nil {
+		common.ReturnByteBuff(bb)
+	}
+	return
+}
+
+func (t baseWriteableFrame) Header() core.FrameHeader {
 	return t.header
 }
 
 // Done can be invoked when a frame has been been processed.
-func (t *tinyFrame) Done() (closed bool) {
+func (t baseWriteableFrame) Done() (closed bool) {
 	defer func() {
-		if e := recover(); e != nil {
-			closed = true
-		}
+		closed = recover() != nil
 	}()
 	close(t.done)
 	return
 }
 
 // DoneNotify notify when frame has been done.
-func (t *tinyFrame) DoneNotify() <-chan struct{} {
+func (t baseWriteableFrame) DoneNotify() <-chan struct{} {
 	return t.done
 }
 
-// RawFrame is basic frame implementation.
-type RawFrame struct {
-	*tinyFrame
-	body *common.ByteBuff
+// baseDefaultFrame is basic frame implementation.
+type baseDefaultFrame struct {
+	header core.FrameHeader
+	body   *common.ByteBuff
+}
+
+func (f *baseDefaultFrame) Header() core.FrameHeader {
+	return f.header
+}
+
+// Release releases resource.
+func (f *baseDefaultFrame) Release() {
+	if f != nil && f.body != nil {
+		common.ReturnByteBuff(f.body)
+		f.body = nil
+	}
 }
 
 // Body returns frame body.
-func (f *RawFrame) Body() *common.ByteBuff {
+func (f *baseDefaultFrame) Body() *common.ByteBuff {
 	return f.body
 }
 
 // Len returns length of frame.
-func (f *RawFrame) Len() int {
+func (f *baseDefaultFrame) Len() int {
 	if f.body == nil {
 		return core.FrameHeaderLen
 	}
 	return core.FrameHeaderLen + f.body.Len()
 }
 
 // WriteTo write frame to writer.
-func (f *RawFrame) WriteTo(w io.Writer) (n int64, err error) {
+func (f *baseDefaultFrame) WriteTo(w io.Writer) (n int64, err error) {
 	var wrote int64
 	wrote, err = f.header.WriteTo(w)
 	if err != nil {
@@ -72,7 +103,7 @@ func (f *RawFrame) WriteTo(w io.Writer) (n int64, err error) {
 	return
 }
 
-func (f *RawFrame) trySeekMetadataLen(offset int) (n int, hasMetadata bool) {
+func (f *baseDefaultFrame) trySeekMetadataLen(offset int) (n int, hasMetadata bool) {
 	raw := f.body.Bytes()
 	if offset > 0 {
 		raw = raw[offset:]
@@ -89,15 +120,15 @@ func (f *RawFrame) trySeekMetadataLen(offset int) (n int, hasMetadata bool) {
 	return
 }
 
-func (f *RawFrame) trySliceMetadata(offset int) ([]byte, bool) {
+func (f *baseDefaultFrame) trySliceMetadata(offset int) ([]byte, bool) {
 	n, ok := f.trySeekMetadataLen(offset)
 	if !ok || n < 0 {
 		return nil, false
 	}
 	return f.body.Bytes()[offset+3 : offset+3+n], true
 }
 
-func (f *RawFrame) trySliceData(offset int) []byte {
+func (f *baseDefaultFrame) trySliceData(offset int) []byte {
 	n, ok := f.trySeekMetadataLen(offset)
 	if !ok {
 		return f.body.Bytes()[offset:]
@@ -108,32 +139,16 @@ func (f *RawFrame) trySliceData(offset int) []byte {
 	return f.body.Bytes()[offset+n+3:]
 }
 
-func newTinyFrame(header core.FrameHeader) *tinyFrame {
-	return &tinyFrame{
+func newBaseWriteableFrame(header core.FrameHeader) baseWriteableFrame {
+	return baseWriteableFrame{
 		header: header,
 		done:   make(chan struct{}),
 	}
 }
 
-// NewRawFrame returns a new RawFrame.
-func NewRawFrame(header core.FrameHeader, body *common.ByteBuff) *RawFrame {
-	return &RawFrame{
-		tinyFrame: newTinyFrame(header),
-		body:      body,
-	}
-}
-
-// FromBytes creates frame from a byte slice.
-func FromBytes(b []byte) (core.Frame, error) {
-	if len(b) < core.FrameHeaderLen {
-		return nil, errIncompleteFrame
-	}
-	header := core.ParseFrameHeader(b[:core.FrameHeaderLen])
-	bb := common.NewByteBuff()
-	_, err := bb.Write(b[core.FrameHeaderLen:])
-	if err != nil {
-		return nil, err
+func newBaseDefaultFrame(header core.FrameHeader, body *common.ByteBuff) *baseDefaultFrame {
+	return &baseDefaultFrame{
+		header: header,
+		body:   body,
 	}
-	raw := NewRawFrame(header, bb)
-	return FromRawFrame(raw)
 }

core/framing/frame_cancel.go

@@ -8,12 +8,12 @@ import (
 
 // CancelFrame is frame of Cancel.
 type CancelFrame struct {
-	*RawFrame
+	*baseDefaultFrame
 }
 
 // WriteableCancelFrame is writeable frame of Cancel.
 type WriteableCancelFrame struct {
-	*tinyFrame
+	baseWriteableFrame
 }
 
 // WriteTo writes current frame to given writer.
@@ -45,13 +45,13 @@ func (f *CancelFrame) Validate() (err error) {
 func NewWriteableCancelFrame(id uint32) *WriteableCancelFrame {
 	h := core.NewFrameHeader(id, core.FrameTypeCancel, 0)
 	return &WriteableCancelFrame{
-		tinyFrame: newTinyFrame(h),
+		baseWriteableFrame: newBaseWriteableFrame(h),
 	}
 }
 
 // NewCancelFrame creates cancel frame.
 func NewCancelFrame(sid uint32) *CancelFrame {
 	return &CancelFrame{
-		NewRawFrame(core.NewFrameHeader(sid, core.FrameTypeCancel, 0), nil),
+		newBaseDefaultFrame(core.NewFrameHeader(sid, core.FrameTypeCancel, 0), nil),
 	}
 }

core/framing/frame_error.go

@@ -17,19 +17,59 @@ const (
 
 // ErrorFrame is error frame.
 type ErrorFrame struct {
-	*RawFrame
+	*baseDefaultFrame
 }
 
-// WriteableErrorFrame is writeable error frame.
-type WriteableErrorFrame struct {
-	*tinyFrame
+type frozenError struct {
 	code core.ErrorCode
 	data []byte
 }
 
-// Error returns error string.
-func (e WriteableErrorFrame) Error() string {
-	return makeErrorString(e.code, e.data)
+// WriteableErrorFrame is writeable error frame.
+type WriteableErrorFrame struct {
+	baseWriteableFrame
+	frozenError
+}
+
+// NewWriteableErrorFrame creates WriteableErrorFrame.
+func NewWriteableErrorFrame(id uint32, code core.ErrorCode, data []byte) *WriteableErrorFrame {
+	h := core.NewFrameHeader(id, core.FrameTypeError, 0)
+	t := newBaseWriteableFrame(h)
+	return &WriteableErrorFrame{
+		baseWriteableFrame: t,
+		frozenError: frozenError{
+			code: code,
+			data: data,
+		},
+	}
+}
+
+// NewErrorFrame returns a new error frame.
+func NewErrorFrame(streamID uint32, code core.ErrorCode, data []byte) *ErrorFrame {
+	b := common.BorrowByteBuff()
+	if err := binary.Write(b, binary.BigEndian, uint32(code)); err != nil {
+		common.ReturnByteBuff(b)
+		panic(err)
+	}
+	if _, err := b.Write(data); err != nil {
+		common.ReturnByteBuff(b)
+		panic(err)
+	}
+	return &ErrorFrame{
+		newBaseDefaultFrame(core.NewFrameHeader(streamID, core.FrameTypeError, 0), b),
+	}
+}
+
+func (c frozenError) Error() string {
+	return makeErrorString(c.ErrorCode(), c.ErrorData())
+}
+
+func (c frozenError) ErrorCode() core.ErrorCode {
+	return c.code
+}
+
+func (c frozenError) ErrorData() []byte {
+	return c.data
 }
 
 // WriteTo writes frame to writer.
@@ -60,6 +100,13 @@ func (e WriteableErrorFrame) Len() int {
 	return core.FrameHeaderLen + 4 + len(e.data)
 }
 
+func (p *ErrorFrame) ToError() error {
+	return frozenError{
+		code: p.ErrorCode(),
+		data: common.CloneBytes(p.ErrorData()),
+	}
+}
+
 // Validate returns error if frame is invalid.
 func (p *ErrorFrame) Validate() (err error) {
 	if p.body.Len() < minErrorFrameLen {
@@ -84,33 +131,6 @@ func (p *ErrorFrame) ErrorData() []byte {
 	return p.body.Bytes()[errDataOff:]
 }
 
-// NewWriteableErrorFrame creates WriteableErrorFrame.
-func NewWriteableErrorFrame(id uint32, code core.ErrorCode, data []byte) *WriteableErrorFrame {
-	h := core.NewFrameHeader(id, core.FrameTypeError, 0)
-	t := newTinyFrame(h)
-	return &WriteableErrorFrame{
-		tinyFrame: t,
-		code:      code,
-		data:      data,
-	}
-}
-
-// NewErrorFrame returns a new error frame.
-func NewErrorFrame(streamID uint32, code core.ErrorCode, data []byte) *ErrorFrame {
-	bf := common.NewByteBuff()
-	var b4 [4]byte
-	binary.BigEndian.PutUint32(b4[:], uint32(code))
-	if _, err := bf.Write(b4[:]); err != nil {
-		panic(err)
-	}
-	if _, err := bf.Write(data); err != nil {
-		panic(err)
-	}
-	return &ErrorFrame{
-		NewRawFrame(core.NewFrameHeader(streamID, core.FrameTypeError, 0), bf),
-	}
-}
-
 func makeErrorString(code core.ErrorCode, data []byte) string {
 	bu := strings.Builder{}
 	bu.WriteString(code.String())