feat(frame): replace framing to Y3 frame (#211) (#225)

* feat(frame): replace framing to Y3 frame

* fix(heartbeat): client sends ping and zipper responses pong

* feat(frame): add FrameType String method

* sytle: fix the issues in codacy

Co-authored-by: Hong Xiaojian <xiaojian.hong@gmail.com>

Author: venjiang

.github/workflows/go.yml

@@ -15,7 +15,7 @@ jobs:
     - name: Set up Go 1.x
       uses: actions/setup-go@v2
       with:
-        go-version: 1.15
+        go-version: 1.16
 
     - name: Test
       run: go test $(go list ./... | grep -v /example)

core/quic/conn.go

@@ -4,8 +4,8 @@ import (
 	"errors"
 	"time"
 
-	"github.com/yomorun/yomo/internal/decoder"
-	"github.com/yomorun/yomo/internal/framing"
+	"github.com/yomorun/yomo/internal/core"
+	"github.com/yomorun/yomo/internal/frame"
 )
 
 const (
@@ -24,9 +24,9 @@ const (
 // Conn represents the QUIC connection.
 type Conn struct {
 	// Signal is the specified stream to receive the signal.
-	Signal decoder.ReadWriter
+	Signal *core.FrameStream
 	// Type is the type of connection. Possible value: source, stream-function, server-sender.
-	Type string
+	Type core.ConnectionType
 	// Name is the name of connection.
 	Name string
 	// Heartbeat is the channel to receive heartbeat.
@@ -44,7 +44,7 @@ type Conn struct {
 }
 
 // NewConn inits a new QUIC connection.
-func NewConn(name string, connType string) *Conn {
+func NewConn(name string, connType core.ConnectionType) *Conn {
 	return &Conn{
 		Name:      name,
 		Type:      connType,
@@ -55,12 +55,12 @@ func NewConn(name string, connType string) *Conn {
 }
 
 // SendSignal sends the signal to client.
-func (c *Conn) SendSignal(f framing.Frame) error {
+func (c *Conn) SendSignal(f frame.Frame) error {
 	if c.Signal == nil {
 		return errors.New("Signal is nil")
 	}
 
-	err := c.Signal.Write(f)
+	_, err := c.Signal.WriteFrame(f)
 	return err
 }
 

core/rx/factory.go

@@ -4,7 +4,10 @@ import (
 	"context"
 
 	"github.com/reactivex/rxgo/v2"
+	"github.com/yomorun/yomo/core/quic"
+	"github.com/yomorun/yomo/internal/core"
 	"github.com/yomorun/yomo/internal/decoder"
+	"github.com/yomorun/yomo/internal/frame"
 	"github.com/yomorun/yomo/logger"
 )
 
@@ -13,11 +16,8 @@ type Factory interface {
 	// FromChannel creates a new Stream from a channel.
 	FromChannel(ctx context.Context, channel chan interface{}) Stream
 
-	// FromReader creates a new Stream from decoder.Reader.
-	FromReader(ctx context.Context, reader decoder.Reader) Stream
-
-	// FromReaderWithDecoder creates a new Stream with decoder.
-	FromReaderWithDecoder(readers chan decoder.Reader, opts ...decoder.Option) Stream
+	// FromQuicStream creates a new Stream from QuicStream.
+	FromQuicStream(ctx context.Context, stream quic.Stream) Stream
 
 	// FromItems creates a new Stream from items.
 	FromItems(ctx context.Context, items []interface{}) Stream
@@ -60,56 +60,32 @@ func (fac *factoryImpl) FromChannel(ctx context.Context, channel chan interface{
 	return CreateObservable(ctx, f)
 }
 
-// FromReader creates a new Stream from decoder.Reader.
-func (fac *factoryImpl) FromReader(ctx context.Context, reader decoder.Reader) Stream {
+// FromQuicStream creates a new RxStream from QUIC Stream.
+func (fac *factoryImpl) FromQuicStream(ctx context.Context, stream quic.Stream) Stream {
 	f := func(ctx context.Context, next chan rxgo.Item) {
 		defer close(next)
 
-		frameChan := reader.Read()
 		for {
-			select {
-			case <-ctx.Done():
-				return
-			case frame, ok := <-frameChan:
-				if !ok {
-					return
-				}
+			f, err := core.ParseFrame(stream)
+			if err != nil {
+				logger.Error("Parse the frame failed", "err", err)
+				break
+			}
 
-				logger.Debug("Receive frame from source.")
-				next <- Of(frame.Data())
+			switch f.Type() {
+			case frame.TagOfDataFrame:
+				dataFrame := f.(*frame.DataFrame)
+				logger.Debug("Receive data frame from source.", "TransactionID", dataFrame.TransactionID())
+				next <- Of(dataFrame.GetCarriage())
+			default:
+				logger.Debug("Only support data frame in RxStream.", "type", f.Type())
 			}
 		}
 	}
 
 	return CreateObservable(ctx, f)
 }
 
-// FromReaderWithDecoder creates a Stream with decoder.
-func (fac *factoryImpl) FromReaderWithDecoder(readers chan decoder.Reader, opts ...decoder.Option) Stream {
-	f := func(ctx context.Context, next chan rxgo.Item) {
-		defer close(next)
-
-		for {
-			select {
-			case <-ctx.Done():
-				return
-			case reader, ok := <-readers:
-				if !ok {
-					return
-				}
-
-				go func() {
-					frameChan := reader.Read()
-					for frame := range frameChan {
-						Of(decoder.FromItems([]interface{}{frame.Data()}, opts...)).SendContext(ctx, next)
-					}
-				}()
-			}
-		}
-	}
-	return CreateObservable(decoder.GetContext(opts...), f)
-}
-
 // FromItems creates a new Stream from items.
 func (fac *factoryImpl) FromItems(ctx context.Context, items []interface{}) Stream {
 	next := make(chan rxgo.Item)

core/rx/stream_operator_test.go

@@ -10,7 +10,6 @@ import (
 	"github.com/reactivex/rxgo/v2"
 	"github.com/stretchr/testify/assert"
 	y3 "github.com/yomorun/y3-codec-golang"
-	"github.com/yomorun/yomo/internal/decoder"
 )
 
 // HELPER FUNCTIONS
@@ -252,16 +251,16 @@ func Test_Subscribe_MultipleKeys(t *testing.T) {
 	})
 }
 
-func Test_RawBytes(t *testing.T) {
-	buf := &bytes.Buffer{}
-	buf.Write([]byte{0, 0, 6, 0, 0, 0, 1, 2, 3})
-	obs := decoder.FromStream(decoder.NewReader(buf))
-	rawBytes := obs.RawBytes()
-	for b := range rawBytes {
-		assert.Equal(t, []byte{1, 2, 3}, b)
-		break
-	}
-}
+// func Test_RawBytes(t *testing.T) {
+// 	buf := &bytes.Buffer{}
+// 	buf.Write([]byte{0, 0, 6, 0, 0, 0, 1, 2, 3})
+// 	obs := decoder.FromStream(decoder.NewReader(buf))
+// 	rawBytes := obs.RawBytes()
+// 	for b := range rawBytes {
+// 		assert.Equal(t, []byte{1, 2, 3}, b)
+// 		break
+// 	}
+// }
 
 func Test_Encode(t *testing.T) {
 	t.Run("string", func(t *testing.T) {

go.mod

@@ -1,6 +1,6 @@
 module github.com/yomorun/yomo
 
-go 1.15
+go 1.16
 
 require (
 	github.com/cenkalti/backoff/v4 v4.1.1
@@ -10,6 +10,7 @@ require (
 	github.com/stretchr/objx v0.1.1 // indirect
 	github.com/stretchr/testify v1.7.0
 	github.com/tidwall/gjson v1.8.1
+	github.com/yomorun/y3 v1.0.2
 	github.com/yomorun/y3-codec-golang v1.7.0
 	go.opentelemetry.io/otel v1.0.0-RC2
 	go.opentelemetry.io/otel/exporters/jaeger v1.0.0-RC2

go.sum

@@ -172,6 +172,8 @@ github.com/tidwall/pretty v1.1.0 h1:K3hMW5epkdAVwibsQEfR/7Zj0Qgt4DxtNumTq/VloO8=
 github.com/tidwall/pretty v1.1.0/go.mod h1:XNkn88O1ChpSDQmQeStsy+sBenx6DDtFZJxhVysOjyk=
 github.com/viant/assertly v0.4.8/go.mod h1:aGifi++jvCrUaklKEKT0BU95igDNaqkvz+49uaYMPRU=
 github.com/viant/toolbox v0.24.0/go.mod h1:OxMCG57V0PXuIP2HNQrtJf2CjqdmbrOx5EkMILuUhzM=
+github.com/yomorun/y3 v1.0.2 h1:eugmFj2dV5om53A/AxQt++SJVAWeOdLjD5NSgngbO8A=
+github.com/yomorun/y3 v1.0.2/go.mod h1:+zwvZrKHe8D3fTMXNTsUsZXuI+kYxv3LRA2fSJEoWbo=
 github.com/yomorun/y3-codec-golang v1.7.0 h1:UYL6zuj132dnqG4N1MHuSY4h3DmC/6DfkMjNtTlugKI=
 github.com/yomorun/y3-codec-golang v1.7.0/go.mod h1:R+y8hQ/AHZ1tDzWtmspVeX7omqVWFJ42gdlXIOp35rA=
 github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=