replaycli-go/api/nats.go

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
}

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
}
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

}
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)
}