package sink

import (
	"errors"
	"github.com/nats-io/nats.go"
)

type NatsSink struct {
	conn *nats.Conn
	in   chan interface{}
}

func NewNatsSink(addr string, topic string) (*NatsSink, error) {
	// Connect to a server
	nc, err := nats.Connect(addr)
	if err != nil {
		return nil, err
	}
	sink := &NatsSink{
		conn: nc,
		in:   make(chan interface{}),
	}
	go sink.init(topic)
	return sink, nil
}

func (ns *NatsSink) init(topic string) error {
	if len(topic) == 1{
		return errors.New("订阅主题不能为空")
	}
	for msg := range ns.in {
		switch m := msg.(type) {
		case nats.Msg:
			ns.conn.Publish(topic, m.Data)
		case string:
			ns.conn.Publish(topic, []byte(m))
		}
	}
	return nil
}

// In returns channel for receiving data
func (ns *NatsSink) In() chan<- interface{} {
	return ns.in
}