rpcServer

Documentation

Overview

Package "rpcServer" provides primitives to interact with the AsyncAPI specification.

Code generated by github.com/lerenn/asyncapi-codegen version (devel) DO NOT EDIT.

Index

• Constants
• Variables
• type AppController
  • func NewAppController(bs BrokerController) (*AppController, error)
  • func (c *AppController) Close()
  • func (c AppController) Errors() <-chan Error
  • func (c *AppController) PublishQueue(params QueueParameters, msg QueueMessage) error
  • func (c *AppController) SubscribeAll(as AppSubscriber) error
  • func (c *AppController) SubscribeRpcQueue(fn func(msg RpcQueueMessage)) error
  • func (c *AppController) UnsubscribeAll()
  • func (c *AppController) UnsubscribeRpcQueue()
• type AppSubscriber
• type BrokerController
• type ClientController
  • func NewClientController(bs BrokerController) (*ClientController, error)
  • func (c *ClientController) Close()
  • func (c ClientController) Errors() <-chan Error
  • func (c *ClientController) PublishRpcQueue(msg RpcQueueMessage) error
  • func (c *ClientController) SubscribeAll(as ClientSubscriber) error
  • func (c *ClientController) SubscribeQueue(params QueueParameters, fn func(msg QueueMessage)) error
  • func (c *ClientController) UnsubscribeAll()
  • func (c *ClientController) UnsubscribeQueue(params QueueParameters)
  • func (cc *ClientController) WaitForQueue(ctx context.Context, params QueueParameters, msg MessageWithCorrelationID, ...) (QueueMessage, error)
• type ClientSubscriber
• type Error
  • func (e *Error) Error() string
• type MessageWithCorrelationID
• type QueueMessage
  • func NewQueueMessage() QueueMessage
  • func (msg QueueMessage) CorrelationID() string
  • func (msg *QueueMessage) SetAsResponseFrom(req MessageWithCorrelationID)
• type QueueParameters
• type RpcQueueMessage
  • func NewRpcQueueMessage() RpcQueueMessage
  • func (msg RpcQueueMessage) CorrelationID() string
  • func (msg *RpcQueueMessage) SetAsResponseFrom(req MessageWithCorrelationID)
• type UniversalMessage

Constants

const (
	// CorrelationIDField is the name of the field that will contain the correlation ID
	CorrelationIDField = "correlation_id"
)

Variables

var (
	// Generic error for AsyncAPI generated code
	ErrAsyncAPI = errors.New("error when using AsyncAPI")

	// ErrContextCancelled is given when a given context is cancelled
	ErrContextCancelled = fmt.Errorf("%w: context cancelled", ErrAsyncAPI)

	// ErrNilBrokerController is raised when a nil broker controller is user
	ErrNilBrokerController = fmt.Errorf("%w: nil broker controller has been used", ErrAsyncAPI)

	// ErrNilAppSubscriber is raised when a nil app subscriber is user
	ErrNilAppSubscriber = fmt.Errorf("%w: nil app subscriber has been used", ErrAsyncAPI)

	// ErrNilClientSubscriber is raised when a nil client subscriber is user
	ErrNilClientSubscriber = fmt.Errorf("%w: nil client subscriber has been used", ErrAsyncAPI)

	// ErrAlreadySubscribedChannel is raised when a subscription is done twice
	// or more without unsubscribing
	ErrAlreadySubscribedChannel = fmt.Errorf("%w: the channel has already been subscribed", ErrAsyncAPI)
)

Types

type AppController

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

AppController is the structure that provides publishing capabilities to the developer and and connect the broker with the App

func NewAppController

func NewAppController(bs BrokerController) (*AppController, error)

NewAppController links the App to the broker

func (*AppController) Close

func (c *AppController) Close()

Close will clean up any existing resources on the controller

func (AppController) Errors

func (c AppController) Errors() <-chan Error

Errors will give back the channel that contains errors and that you can listen to handle errors Please take a look at Error struct form information on error

func (*AppController) PublishQueue

func (c *AppController) PublishQueue(params QueueParameters, msg QueueMessage) error

PublishQueue will publish messages to '{queue}' channel

func (*AppController) SubscribeAll

func (c *AppController) SubscribeAll(as AppSubscriber) error

SubscribeAll will subscribe to channels without parameters on which the app is expecting messages. For channels with parameters, they should be subscribed independently.

func (*AppController) SubscribeRpcQueue

func (c *AppController) SubscribeRpcQueue(fn func(msg RpcQueueMessage)) error

SubscribeRpcQueue will subscribe to new messages from 'rpc_queue' channel

func (*AppController) UnsubscribeAll

func (c *AppController) UnsubscribeAll()

UnsubscribeAll will unsubscribe all remaining subscribed channels

func (*AppController) UnsubscribeRpcQueue

func (c *AppController) UnsubscribeRpcQueue()

UnsubscribeRpcQueue will unsubscribe messages from 'rpc_queue' channel

