RSocket Go bug report

rsocket_test.go

//nolint:funlen
package client_test

import (
	"context"
	"fmt"
	"testing"

	"github.com/rsocket/rsocket-go"
	"github.com/rsocket/rsocket-go/extension"
	"github.com/rsocket/rsocket-go/payload"
	"github.com/rsocket/rsocket-go/rx/flux"
)

func TestRsocket(t *testing.T) {
	readyCh := make(chan struct{})
	go rsocketServer(readyCh)
	<-readyCh
	rsocketClient()
}

func rsocketServer(readyCh chan struct{}) {
	requestStreamHandler := rsocket.RequestStream(func(request payload.Payload) flux.Flux {
		f := flux.Create(func(ctx context.Context, sink flux.Sink) {
			defer sink.Complete()
			for i := 0; i < 10; i++ {
				sink.Next(payload.NewString(fmt.Sprintf("%d", i), extension.TextPlain.String()))
			}
		})

		return f
	})

	err := rsocket.Receive().
		OnStart(func() { close(readyCh) }).
		Acceptor(func(ctx context.Context, setup payload.SetupPayload, sendingSocket rsocket.CloseableRSocket) (rsocket.RSocket, error) {
			return rsocket.NewAbstractSocket(requestStreamHandler), nil
		}).
		Transport(rsocket.TCPServer().SetAddr(":7878").Build()).
		Serve(context.Background())

	panic(err)
}

func rsocketClient() {
	tp := rsocket.TCPClient().SetHostAndPort("127.0.0.1", 7878).Build()
	client, err := rsocket.Connect().Transport(tp).Start(context.Background())
	if err != nil {
		panic(err)
	}
	defer client.Close()

	res, err := client.RequestStream(payload.Empty()).BlockSlice(context.Background())
	if err != nil {
		panic(err)
	}

	for i := range res {
		fmt.Println(res[i].Data())
	}
}