uplink/app/pubsub/pubsubwriter.go

package pubsub

import (
	"io"

	cskrpubsub "github.com/cskr/pubsub"
)

type PubSubWriter struct {
	PubSub      *cskrpubsub.PubSub
	topic       string
	fullControl bool
	closed      bool
}

func NewPubSubWriter() *PubSubWriter {
	pipe := new(PubSubWriter)
	pipe.PubSub = cskrpubsub.New(1)
	pipe.fullControl = true
	return pipe
}

func NewPubSubWriterForTopic(pubsub *cskrpubsub.PubSub, topic string) *PubSubWriter {
	pipe := new(PubSubWriter)
	pipe.PubSub = pubsub
	pipe.topic = topic
	return pipe
}

func (pipe *PubSubWriter) Write(p []byte) (n int, err error) {
	if pipe.closed {
		err = io.EOF
		return
	}
	pipe.PubSub.Pub(p, pipe.topic)
	n = len(p)
	return
}

func (pipe *PubSubWriter) Close() (err error) {
	if pipe.closed {
		err = io.EOF
		return
	}
	pipe.PubSub.Close(pipe.topic)
	if pipe.fullControl {
		pipe.PubSub.Shutdown()
	}
	pipe.closed = true
	return
}

func (pipe *PubSubWriter) Sub() io.ReadCloser {
	return &PubSubReader{
		channel: pipe.PubSub.Sub(pipe.topic),
		pubsub:  pipe.PubSub,
		closed:  pipe.closed,
	}
}