rpc

Documentation

Index

• func GRPCClientOptions(tlsCertificateAuthorityPath, tlsCertificatePath, tlsPrivateKey string, ...) []grpc.DialOption
• func GRPCServerOptions(mTLS bool, ...) []grpc.ServerOption
• func GenerateSelfSignedCertificate(cn string, san []string, ipAddresses []net.IP) (*tls.Certificate, error)
• func ListLocalIP() []net.IP
• func Server(config ServerConfig) *grpc.Server
• type ClientConfig
• type Dialer
  • func GRPCDialer(config ClientConfig) Dialer
• type FSM
• type MqttServer
  • func NewMQTTServer(state State, fsm FSM, localPublishCh, remotePublishCh chan *packet.Publish) *MqttServer
  • func (s *MqttServer) CreateSubscription(ctx context.Context, r *api.CreateSubscriptionRequest) (*api.CreateSubscriptionResponse, error)
  • func (s *MqttServer) DeleteSubscription(ctx context.Context, r *api.DeleteSubscriptionRequest) (*api.DeleteSubscriptionResponse, error)
  • func (s *MqttServer) DistributeMessage(ctx context.Context, r *api.DistributeMessageRequest) (*api.DistributeMessageResponse, error)
  • func (s *MqttServer) ListSessionMetadatas(ctx context.Context, r *api.ListSessionMetadatasRequest) (*api.ListSessionMetadatasResponse, error)
  • func (s *MqttServer) ListSubscriptions(ctx context.Context, r *api.ListSubscriptionsRequest) (*api.ListSubscriptionsResponse, error)
  • func (s *MqttServer) Serve(grpcServer *grpc.Server)
• type NodeRPCServer
  • func NewNodeRPCServer(cancelCh chan<- struct{}) *NodeRPCServer
  • func (n *NodeRPCServer) Shutdown(ctx context.Context, _ *api.ShutdownRequest) (*api.ShutdownResponse, error)
• type ServerConfig
• type State

Functions

func GRPCClientOptions

func GRPCClientOptions(tlsCertificateAuthorityPath, tlsCertificatePath, tlsPrivateKey string, insecureSkipVerify bool) []grpc.DialOption

func GRPCServerOptions

func GRPCServerOptions(mTLS bool, tlsCertificateAuthorityPath, tlsCertificatePath, tlsPrivateKey string) []grpc.ServerOption

func GenerateSelfSignedCertificate

func GenerateSelfSignedCertificate(cn string, san []string, ipAddresses []net.IP) (*tls.Certificate, error)

func ListLocalIP

func ListLocalIP() []net.IP

func Server

func Server(config ServerConfig) *grpc.Server

Types

type ClientConfig

type ClientConfig struct {
	TLSCertificateAuthorityPath string
	TLSCertificatePath          string
	TLSPrivateKeyPath           string
	InsecureSkipVerify          bool
}

type Dialer

type Dialer func(address string, opts ...grpc.DialOption) (*grpc.ClientConn, error)

func GRPCDialer

func GRPCDialer(config ClientConfig) Dialer

type FSM

type FSM interface {
	RetainedMessage(ctx context.Context, publish *packet.Publish) error
	DeleteRetainedMessage(ctx context.Context, topic []byte) error
	Subscribe(ctx context.Context, id string, pattern []byte, qos int32) error
	SubscribeFrom(ctx context.Context, id string, peer uint64, pattern []byte, qos int32) error
	Unsubscribe(ctx context.Context, id string, pattern []byte) error
	DeleteSessionMetadata(ctx context.Context, id, mountpoint string) error
	CreateSessionMetadata(ctx context.Context, id, clientID string, lwt *packet.Publish, mountpoint string) error
}

type MqttServer

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

func NewMQTTServer

func NewMQTTServer(state State, fsm FSM, localPublishCh, remotePublishCh chan *packet.Publish) *MqttServer

func (*MqttServer) CreateSubscription

func (s *MqttServer) CreateSubscription(ctx context.Context, r *api.CreateSubscriptionRequest) (*api.CreateSubscriptionResponse, error)

func (*MqttServer) DeleteSubscription

func (s *MqttServer) DeleteSubscription(ctx context.Context, r *api.DeleteSubscriptionRequest) (*api.DeleteSubscriptionResponse, error)

func (*MqttServer) DistributeMessage

func (s *MqttServer) DistributeMessage(ctx context.Context, r *api.DistributeMessageRequest) (*api.DistributeMessageResponse, error)

func (*MqttServer) ListSessionMetadatas

func (s *MqttServer) ListSessionMetadatas(ctx context.Context, r *api.ListSessionMetadatasRequest) (*api.ListSessionMetadatasResponse, error)

func (*MqttServer) ListSubscriptions

func (s *MqttServer) ListSubscriptions(ctx context.Context, r *api.ListSubscriptionsRequest) (*api.ListSubscriptionsResponse, error)

func (*MqttServer) Serve

func (s *MqttServer) Serve(grpcServer *grpc.Server)

type NodeRPCServer

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

func NewNodeRPCServer

func NewNodeRPCServer(cancelCh chan<- struct{}) *NodeRPCServer

func (*NodeRPCServer) Shutdown

func (n *NodeRPCServer) Shutdown(ctx context.Context, _ *api.ShutdownRequest) (*api.ShutdownResponse, error)

type ServerConfig

type ServerConfig struct {
	TLSCertificateAuthorityPath string
	TLSCertificatePath          string
	TLSPrivateKeyPath           string
	VerifyClientCert            bool
}

type State

type State interface {
	ListSessionMetadatas() []*api.SessionMetadatas
	ListSubscriptions() ([][]byte, []uint64, []string, []int32, error)
}