natsio/natsio.go

77 lines
1.4 KiB
Go
Raw Permalink Normal View History

2021-10-08 04:23:59 +00:00
package natsio
import (
"errors"
"fmt"
"github.com/nats-io/nats.go"
)
2021-12-02 03:29:52 +00:00
// Server ...
type Server struct {
instance *nats.Conn
2022-03-30 12:17:14 +00:00
Config Config
2021-12-02 03:29:52 +00:00
}
// JetStream ...
type JetStream struct {
instance nats.JetStreamContext
}
2021-10-08 04:23:59 +00:00
var (
2021-12-02 03:29:52 +00:00
natsServer Server
natsJetStream JetStream
2021-10-08 04:23:59 +00:00
)
// Connect ...
func Connect(cfg Config) error {
if cfg.URL == "" {
return errors.New("connect URL is required")
}
// Connect options
opts := make([]nats.Option, 0)
// Has authentication
if cfg.User != "" {
opts = append(opts, nats.UserInfo(cfg.User, cfg.Password))
}
// If it has TLS
if cfg.TLS != nil {
opts = append(opts, nats.ClientCert(cfg.TLS.CertFilePath, cfg.TLS.KeyFilePath))
opts = append(opts, nats.RootCAs(cfg.TLS.RootCAFilePath))
}
nc, err := nats.Connect(cfg.URL, opts...)
if err != nil {
msg := fmt.Sprintf("error when connecting to NATS: %s", err.Error())
return errors.New(msg)
}
2022-08-30 04:03:02 +00:00
fmt.Printf("⚡️[natsio]: connected to %s \n", cfg.URL)
2021-10-08 04:23:59 +00:00
// Set client
2021-12-02 03:29:52 +00:00
natsServer.instance = nc
2022-03-30 12:17:14 +00:00
natsServer.Config = cfg
2021-10-08 04:23:59 +00:00
// Create jet stream context
2021-12-02 03:29:52 +00:00
js, err := nc.JetStream(nats.PublishAsyncMaxPending(256))
2021-10-08 04:23:59 +00:00
if err != nil {
msg := fmt.Sprintf("error when create NATS JetStream: %s", err.Error())
return errors.New(msg)
}
2021-12-02 03:29:52 +00:00
natsJetStream.instance = js
2021-10-08 04:23:59 +00:00
return nil
}
2021-12-02 03:29:52 +00:00
// GetServer ...
func GetServer() Server {
return natsServer
2021-12-02 03:01:01 +00:00
}
2021-10-08 04:23:59 +00:00
// GetJetStream ...
2021-12-02 03:29:52 +00:00
func GetJetStream() JetStream {
return natsJetStream
2021-10-08 04:23:59 +00:00
}