transport

Documentation

Index

• Constants
• Variables
• func RegisterFunction[Req, Resp proto.Message](t *EtcdTransportServer, handler func(context.Context, Req) (Resp, error)) error
• func RegisterWatcher(lc fx.Lifecycle, t *EtcdTransportServer, agentName string)
• func SendRequest[RespValue any, Resp ExactMessage[RespValue]](ctx context.Context, t *EtcdTransportClient, agentName string, ...) (*RespValue, error)
• type EtcdTransportClient
  • func NewEtcdTransportClient(client *etcdclient.Client) (*EtcdTransportClient, error)
  • func (t *EtcdTransportClient) SendRequest(ctx context.Context, agentName string, msg proto.Message) (*Response, error)
• type EtcdTransportServer
  • func NewEtcdTransportServer(client *etcdclient.Client) (*EtcdTransportServer, error)
  • func (c *EtcdTransportServer) RegisterWatcher(agentName string) error
  • func (c *EtcdTransportServer) Stop()
• type ExactMessage
• type HandlerRegistry
  • func NewHandlerRegistry() *HandlerRegistry
• type Request
• type Response
• type Result
  • func SendRequests[RespValue any, Resp ExactMessage[RespValue]](ctx context.Context, t *EtcdTransportClient, agents []string, ...) ([]Result[*RespValue], error)

Constants

const (
	// RPCBasePath is the base path for all etc based RPCs.
	RPCBasePath = "/rpc"

	// RPCRequestPath is the path for RPC requests.
	RPCRequestPath = "/request"

	// RPCResponsePath is the path for RPC responses.
	RPCResponsePath = "/response"
)

Variables

var TransportClientModule = fx.Options(
	fx.Provide(NewEtcdTransportClient),
)

TransportClientModule is the client fx provider for etcd transport.

var TransportServerModule = fx.Options(
	fx.Provide(NewEtcdTransportServer),
)

TransportServerModule is the server fx provider for etcd transport.

Functions

func RegisterFunction

func RegisterFunction[Req, Resp proto.Message](
	t *EtcdTransportServer,
	handler func(context.Context, Req) (Resp, error),
) error

RegisterFunction register a function as a handler in the registry Only one function for a given Req type can be registered.

func RegisterWatcher

func RegisterWatcher(lc fx.Lifecycle, t *EtcdTransportServer, agentName string)

RegisterWatcher allows to register a server on the etcd transport.

func SendRequest

func SendRequest[RespValue any, Resp ExactMessage[RespValue]](ctx context.Context, t *EtcdTransportClient, agentName string, msg proto.Message) (*RespValue, error)

SendRequest allows consumers of the etcd transport to send single request to agents.

Types

type EtcdTransportClient

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

EtcdTransportClient is the client side of the etcd transport.

func NewEtcdTransportClient

func NewEtcdTransportClient(client *etcdclient.Client) (*EtcdTransportClient, error)

NewEtcdTransportClient creates a new client on the etcd transport.

func (*EtcdTransportClient) SendRequest

func (t *EtcdTransportClient) SendRequest(ctx context.Context, agentName string, msg proto.Message) (*Response, error)

SendRequest sends a request to etcd, supposed to be consumed by an agent.

type EtcdTransportServer

type EtcdTransportServer struct {
	Registry *HandlerRegistry
	// contains filtered or unexported fields
}

EtcdTransportServer is the server side for the etcd transport.

func NewEtcdTransportServer

func NewEtcdTransportServer(client *etcdclient.Client) (*EtcdTransportServer, error)

NewEtcdTransportServer creates and returns a new etcd transport server module.

func (*EtcdTransportServer) RegisterWatcher

func (c *EtcdTransportServer) RegisterWatcher(agentName string) error

RegisterWatcher register an agent on the etcd transport.

func (*EtcdTransportServer) Stop

func (c *EtcdTransportServer) Stop()

Stop stops the etcd transport server.

type ExactMessage

type ExactMessage[T any] interface {
	proto.Message
	*T
}

ExactMessage is like proto.Message, but known to point to T.

type HandlerRegistry

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

HandlerRegistry allow registering handlers and can start a dispatcher.

This is intended to be used at at fx provide/invoke stage.

func NewHandlerRegistry

func NewHandlerRegistry() *HandlerRegistry

NewHandlerRegistry creates a new HandlerRegistry.

type Request

type Request struct {
	ID     string
	Client string
	Data   []byte
}

Request is the raw request on the etcd transport.

type Response

type Response struct {
	ID     string
	Client string
	Data   []byte
	Error  error
}

Response is the raw response on the etcd transport.

type Result

type Result[Resp any] struct {
	Client  string
	Success Resp
	Err     error
}

Result is a result from one agent.

func SendRequests

func SendRequests[RespValue any, Resp ExactMessage[RespValue]](ctx context.Context, t *EtcdTransportClient, agents []string, msg proto.Message) ([]Result[*RespValue], error)

SendRequests allows consumers of the etcd transport to send requests to agents.