rsocket

rsocket-go

rsocket-go is an implementation of the RSocket protocol in Go. It is still under development, APIs are unstable and maybe change at any time until release of v1.0.0. Please do not use it in a production environment.

Features

• Design For Golang.
• Thin reactive-streams implementation.
• Simulate Java SDK API.

Getting started

Start an echo server

package main

import (
	"github.com/rsocket/rsocket-go"
	"github.com/rsocket/rsocket-go/payload"
	"github.com/rsocket/rsocket-go/rx"
)

func main() {
	// Create and serve
	err := rsocket.Receive().
		Acceptor(func(setup payload.SetupPayload, sendingSocket rsocket.EnhancedRSocket) rsocket.RSocket {
			// bind responder
			return rsocket.NewAbstractSocket(
				rsocket.RequestResponse(func(msg payload.Payload) rx.Mono {
					return rx.JustMono(msg)
				}),
			)
		}).
		Transport("tcp://127.0.0.1:7878").
		Serve()
	panic(err)
}

Connect to echo server

package main

import (
	"context"
	"log"
	
	"github.com/rsocket/rsocket-go"
	"github.com/rsocket/rsocket-go/payload"
	"github.com/rsocket/rsocket-go/rx"
)

func main() {
	// Connect to server
	client, err := rsocket.Connect().
		SetupPayload(payload.NewString("Hello", "World")).
		Transport("tcp://127.0.0.1:7878").
		Start()
	if err != nil {
		panic(err)
	}
	defer client.Close()
	// Send request
	client.RequestResponse(payload.NewString("你好", "世界")).
		DoOnSuccess(func(ctx context.Context, s rx.Subscription, elem payload.Payload) {
			log.Println("receive response:", elem)
		}).
		Subscribe(context.Background())
}

NOTICE: more server examples are Here

Advanced

Load Balance

Basic load balance feature is in preview, please checkout current master branch. It's a client side load-balancer.

Here're some example codes:

package main

import (
	"context"
	"fmt"
	"log"
	"strings"

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

func main() {
	// Use a chan as discovery for brokers. 
	// Writing a string slice will refresh the brokers of load balancer.  
	brokers := make(chan []string, 0)
	// Load Balance Example
	// It's EASY!!! Just set more than one transport URI.
	setup := payload.NewString("こんにちは、世界!", "Go")
	clientLB, err := rsocket.Connect().
		SetupPayload(setup).
		Acceptor(func(socket rsocket.RSocket) rsocket.RSocket {
			return rsocket.NewAbstractSocket(
				rsocket.FireAndForget(func(msg payload.Payload) {
					// For example:
					// You can refresh brokers list using FNF payload from server.
					brokers <- strings.Split(msg.DataUTF8(), ",")
				}),
            )
		}).
		Transports(brokers, rsocket.WithInitTransports("tcp://127.0.0.1:8000", "tcp://127.0.0.1:7878")).
		Start()
	if err != nil {
		panic(err)
	}
	defer func() {
		_ = clientLB.Close()
	}()
	for i := 0; i < 100; i++ {
		clientLB.RequestResponse(payload.NewString("Hello World!", fmt.Sprintf("%d", i))).
			DoOnError(func(ctx context.Context, err error) {
				log.Println("oops:", err)
			}).
			DoOnSuccess(func(ctx context.Context, s rx.Subscription, elem payload.Payload) {
				log.Println("okey:", elem)
			}).
			Subscribe(context.Background())
	}
}

NOTICE: example codes are here

Reactor API

Mono and Flux are two parts of Reactor API.

Mono

Mono completes successfully by emitting an element, or with an error. Here is a tiny example:

package main

import (
	"context"
	
	"github.com/rsocket/rsocket-go/payload"
	"github.com/rsocket/rsocket-go/rx"
)

func main() {
	// Create a Mono which produce a simple payload.
	mono := rx.NewMono(func(ctx context.Context, sink rx.MonoProducer) {
		// Use context API if you want.
		sink.Success(payload.NewString("foo", "bar"))
	})

	done := make(chan struct{})

	mono.
		DoFinally(func(ctx context.Context, st rx.SignalType) {
			close(done)
		}).
		DoOnSuccess(func(ctx context.Context, s rx.Subscription, elem payload.Payload) {
			// Handle and consume payload.
			// Do something here...
		}).
		SubscribeOn(rx.ElasticScheduler()).
		Subscribe(context.Background())

	<-done
}

Flux

Flux emits 0 to N elements, and then completes (successfully or with an error). Here is tiny example:

package main

import (
	"context"
	"time"
	
	"github.com/rsocket/rsocket-go/payload"
	"github.com/rsocket/rsocket-go/rx"
)

func main() {
	// Create a Flux and produce 10 elements.
	flux := rx.NewFlux(func(ctx context.Context, producer rx.Producer) {
		for i := 0; i < 10; i++ {
			producer.Next(payload.NewString("hello", time.Now().String()))
		}
		producer.Complete()
	})
	flux.DoOnNext(func(ctx context.Context, s rx.Subscription, elem payload.Payload) {
			// Handle and consume elements
			// Do something here...
		}).
		Subscribe(context.Background())
}

Backpressure & RequestN

Flux support backpressure.

You can call func Request in Subscription or use LimitRate before subscribe.

// Here is an example which consume Payload one by one.
flux.Subscribe(
    context.Background(),
    rx.OnSubscribe(func(ctx context.Context, s rx.Subscription) {
        // Init Request 1 element.
        s.Request(1)
    }),
    rx.OnNext(func(ctx context.Context, s rx.Subscription, elem payload.Payload) {
        // Consume element, do something...

        // Request for next one manually.
        s.Request(1)
    }),
)

Dependencies

• ants
• bytebufferpool
• testify
• websocket

TODO

Transport

• TCP
• Websocket
• Aeron
• QUIC

Duplex Socket

• MetadataPush
• RequestFNF
• RequestResponse
• RequestStream
• RequestChannel

• Resume
• Keepalive
• Fragmentation
• Thin Reactor
• Cancel
• Error
• Flow Control: RequestN
• Flow Control: Lease
• Load Balance
• Reconnect