nats-io/stan.go simple publisher

nats-stan-publisher.go

package main

import (
	"fmt"
	"github.com/nats-io/nats.go"
	_ "github.com/nats-io/nats.go"
	"github.com/nats-io/stan.go"
	"os"
	"time"
)

type QueueHandler func(msg QueueMsg) (status string)
type QueueMsg interface {
	Ack() error
	Data() []byte
}

func main() {
	cfg := &Config{
		Host:      "nats-url.ru",
		ClusterId: "nats-cluster-id",
		Subject:   "test",
		ClientId:  "test-client",
		Login:     "login",
		Password:  "password",
	}
	err := Publish(cfg)
	if err != nil {
		println(err.Error())
	}
}

type Config struct {
	Host      string
	ClusterId string
	Subject   string
	ClientId  string
	Login     string
	Password  string
}

func Publish(cfg *Config) error {
	ncOpts := []nats.Option{
		nats.Name(cfg.ClientId),
		nats.MaxReconnects(-1),
		nats.ReconnectBufSize(-1),
		nats.UserInfo(cfg.Login, cfg.Password),
	}

	nc, err := nats.Connect(cfg.Host, ncOpts...)
	if err != nil {
		println("nats connect err, url", cfg.Host, "cluster", cfg.ClusterId, err.Error())
		os.Exit(1)
	}

	conn, err := stan.Connect(cfg.ClusterId, cfg.ClientId, stan.NatsConn(nc))
	if err != nil {
		return err
	}

	for i := 0; i < 1; i++ {
		err = conn.Publish(cfg.Subject, []byte("[{\"prop1\":12345,\"time\":null,\"country\":\"RU\",\"is_virtual\":false}]"))
		if err != nil {
			println(err.Error())
		}
		fmt.Println(i)
		time.Sleep(time.Second)
	}

	return nil
}