pmtp

Documentation

Index

• Constants
• Variables
• type Client
  • func Connect(uri string) (Client, error)
  • func Dial(url string) (Client, error)
  • func DialContext(ctx context.Context, url string) (Client, error)
• type Code
  • func FindCodeByName(name string) Code
• type ConnURL
  • func ConvertURL(from *url.URL) (u *ConnURL, err error)
  • func ParseURL(urlStr string) (u *ConnURL, err error)
  • func (u *ConnURL) Dialer() *Dialer
  • func (u *ConnURL) URL() *url.URL
• type Dialer
  • func NewDialer(secret string) *Dialer
  • func (d *Dialer) Dial(u *ConnURL) (Client, error)
  • func (d *Dialer) DialContext(ctx context.Context, u *ConnURL) (Client, error)
  • func (d *Dialer) DialUpgrade(ctx context.Context, u *url.URL, proto string) (conn net.Conn, err error)
  • func (d *Dialer) RoundTrip(req *http.Request) (conn net.Conn, resp *http.Response, err error)
• type PoolMux
  • func NewPoolMux(code Code, connect func(c Code) (Client, error)) (*PoolMux, error)
  • func NewPoolMuxConfig(cfg PoolMuxConfig) (*PoolMux, error)
  • func (p *PoolMux) Acquire() (cli Client, unique bool, err error)
  • func (p *PoolMux) Busy() int
  • func (p *PoolMux) Call(method string, args, reply any) error
  • func (p *PoolMux) Close() error
  • func (p *PoolMux) Code() Code
  • func (p *PoolMux) MaxConns() int
  • func (p *PoolMux) Preconnect(n int) error
  • func (p *PoolMux) Release(cli Client, unique bool)
• type PoolMuxConfig
• type Server
• type ServerFunc
  • func (f ServerFunc) ServeRPC(method string, body json.RawMessage) (any, error)
• type ServerProtocol
• type Stream
• type UpgradeListener
  • func NewUpgradeListener(name string) *UpgradeListener
  • func (u *UpgradeListener) Accept() (net.Conn, error)
  • func (u *UpgradeListener) Addr() net.Addr
  • func (u *UpgradeListener) Close() error
  • func (u *UpgradeListener) ServeHTTP(w http.ResponseWriter, r *http.Request)
• type UpgradeServer
  • func MakeRPCServer[Arg any](sv func(conn net.Conn, code Code, arg Arg)) UpgradeServer[*rpcServerAdapter[Arg], Arg]
  • func MakeUpgradeServer[Proto ServerProtocol[Arg], Arg any](protos ...Proto) UpgradeServer[Proto, Arg]
  • func (u *UpgradeServer[Proto, Arg]) Upgrade(w http.ResponseWriter, r *http.Request, arg Arg)
• type YamuxCode
  • func (c YamuxCode) Open(conn net.Conn) (Client, error)
  • func (c YamuxCode) Serve(conn net.Conn, sv Server)
  • func (c YamuxCode) String() string

Constants

const (
	FlagReply jrpcSeq = 1 << 31
	FlagMask  jrpcSeq = (1 << 31) - 1
)

const DefaultURL = "pmtp://127.0.0.1"

Variables

var RpcCodeList = []Code{
	YamuxCode{CodeJRPC}, CodeJRPC,
	YamuxCode{CodeJSON}, CodeJSON,
}

var RpcUpgradeServer = MakeUpgradeServer(RpcCodeList...)

Types

type Client

type Client interface {
	Busy() int // number of pending requests, -1 if closed
	Call(method string, args any, reply any) error
	Close() error
	Code() Code
}

func Connect

func Connect(uri string) (Client, error)

Connect to a server using the shared client pool.

func Dial

func Dial(url string) (Client, error)

func DialContext

func DialContext(ctx context.Context, url string) (Client, error)

type Code

type Code interface {
	Open(conn net.Conn) (Client, error)
	Serve(conn net.Conn, sv Server)
	String() string
}

var CodeJRPC Code = jrpcCode{}

var CodeJSON Code = &stdCodeAdapter{jsonCode{}, "json-rpc"}

var CodeYRPC Code = YamuxCode{CodeJRPC}

func FindCodeByName

func FindCodeByName(name string) Code

type ConnURL

type ConnURL struct {
	Path           string
	RawQuery       string
	Host           string
	Secret         string
	DisablePoolMux bool
	DisableYamux   bool
	Code           Code
	Websocket      bool
	TLS            bool
}

URL parsing.

func ConvertURL

func ConvertURL(from *url.URL) (u *ConnURL, err error)

func ParseURL

func ParseURL(urlStr string) (u *ConnURL, err error)

func (*ConnURL) Dialer

func (u *ConnURL) Dialer() *Dialer

func (*ConnURL) URL

func (u *ConnURL) URL() *url.URL

type Dialer

type Dialer struct {
	NetDialContext  func(ctx context.Context, network string, address string) (net.Conn, error)
	TLSClientConfig *tls.Config
}

Dialer is a dialer for RPC connections over HTTP and WebSocket.

func NewDialer

func NewDialer(secret string) *Dialer

func (*Dialer) Dial

func (d *Dialer) Dial(u *ConnURL) (Client, error)

