feat(rx): add y3Encode (#127)

Co-authored-by: Njjwygjj <wangyaoguang@cel.la>

example/flow/app.go

@@ -43,7 +43,8 @@ func Handler(rxstream rx.RxStream) rx.RxStream {
 		OnObserve(callback).
 		Debounce(rxgo.WithDuration(50 * time.Millisecond)).
 		Map(printer).
-		StdOut()
+		StdOut().
+		Encode(0x10)
 
 	return stream
 }

example/sink/app.go

@@ -25,6 +25,7 @@ func Handler(rxstream rx.RxStream) rx.RxStream {
 		Subscribe(0x11).
 		OnObserve(callback).
 		AuditTime(100 * time.Millisecond).
-		Map(store)
+		Map(store).
+		Encode(0x10)
 	return stream
 }

internal/cmd/run.go

@@ -7,7 +7,6 @@ import (
 	"plugin"
 
 	"github.com/spf13/cobra"
-	y3 "github.com/yomorun/y3-codec-golang"
 	"github.com/yomorun/yomo/internal/dispatcher"
 	"github.com/yomorun/yomo/internal/serverless"
 	"github.com/yomorun/yomo/pkg/quic"
@@ -19,8 +18,6 @@ type RunOptions struct {
 	baseOptions
 	// Port is the port number of UDP host for Serverless function (default is 4242).
 	Port int
-
-	mockSink bool
 }
 
 // NewCmdRun creates a new command run.
@@ -47,7 +44,6 @@ func NewCmdRun() *cobra.Command {
 			endpoint := fmt.Sprintf("0.0.0.0:%d", opts.Port)
 			quicHandler := &quicServerHandler{
 				serverlessHandle: slHandler,
-				mockSink:         opts.mockSink,
 			}
 
 			err = serverless.Run(endpoint, quicHandler)
@@ -59,14 +55,12 @@ func NewCmdRun() *cobra.Command {
 
 	cmd.Flags().StringVarP(&opts.Filename, "file-name", "f", "app.go", "Serverless function file (default is app.go)")
 	cmd.Flags().IntVarP(&opts.Port, "port", "p", 4242, "Port is the port number of UDP host for Serverless function (default is 4242)")
-	cmd.Flags().BoolVar(&opts.mockSink, "mock-sink", false, "Indicates whether the Serverless is a mock sink")
 
 	return cmd
 }
 
 type quicServerHandler struct {
 	serverlessHandle plugin.Symbol
-	mockSink         bool
 }
 
 func (s quicServerHandler) Listen() error {
@@ -76,21 +70,12 @@ func (s quicServerHandler) Listen() error {
 func (s quicServerHandler) Read(st quic.Stream) error {
 	stream := dispatcher.Dispatcher(s.serverlessHandle, rx.FromReaderWithY3(st))
 
-	y3codec := y3.NewCodec(0x10)
-
 	go func() {
 		for customer := range stream.Observe() {
 			if customer.Error() {
 				fmt.Println(customer.E.Error())
 			} else if customer.V != nil {
-				// HACK
-				if s.mockSink {
-					st.Write([]byte("Finish sink!"))
-				} else {
-					// use Y3 codec to encode the data
-					sendingBuf, _ := y3codec.Marshal(customer.V)
-					st.Write(sendingBuf)
-				}
+				st.Write((customer.V).([]byte))
 			}
 		}
 	}()

pkg/rx/rxstream.go

@@ -13,6 +13,7 @@ type RxStream interface {
 	rxgo.Iterable
 	MergeReadWriterWithFunc(rwf func() (io.ReadWriter, func()), opts ...rxgo.Option) RxStream
 	Subscribe(key byte) RxStream
+	Encode(key byte, opts ...rxgo.Option) RxStream
 	OnObserve(function func(v []byte) (interface{}, error)) RxStream
 	StdOut(opts ...rxgo.Option) RxStream
 	AuditTime(timespan time.Duration, opts ...rxgo.Option) RxStream

pkg/rx/rxstream_operator.go

@@ -571,6 +571,42 @@ func (s *RxStreamImpl) OnObserve(function func(v []byte) (interface{}, error)) R
 	return CreateObservable(f)
 }
 
+func (s *RxStreamImpl) Encode(key byte, opts ...rxgo.Option) RxStream {
+	y3codec := y3.NewCodec(key)
+
+	f := func(ctx context.Context, next chan rxgo.Item) {
+		defer close(next)
+		observe := s.Observe(opts...)
+
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case item, ok := <-observe:
+				if !ok {
+					return
+				}
+
+				if item.Error() {
+					return
+				}
+
+				buf, err := y3codec.Marshal(item.V)
+
+				if err != nil {
+					return
+				}
+
+				if !Of(buf).SendContext(ctx, next) {
+					return
+				}
+
+			}
+		}
+	}
+	return CreateObservable(f, opts...)
+}
+
 func CreateObservable(f func(ctx context.Context, next chan rxgo.Item), opts ...rxgo.Option) RxStream {
 	next := make(chan rxgo.Item)
 	ctx := context.Background()