100 lines
2.4 KiB
Go
100 lines
2.4 KiB
Go
|
package nats
|
||
|
|
|
||
|
|
import (
|
||
|
|
"github.com/nats-io/nats"
|
||
|
|
)
|
||
|
|
|
||
|
|
// NatsAdaptor is a configuration struct for interacting with a nats server.
|
||
|
|
// Name is a logical name for the adaptor/nats server connection.
|
||
|
|
// Host is in the form "localhost:4222" which is the hostname/ip and port of the nats server.
|
||
|
|
// ClientID is a unique identifier integer that specifies the identity of the client.
|
||
|
|
type NatsAdaptor struct {
|
||
|
|
name string
|
||
|
|
Host string
|
||
|
|
clientID int
|
||
|
|
username string
|
||
|
|
password string
|
||
|
|
client *nats.Conn
|
||
|
|
}
|
||
|
|
|
||
|
|
// NewNatsAdaptor populates a new NatsAdaptor.
|
||
|
|
func NewNatsAdaptor(name string, host string, clientID int) *NatsAdaptor {
|
||
|
|
return &NatsAdaptor{
|
||
|
|
name: name,
|
||
|
|
Host: host,
|
||
|
|
clientID: clientID,
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
// NewNatsAdaptorWithAuth populates a NatsAdaptor including username and password.
|
||
|
|
func NewNatsAdaptorWithAuth(name string, host string, clientID int, username string, password string) *NatsAdaptor {
|
||
|
|
return &NatsAdaptor{
|
||
|
|
name: name,
|
||
|
|
Host: host,
|
||
|
|
clientID: clientID,
|
||
|
|
username: username,
|
||
|
|
password: password,
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
// Name returns the logical client name.
|
||
|
|
func (a *NatsAdaptor) Name() string { return a.name }
|
||
|
|
|
||
|
|
// Connect makes a connection to the Nats server.
|
||
|
|
func (a *NatsAdaptor) Connect() (errs []error) {
|
||
|
|
|
||
|
|
auth := ""
|
||
|
|
if a.username != "" && a.password != "" {
|
||
|
|
auth = a.username + ":" + a.password + "@"
|
||
|
|
}
|
||
|
|
|
||
|
|
defaultURL := "nats://" + auth + a.Host
|
||
|
|
|
||
|
|
var err error
|
||
|
|
a.client, err = nats.Connect(defaultURL)
|
||
|
|
if err != nil {
|
||
|
|
return append(errs, err)
|
||
|
|
}
|
||
|
|
return
|
||
|
|
}
|
||
|
|
|
||
|
|
// Disconnect from the nats server. Returns an error if the client doesn't exist.
|
||
|
|
func (a *NatsAdaptor) Disconnect() (err error) {
|
||
|
|
if a.client != nil {
|
||
|
|
a.client.Close()
|
||
|
|
}
|
||
|
|
return
|
||
|
|
}
|
||
|
|
|
||
|
|
// Finalize is simply a helper method for the disconnect.
|
||
|
|
func (a *NatsAdaptor) Finalize() (errs []error) {
|
||
|
|
a.Disconnect()
|
||
|
|
return
|
||
|
|
}
|
||
|
|
|
||
|
|
// Publish sends a message with the particular topic to the nats server.
|
||
|
|
func (a *NatsAdaptor) Publish(topic string, message []byte) bool {
|
||
|
|
if a.client == nil {
|
||
|
|
return false
|
||
|
|
}
|
||
|
|
a.client.Publish(topic, message)
|
||
|
|
return true
|
||
|
|
}
|
||
|
|
|
||
|
|
// On is an event-handler style subscriber to a particular topic (named event).
|
||
|
|
// Supply a handler function to use the bytes returned by the server.
|
||
|
|
func (a *NatsAdaptor) On(event string, f func(s []byte)) bool {
|
||
|
|
if a.client == nil {
|
||
|
|
return false
|
||
|
|
}
|
||
|
|
a.client.Subscribe(event, func(msg *nats.Msg) {
|
||
|
|
incoming := msg.Data
|
||
|
|
if string(incoming) == "PING" {
|
||
|
|
a.Publish(event, []byte("PONG"))
|
||
|
|
} else {
|
||
|
|
f(msg.Data)
|
||
|
|
}
|
||
|
|
})
|
||
|
|
return true
|
||
|
|
}
|