consumers.go

package server

import "time"

type Service struct{}

type ConsumerType int

const (
	SingleConsumer ConsumerType = iota
	PusherConsumer
	PollerConsumer
)

type ConsumerOptions struct {
	Type    ConsumerType
	Period  time.Duration
	Service *Service
}

func NewConsumer(o *ConsumerOptions) Consumer {
	switch o.Type {
	case SingleConsumer:
		return NewSingle(o.Service)
	case PusherConsumer:
		return NewPusher(o.Service)
	case PollerConsumer:
		return NewPoller(o.Service, o.Period)
	default:
		panic("invalid ConsumerType")
	}
}

type Consumer interface {
	Chan() chan string
	Close()
}

type consumer struct {
	out  chan string
	done chan struct{}
}

func (c *consumer) Chan() {
	return c.out
}

func (c *consumer) Close() {
	close(c.out)
}

func (c *consumer) send(s string) bool {
	select {
	case p.out <- s:
		return true
	case <-p.done:
		return false
	}
}

func (c *consumer) sleep(d time.Duration) bool {
	select {
	case <-time.After(d):
		return true
	case <-c.done:
		return false
	}
}

type Pusher struct {
	consumer
	srv *Service
}

func NewPusher(srv *Service) *Pusher {
	p := &Pusher{srv: srv}
	go p.run()
	return p
}

func (p *Poller) run() {
	p.key = p.srv.Listen(p.Send)
	<-p.done
	p.srv.UnListen(p.key)
}

type Single struct {
	consumer
	srv *Service
}

func NewSingle(srv *Service) *Single {
	s := &Single{srv: srv}
	go s.run()
	return s
}

func (s *Single) run() {
	s.send(s.srv.Image())
}

type Poller struct {
	consumer
	srv    *Service
	period time.Duration
}

func NewPoller(srv *Service, period time.Duration) *Poller {
	p := &Poller{srv: srv, period: period}
	go p.run()
	return p
}

func (p *Poller) run() {
	for {
		if !p.send(p.srv.Image()) {
			break
		}
		if !p.sleep(p.period) {
			break
		}
	}
}