type AppSubscriber

type AppSubscriber interface {
	// RpcQueue
	RpcQueue(msg RpcQueueMessage)
}

AppSubscriber represents all handlers that are expecting messages for App

type BrokerController

type BrokerController interface {
	// Publish a message to the broker
	Publish(channel string, mw UniversalMessage) error

	// Subscribe to messages from the broker
	Subscribe(channel string) (msgs chan UniversalMessage, stop chan interface{}, err error)
}

BrokerController represents the functions that should be implemented to connect the broker to the application or the client

type ClientController

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

ClientController is the structure that provides publishing capabilities to the developer and and connect the broker with the Client

func NewClientController

func NewClientController(bs BrokerController) (*ClientController, error)

NewClientController links the Client to the broker

func (*ClientController) Close

func (c *ClientController) Close()

Close will clean up any existing resources on the controller

func (ClientController) Errors

func (c ClientController) Errors() <-chan Error

Errors will give back the channel that contains errors and that you can listen to handle errors Please take a look at Error struct form information on error

func (*ClientController) PublishRpcQueue

func (c *ClientController) PublishRpcQueue(msg RpcQueueMessage) error

PublishRpcQueue will publish messages to 'rpc_queue' channel

func (*ClientController) SubscribeAll

func (c *ClientController) SubscribeAll(as ClientSubscriber) error

SubscribeAll will subscribe to channels without parameters on which the app is expecting messages. For channels with parameters, they should be subscribed independently.

func (*ClientController) SubscribeQueue

func (c *ClientController) SubscribeQueue(params QueueParameters, fn func(msg QueueMessage)) error

SubscribeQueue will subscribe to new messages from '{queue}' channel

func (*ClientController) UnsubscribeAll

func (c *ClientController) UnsubscribeAll()

UnsubscribeAll will unsubscribe all remaining subscribed channels

func (*ClientController) UnsubscribeQueue

func (c *ClientController) UnsubscribeQueue(params QueueParameters)

UnsubscribeQueue will unsubscribe messages from '{queue}' channel

func (*ClientController) WaitForQueue

func (cc *ClientController) WaitForQueue(ctx context.Context, params QueueParameters, msg MessageWithCorrelationID, pub func() error) (QueueMessage, error)

WaitForQueue will wait for a specific message by its correlation ID

The pub function is the publication function that should be used to send the message It will be called after subscribing to the channel to avoid race condition, and potentially loose the message

type ClientSubscriber

type ClientSubscriber interface {
	// Queue
	Queue(msg QueueMessage)
}

ClientSubscriber represents all handlers that are expecting messages for Client

type Error

type Error struct {
	Channel string
	Err     error
}

func (*Error) Error

func (e *Error) Error() string

type MessageWithCorrelationID

type MessageWithCorrelationID interface {
	CorrelationID() string
}

type QueueMessage

type QueueMessage struct {
	// Headers will be used to fill the message headers
	Headers struct {
		CorrelationID *string `json:"correlation_id"`
	}

	// Payload will be inserted in the message payload
	Payload struct {
		Result *float64 `json:"result"`
	}
}

QueueMessage is the message expected for 'Queue' channel

func NewQueueMessage

func NewQueueMessage() QueueMessage

func (QueueMessage) CorrelationID

func (msg QueueMessage) CorrelationID() string

CorrelationID will give the correlation ID of the message, based on AsyncAPI spec

func (*QueueMessage) SetAsResponseFrom

func (msg *QueueMessage) SetAsResponseFrom(req MessageWithCorrelationID)

SetAsResponseFrom will correlate the message with the one passed in parameter. It will assign the 'req' message correlation ID to the message correlation ID, both specified in AsyncAPI spec.

type QueueParameters

type QueueParameters struct {
	Queue string
}

QueueParameters represents Queue channel parameters

type RpcQueueMessage

type RpcQueueMessage struct {
	// Headers will be used to fill the message headers
	Headers struct {
		CorrelationID *string `json:"correlation_id"`
	}

	// Payload will be inserted in the message payload
	Payload struct {
		Numbers []float64 `json:"numbers"`
	}
}

RpcQueueMessage is the message expected for 'RpcQueue' channel

func NewRpcQueueMessage

func NewRpcQueueMessage() RpcQueueMessage

func (RpcQueueMessage) CorrelationID

func (msg RpcQueueMessage) CorrelationID() string

CorrelationID will give the correlation ID of the message, based on AsyncAPI spec

func (*RpcQueueMessage) SetAsResponseFrom

func (msg *RpcQueueMessage) SetAsResponseFrom(req MessageWithCorrelationID)

SetAsResponseFrom will correlate the message with the one passed in parameter. It will assign the 'req' message correlation ID to the message correlation ID, both specified in AsyncAPI spec.

type UniversalMessage

type UniversalMessage struct {
	CorrelationID *string
	Payload       []byte
}

UniversalMessage is a wrapper that will contain all information regarding a message