socket

Documentation

Index

• type AbstractRSocket
  • func (p AbstractRSocket) FireAndForget(msg payload.Payload)
  • func (p AbstractRSocket) MetadataPush(msg payload.Payload)
  • func (p AbstractRSocket) RequestChannel(msgs rx.Publisher) rx.Flux
  • func (p AbstractRSocket) RequestResponse(msg payload.Payload) rx.Mono
  • func (p AbstractRSocket) RequestStream(msg payload.Payload) rx.Flux
• type ClientSocket
  • func NewClient(uri *transport.URI, socket *DuplexRSocket) ClientSocket
  • func NewClientResume(uri *transport.URI, socket *DuplexRSocket) ClientSocket
• type Closeable
• type DuplexRSocket
  • func NewClientDuplexRSocket(mtu int, scheduler rx.Scheduler, keepaliveInterval time.Duration) (s *DuplexRSocket)
  • func NewServerDuplexRSocket(mtu int, scheduler rx.Scheduler) *DuplexRSocket
  • func (p *DuplexRSocket) Close() (err error)
  • func (p *DuplexRSocket) FireAndForget(sending payload.Payload)
  • func (p *DuplexRSocket) MetadataPush(payload payload.Payload)
  • func (p *DuplexRSocket) RequestChannel(publisher rx.Publisher) rx.Flux
  • func (p *DuplexRSocket) RequestResponse(pl payload.Payload) rx.Mono
  • func (p *DuplexRSocket) RequestStream(sending payload.Payload) rx.Flux
  • func (p *DuplexRSocket) SetResponder(responder Responder)
  • func (p *DuplexRSocket) SetTransport(tp *transport.Transport)
• type Responder
• type ServerSocket
  • func NewServer(socket *DuplexRSocket) ServerSocket
  • func NewServerResume(socket *DuplexRSocket, token []byte) ServerSocket
• type SetupInfo
  • func (p *SetupInfo) ToFrame() *framing.FrameSetup

Types

type AbstractRSocket

type AbstractRSocket struct {
	FF func(payload.Payload)
	MP func(payload.Payload)
	RR func(payload.Payload) rx.Mono
	RS func(payload.Payload) rx.Flux
	RC func(rx.Publisher) rx.Flux
}

AbstractRSocket represents an abstract RSocket.

func (AbstractRSocket) FireAndForget

func (p AbstractRSocket) FireAndForget(msg payload.Payload)

FireAndForget starts a request of FireAndForget.

func (AbstractRSocket) MetadataPush

func (p AbstractRSocket) MetadataPush(msg payload.Payload)

MetadataPush starts a request of MetadataPush.

func (AbstractRSocket) RequestChannel

func (p AbstractRSocket) RequestChannel(msgs rx.Publisher) rx.Flux

RequestChannel starts a request of RequestChannel.

func (AbstractRSocket) RequestResponse

func (p AbstractRSocket) RequestResponse(msg payload.Payload) rx.Mono

RequestResponse starts a request of RequestResponse.

func (AbstractRSocket) RequestStream

func (p AbstractRSocket) RequestStream(msg payload.Payload) rx.Flux

RequestStream starts a request of RequestStream.

type ClientSocket

type ClientSocket interface {
	Closeable
	Responder
	// Setup setups current socket.
	Setup(ctx context.Context, setup *SetupInfo) (err error)
}

ClientSocket represents a client-side socket.

func NewClient

func NewClient(uri *transport.URI, socket *DuplexRSocket) ClientSocket

NewClient create a simple client-side socket.

func NewClientResume

func NewClientResume(uri *transport.URI, socket *DuplexRSocket) ClientSocket

NewClientResume creates a client-side socket with resume support.

type Closeable

type Closeable interface {
	io.Closer
	// OnClose bind a handler when closing.
	OnClose(closer func())
}

Closeable represents a closeable target.

type DuplexRSocket

type DuplexRSocket struct {
	// contains filtered or unexported fields
}

DuplexRSocket represents a socket of RSocket which can be a requester or a responder.

func NewClientDuplexRSocket

func NewClientDuplexRSocket(
	mtu int,
	scheduler rx.Scheduler,
	keepaliveInterval time.Duration,
) (s *DuplexRSocket)

NewClientDuplexRSocket creates a new client-side DuplexRSocket.

func NewServerDuplexRSocket

func NewServerDuplexRSocket(mtu int, scheduler rx.Scheduler) *DuplexRSocket

NewServerDuplexRSocket creates a new server-side DuplexRSocket.

func (*DuplexRSocket) Close

func (p *DuplexRSocket) Close() (err error)

Close close current socket.

func (*DuplexRSocket) FireAndForget

func (p *DuplexRSocket) FireAndForget(sending payload.Payload)

FireAndForget start a request of FireAndForget.

func (*DuplexRSocket) MetadataPush

func (p *DuplexRSocket) MetadataPush(payload payload.Payload)

MetadataPush start a request of MetadataPush.

func (*DuplexRSocket) RequestChannel

func (p *DuplexRSocket) RequestChannel(publisher rx.Publisher) rx.Flux

RequestChannel start a request of RequestChannel.

func (*DuplexRSocket) RequestResponse

func (p *DuplexRSocket) RequestResponse(pl payload.Payload) rx.Mono

RequestResponse start a request of RequestResponse.

func (*DuplexRSocket) RequestStream

func (p *DuplexRSocket) RequestStream(sending payload.Payload) rx.Flux

RequestStream start a request of RequestStream.

func (*DuplexRSocket) SetResponder

func (p *DuplexRSocket) SetResponder(responder Responder)

SetResponder sets a responder for current socket.

func (*DuplexRSocket) SetTransport

func (p *DuplexRSocket) SetTransport(tp *transport.Transport)

SetTransport sets a transport for current socket.

type Responder

type Responder interface {
	// FireAndForget is a single one-way message.
	FireAndForget(msg payload.Payload)
	// MetadataPush sends asynchronous Metadata frame.
	MetadataPush(msg payload.Payload)
	// RequestResponse request single response.
	RequestResponse(msg payload.Payload) rx.Mono
	// RequestStream request a completable stream.
	RequestStream(msg payload.Payload) rx.Flux
	// RequestChannel request a completable stream in both directions.
	RequestChannel(msgs rx.Publisher) rx.Flux
}

Responder is a contract providing different interaction models for RSocket protocol.

type ServerSocket

type ServerSocket interface {
	Closeable
	Responder
	// SetResponder sets a responder for current socket.
	SetResponder(responder Responder)
	// SetTransport sets a transport for current socket.
	SetTransport(tp *transport.Transport)
	// Pause pause current socket.
	Pause() bool
	// Start starts current socket.
	Start(ctx context.Context) error
	// Token returns token of socket.
	Token() (token []byte, ok bool)
}

ServerSocket represents a server-side socket.

func NewServer

func NewServer(socket *DuplexRSocket) ServerSocket

NewServer creates a new server-side socket.

func NewServerResume

func NewServerResume(socket *DuplexRSocket, token []byte) ServerSocket

NewServerResume creates a new server-side socket with resume support.

type SetupInfo

type SetupInfo struct {
	Version           common.Version
	KeepaliveInterval time.Duration
	KeepaliveLifetime time.Duration
	Token             []byte
	DataMimeType      []byte
	Data              []byte
	MetadataMimeType  []byte
	Metadata          []byte
}

SetupInfo represents basic info of setup.

func (*SetupInfo) ToFrame

func (p *SetupInfo) ToFrame() *framing.FrameSetup

ToFrame converts current SetupInfo to a frame of Setup.