events.go

package goqless

import (
	"fmt"
	"github.com/garyburd/redigo/redis"
)

type Events struct {
	conn redis.Conn
	ch   chan interface{}
	psc  *redis.PubSubConn
	host string
	port string
}

func NewEvents(host, port string) *Events {
	return &Events{host: host, port: port}
}

func (e *Events) Listen() (chan interface{}, error) {
	var err error
	e.conn, err = redis.Dial("tcp", fmt.Sprintf("%s:%s", e.host, e.port))
	if err != nil {
		return nil, err
	}

	e.psc = &redis.PubSubConn{e.conn}
	e.ch = make(chan interface{}, 1)

	go func() {
		for {
			val := e.psc.Receive()
			if v, ok := val.(error); ok {
				e.ch <- v
				close(e.ch)
				break
			}
			e.ch <- val
		}
	}()

	for _, i := range []string{"ql:log", "canceled", "completed", "failed", "popped", "stalled", "put", "track", "untrack"} {
		err := e.psc.Subscribe(i)
		if err != nil {
			close(e.ch)
			return nil, err
		}
	}

	return e.ch, nil
}

func (e *Events) Unsubscribe() {
	if e.psc != nil {
		e.psc.Unsubscribe()
		e.psc.Close()
	}
}