replaycli-go/api/nats.go

146 lines
2.9 KiB
Go
Raw Permalink Normal View History

2021-11-11 14:07:20 +00:00
package api
import (
"encoding/json"
"errors"
"fmt"
"github.com/nats-io/nats.go"
"log"
"os"
"time"
)
var nc *nats.Conn
type CallOpts struct {
To int
}
2021-11-14 13:31:07 +00:00
func NC() (nc *nats.Conn, err error) {
if nc == nil {
apikey = os.Getenv("REPLAY_APIKEY")
if apikey == "" {
log.Printf("Connecting to NATS Server w/o TK.")
nc, err = nats.Connect(nats.DefaultURL)
} else {
log.Printf("Connecting to NATS Server w TK: " + apikey)
nc, err = nats.Connect(nats.DefaultURL, nats.Token(apikey))
}
if err != nil {
return nil, err
}
}
return nc, nil
}
2021-11-11 14:07:20 +00:00
func Call(s string, in interface{}, out interface{}, opts ...*CallOpts) error {
var err error
opt := &CallOpts{}
if opts != nil && len(opts) > 1 {
opt = opts[0]
}
if opt.To == 0 {
opt.To = 60
}
defer func() {
if nc != nil && err != nil {
err1 := nc.Publish("err", []byte(fmt.Sprintf("Error calling nats: %s: \nIN: %v\nOUT: %v", s, err.Error(), in, out)))
if err1 != nil {
log.Printf("Error calling nats: %s: \nIN: %v\nOUT: %v", s, err.Error(), in, out)
log.Printf("Inner error: %s", err1.Error())
}
} else if nc == nil && err != nil {
log.Printf("Error calling nats: %s: \nIN: %v\nOUT: %v\n(nc is nil)", s, err.Error(), in, out)
}
}()
if nc == nil {
apikey = os.Getenv("REPLAY_APIKEY")
if apikey == "" {
log.Printf("Connecting to NATS Server w/o TK.")
nc, err = nats.Connect(nats.DefaultURL)
} else {
log.Printf("Connecting to NATS Server w TK: " + apikey)
nc, err = nats.Connect(nats.DefaultURL, nats.Token(apikey))
}
if err != nil {
return err
}
}
var bs []byte
if in != nil {
bs, err = json.Marshal(in)
if err != nil {
return err
}
} else {
bs = []byte("{}")
}
msg, err := nc.Request(s, bs, time.Second+time.Duration(opt.To))
if err != nil {
return err
}
if out != nil {
err = json.Unmarshal(msg.Data, out)
if err != nil {
out = bs
return err
}
}
errstr := msg.Header.Get("ERR")
if errstr != "" {
return errors.New(errstr)
}
return nil
}
func Close() error {
return nc.Drain()
}
func Pub(s string, in interface{}) error {
var err error
defer func() {
if nc != nil && err != nil {
err1 := nc.Publish("err", []byte(fmt.Sprintf("Error calling nats: %s: \nIN: %v", s, err.Error(), in)))
if err1 != nil {
log.Printf("Error calling nats: %s: \nIN: %v", s, err.Error(), in)
log.Printf("Inner error: %s", err1.Error())
}
} else if nc == nil && err != nil {
log.Printf("Error calling nats: %s: \nIN: %v\n(nc is nil)", s, err.Error(), in)
}
}()
bs, err := json.Marshal(in)
if err != nil {
return err
}
err = nc.Publish(s, bs)
return err
}
2021-11-14 13:31:07 +00:00
func Init() error {
var err error
nc, err = NC()
if err != nil {
return err
}
nc.Subscribe("exit", func(msg *nats.Msg) {
log.Printf("Exit Required")
os.Exit(0)
})
return nil
}
func Err(e error) {
Pub("err", e.Error())
}
func Dbg(e string) {
Pub("dbg", e)
}