connect

Documentation

Index

• type INatsMessageListener
• type NatsConnection
  • func NewNatsConnection() *NatsConnection
  • func (c *NatsConnection) Close(ctx context.Context, correlationId string) error
  • func (c *NatsConnection) Configure(ctx context.Context, config *cconf.ConfigParams)
  • func (c *NatsConnection) CreateQueue() error
  • func (c *NatsConnection) DeleteQueue() error
  • func (c *NatsConnection) GetConnection() *nats.Conn
  • func (c *NatsConnection) IsOpen() bool
  • func (c *NatsConnection) Open(ctx context.Context, correlationId string) error
  • func (c *NatsConnection) Publish(ctx context.Context, subject string, message *nats.Msg) error
  • func (c *NatsConnection) ReadQueueNames() ([]string, error)
  • func (c *NatsConnection) SetReferences(ctx context.Context, references cref.IReferences)
  • func (c *NatsConnection) Subscribe(ctx context.Context, subject string, queue string, ...) error
  • func (c *NatsConnection) Unsubscribe(ctx context.Context, subject string, queue string, ...) error
• type NatsConnectionResolver
  • func NewNatsConnectionResolver() *NatsConnectionResolver
  • func (c *NatsConnectionResolver) Compose(correlationId string, connections []*ccon.ConnectionParams, ...) (*cconf.ConfigParams, error)
  • func (c *NatsConnectionResolver) Configure(ctx context.Context, config *cconf.ConfigParams)
  • func (c *NatsConnectionResolver) Resolve(correlationId string) (*cconf.ConfigParams, error)
  • func (c *NatsConnectionResolver) SetReferences(ctx context.Context, references cref.IReferences)
• type NatsSubscription

Types

type INatsMessageListener

type INatsMessageListener interface {
	OnMessage(message *nats.Msg)
}

type NatsConnection

type NatsConnection struct {

	// The logger.
	Logger *clog.CompositeLogger
	// The connection resolver.
	ConnectionResolver *NatsConnectionResolver
	// The configuration options.
	Options *cconf.ConfigParams

	// The NATS connection object.
	Connection *nats.Conn
	// contains filtered or unexported fields
}

NATS connection using plain driver. By defining a connection and sharing it through multiple message queues you can reduce number of used connections.

Configuration parameters
	- connection(s):
		- discovery_key:             (optional) a key to retrieve the connection from IDiscovery
		- host:                      host name or IP address
		- port:                      port number (default: 27017)
		- uri:                       resource URI or connection string with all parameters in it
	- credential(s):
		- store_key:                 (optional) a key to retrieve the credentials from ICredentialStore
		- username:                  user name
		- password:                  user password
	- options:
		- retry_connect:        (optional) turns on/off automated reconnect when connection is log (default: true)
		- max_reconnect:        (optional) maximum reconnection attempts (default: 3)
		- reconnect_timeout:    (optional) number of milliseconds to wait on each reconnection attempt (default: 3000)
		- flush_timeout:        (optional) number of milliseconds to wait on flushing messages (default: 3000)

References
	- *:logger:*:*:1.0           (optional) ILogger components to pass log messages
	- *:discovery:*:*:1.0        (optional) IDiscovery services
	- *:credential-store:*:*:1.0 (optional) Credential stores to resolve credentials

func NewNatsConnection

func NewNatsConnection() *NatsConnection

NewNatsConnection creates a new instance of the connection component.

func (*NatsConnection) Close

func (c *NatsConnection) Close(ctx context.Context, correlationId string) error

Closes component and frees used resources.

• correlationId (optional) transaction id to trace execution through call chain.

Return error or nil no errors occured

func (*NatsConnection) Configure

func (c *NatsConnection) Configure(ctx context.Context, config *cconf.ConfigParams)

Configures component by passing configuration parameters.

• config configuration parameters to be set.

func (*NatsConnection) CreateQueue

func (c *NatsConnection) CreateQueue() error

func (*NatsConnection) DeleteQueue

func (c *NatsConnection) DeleteQueue() error

func (*NatsConnection) GetConnection

func (c *NatsConnection) GetConnection() *nats.Conn

func (*NatsConnection) IsOpen

func (c *NatsConnection) IsOpen() bool

Checks if the component is opened. Returns true if the component has been opened and false otherwise.

func (*NatsConnection) Open

func (c *NatsConnection) Open(ctx context.Context, correlationId string) error

Opens the component.

• correlationId (optional) transaction id to trace execution through call chain.
• Return error or nil no errors occured.

func (*NatsConnection) Publish

func (c *NatsConnection) Publish(ctx context.Context, subject string, message *nats.Msg) error

Publish a message to a specified topic

Parameters:

• subject a subject (topic) name
• message a message to be published

Returns: error or nil for success

func (*NatsConnection) ReadQueueNames

func (c *NatsConnection) ReadQueueNames() ([]string, error)

func (*NatsConnection) SetReferences

func (c *NatsConnection) SetReferences(ctx context.Context, references cref.IReferences)

Sets references to dependent components.

• references references to locate the component dependencies.

func (*NatsConnection) Subscribe

func (c *NatsConnection) Subscribe(ctx context.Context, subject string, queue string, listener INatsMessageListener) error

Subscribe to a topic

Parameters:

• subject a subject (topic) name
• queue (optional) a queue group
• listener a message listener

Returns: err or nil for success

func (*NatsConnection) Unsubscribe

func (c *NatsConnection) Unsubscribe(ctx context.Context, subject string, queue string, listener INatsMessageListener) error

Unsubscribe from a previously subscribed topic topic

Parameters:

• subject a subject (topic) name
• queue (optional) a queue group
• listener a message listener

Returns: err or nil for success

type NatsConnectionResolver

type NatsConnectionResolver struct {
	// The connections resolver.
	ConnectionResolver *ccon.ConnectionResolver
	//The credentials resolver.
	CredentialResolver *cauth.CredentialResolver
}

func NewNatsConnectionResolver

func NewNatsConnectionResolver() *NatsConnectionResolver

func (*NatsConnectionResolver) Compose

func (c *NatsConnectionResolver) Compose(correlationId string, connections []*ccon.ConnectionParams, credential *cauth.CredentialParams) (*cconf.ConfigParams, error)

Compose method are composes Nats connection options from connection and credential parameters. Parameters:

• correlationId string (optional) transaction id to trace execution through call chain.
• connection *ccon.ConnectionParams connection parameters
• credential *cauth.CredentialParams credential parameters

Returns: options *cconf.ConfigParams, err error resolved options or error.

func (*NatsConnectionResolver) Configure

func (c *NatsConnectionResolver) Configure(ctx context.Context, config *cconf.ConfigParams)

Configure are configures component by passing configuration parameters. Parameters:

• config *cconf.ConfigParams

configuration parameters to be set.

func (*NatsConnectionResolver) Resolve

func (c *NatsConnectionResolver) Resolve(correlationId string) (*cconf.ConfigParams, error)

Resolves Nats connection options from connection and credential parameters. Parameters:

• correlationId string (optional) transaction id to trace execution through call chain.

Returns options *cconf.ConfigParams, err error receives resolved options or error.

func (*NatsConnectionResolver) SetReferences

func (c *NatsConnectionResolver) SetReferences(ctx context.Context, references cref.IReferences)

SetReferences are sets references to dependent components. Parameters:

• references cref.IReferences references to locate the component dependencies.

type NatsSubscription

type NatsSubscription struct {
	Subject    string
	QueueGroup string
	Listener   INatsMessageListener
	Handler    *nats.Subscription
}