func (*Dialer) DialContext

func (d *Dialer) DialContext(ctx context.Context, u *ConnURL) (Client, error)

func (*Dialer) DialUpgrade

func (d *Dialer) DialUpgrade(ctx context.Context, u *url.URL, proto string) (conn net.Conn, err error)

func (*Dialer) RoundTrip

func (d *Dialer) RoundTrip(req *http.Request) (conn net.Conn, resp *http.Response, err error)

type PoolMux

type PoolMux struct {
	Connect func(c Code) (Client, error)
	// contains filtered or unexported fields
}

PoolMux is a pool of pmtp.Clients, if a code is of type ymux, the pool will be initially populated with the underlying code and switch to a shared ymux connection when the pool is drained.

func NewPoolMux

func NewPoolMux(code Code, connect func(c Code) (Client, error)) (*PoolMux, error)

NewPoolMux creates a new pool of pmtp.Clients with a default size.

func NewPoolMuxConfig

func NewPoolMuxConfig(cfg PoolMuxConfig) (*PoolMux, error)

NewPoolMuxConfig creates a new pool of pmtp.Clients with a given configuration.

func (*PoolMux) Acquire

func (p *PoolMux) Acquire() (cli Client, unique bool, err error)

Acquire returns a pmtp.Client from the pool, if the pool is closed, it returns io.ErrClosedPipe. If the pool is drained, it will share another connection or return the muxed connection.

func (*PoolMux) Busy

func (p *PoolMux) Busy() int

Implement the pmtp.Client interface. Busy returns the number of busy connections in the pool.

func (*PoolMux) Call

func (p *PoolMux) Call(method string, args, reply any) error

Call is a wrapper around pmtp.Client.Call, it acquires a client from the pool,

func (*PoolMux) Close

func (p *PoolMux) Close() error

Close closes the pool and all of its connections.

func (*PoolMux) Code

func (p *PoolMux) Code() Code

Code returns the code of the pool.

func (*PoolMux) MaxConns

func (p *PoolMux) MaxConns() int

Returns the MaxConns of the pool.

func (*PoolMux) Preconnect

func (p *PoolMux) Preconnect(n int) error

Preconnect ensures that the pool is prepopulated with a given number of connections.

func (*PoolMux) Release

func (p *PoolMux) Release(cli Client, unique bool)

Release returns a pmtp.Client to the pool, if the client is unique and not closed, it will be added to the pool.

type PoolMuxConfig

type PoolMuxConfig struct {
	// Code is the code of the pool.
	Code Code
	// Connect is the function used to create a new connection.
	Connect func(c Code) (Client, error)
	// MaxConns is the maximum number of connections in the pool.
	MaxConns int
	// Preconnect is the number of connections to preconnect.
	Preconnect int
	// AfterClose is a function called after the pool is closed.
	AfterClose func()
}

type Server

type Server interface {
	ServeRPC(method string, body json.RawMessage) (any, error)
}

type ServerFunc

type ServerFunc func(method string, body json.RawMessage) (any, error)

func (ServerFunc) ServeRPC

func (f ServerFunc) ServeRPC(method string, body json.RawMessage) (any, error)

type ServerProtocol

type ServerProtocol[Arg any] interface {
	Serve(conn net.Conn, arg Arg)
	String() string
}

type Stream

type Stream interface {
	Close() error
	Write(h uint32, v *jrpcPacket) error
	ReadHeader() (h uint32, err error)
	ReadBody(v *jrpcPacket) error
}

type UpgradeListener

type UpgradeListener struct {
	UpgradeServer[erasedProtocol, any]
	// contains filtered or unexported fields
}

func NewUpgradeListener

func NewUpgradeListener(name string) *UpgradeListener

func (*UpgradeListener) Accept

func (u *UpgradeListener) Accept() (net.Conn, error)

func (*UpgradeListener) Addr

func (u *UpgradeListener) Addr() net.Addr

func (*UpgradeListener) Close

func (u *UpgradeListener) Close() error

func (*UpgradeListener) ServeHTTP

func (u *UpgradeListener) ServeHTTP(w http.ResponseWriter, r *http.Request)

type UpgradeServer

type UpgradeServer[Proto ServerProtocol[Arg], Arg any] struct {
	Websocket websocket.Upgrader
	Protocols map[string]Proto
}

func MakeRPCServer

func MakeRPCServer[Arg any](sv func(conn net.Conn, code Code, arg Arg)) UpgradeServer[*rpcServerAdapter[Arg], Arg]

func MakeUpgradeServer

func MakeUpgradeServer[Proto ServerProtocol[Arg], Arg any](protos ...Proto) UpgradeServer[Proto, Arg]

func (*UpgradeServer[Proto, Arg]) Upgrade

func (u *UpgradeServer[Proto, Arg]) Upgrade(w http.ResponseWriter, r *http.Request, arg Arg)

type YamuxCode

type YamuxCode struct {
	Code
}

func (YamuxCode) Open

func (c YamuxCode) Open(conn net.Conn) (Client, error)

func (YamuxCode) Serve

func (c YamuxCode) Serve(conn net.Conn, sv Server)

func (YamuxCode) String

func (c YamuxCode) String() string