Only send an error if there is a receiver

cmd/watcher/main.go

@@ -31,7 +31,7 @@ func main() {
 	r := ops.Registrar{}
 	r.Database = db
 	r.Listener = ls
-	r.SetLogger(logger)
+	r.Logger = logger
 	if err := r.Record("ETH-USD", "BTC-USD"); err != nil {
 		logger.Info("Error: %v", err)
 		return

gdax/listener_service.go

@@ -4,9 +4,10 @@ import (
 	"github.com/kcotugno/tacitus"
 
 	"errors"
-	"sync"
 )
 
+const buf = 1024
+
 type ListenerService struct {
 	Client ListenerClient
 	Logger tacitus.Logger
@@ -14,19 +15,11 @@ type ListenerService struct {
 	trades chan tacitus.Trade
 	err    chan error
 
-	subs []string
-
-	closed bool
-
-	shouldRestart bool
-	restMu        sync.Mutex
-
-	errorsMu   sync.Mutex
-	sendErrors bool
+	dirty bool
 }
 
 func (s *ListenerService) Open() error {
-	if s.closed {
+	if s.dirty {
 		return errors.New("Already used")
 	}
 
@@ -36,9 +29,8 @@ func (s *ListenerService) Open() error {
 	}
 	s.Logger.Info("GDAX listener started")
 
-	s.trades = make(chan tacitus.Trade, 1024)
+	s.trades = make(chan tacitus.Trade, buf)
 	s.err = make(chan error)
-	s.setRestart(true)
 
 	s.listen()
 
@@ -46,15 +38,22 @@ func (s *ListenerService) Open() error {
 }
 
 func (s *ListenerService) Close() error {
-	s.closed = true
-	s.setRestart(false)
 	s.Client.Close()
 	s.Logger.Info("GDAX listener stopped")
 
+	close(s.trades)
+	close(s.err)
+
 	return nil
 }
 
 func (s *ListenerService) Subscribe(products ...string) {
+	if s.dirty {
+		return
+	}
+
+	s.dirty = true
+
 	req := Request{}
 	req.Type = Subscribe
 
@@ -62,8 +61,6 @@ func (s *ListenerService) Subscribe(products ...string) {
 	req.Channels[0].Name = "matches"
 	req.Channels[0].ProductIds = products
 
-	s.subs = products
-
 	s.Client.Send(req)
 }
 
@@ -75,10 +72,6 @@ func (s *ListenerService) Error() <-chan error {
 	return s.err
 }
 
-func (s *ListenerService) SetLogger(logger tacitus.Logger) {
-	s.Logger = logger
-}
-
 func (s *ListenerService) listen() {
 	go func() {
 		for msg := range s.Client.Stream() {
@@ -106,50 +99,16 @@ func (s *ListenerService) listen() {
 		for err := range s.Client.Error() {
 			s.Logger.Info("Error from GDAX client: %v", err)
 
-			if s.SendErrors() {
-				s.err <- err
+			select {
+			case s.err <- err:
+			default:
 			}
+
+			close(s.trades)
+			close(s.err)
 		}
 
 		s.Logger.Info("Error closed")
-
-		if s.restart() {
-			err := s.Open()
-			if err != nil {
-				s.Logger.Info("GDAX Unable to reastablish connection")
-				s.Close()
-				return
-			}
-			s.Subscribe(s.subs...)
-		}
 	}()
 
 }
-
-func (s *ListenerService) restart() bool {
-	s.restMu.Lock()
-	defer s.restMu.Unlock()
-
-	return s.shouldRestart
-}
-
-func (s *ListenerService) setRestart(restart bool) {
-	s.restMu.Lock()
-	defer s.restMu.Unlock()
-
-	s.shouldRestart = restart
-}
-
-func (s *ListenerService) SendErrors() bool {
-	s.errorsMu.Lock()
-	defer s.errorsMu.Unlock()
-
-	return s.sendErrors
-}
-
-func (s *ListenerService) SetSendErrors(send bool) {
-	s.errorsMu.Lock()
-	defer s.errorsMu.Unlock()
-
-	s.sendErrors = send
-}

listener.go

@@ -6,7 +6,4 @@ type ListenerService interface {
 	Subscribe(products ...string)
 	Stream() <-chan Trade
 	Error() <-chan error
-	SendErrors() bool
-	SetSendErrors(send bool)
-	SetLogger(logger Logger)
 }

ops/registrar.go

@@ -7,8 +7,9 @@ import (
 type Registrar struct {
 	Database tacitus.DatabaseClientService
 	Listener tacitus.ListenerService
+	Logger tacitus.Logger
 
-	logger tacitus.Logger
+	err chan error
 }
 
 func (r *Registrar) Record(products ...string) error {
@@ -22,26 +23,38 @@ func (r *Registrar) Record(products ...string) error {
 	return nil
 }
 
-func (r *Registrar) SetLogger(logger tacitus.Logger) {
-	r.Listener.SetLogger(logger)
-	r.logger = logger
-}
-
 func (r *Registrar) record() {
 	go func() {
 		for t := range r.Listener.Stream() {
 			go func(trade tacitus.Trade) {
 				_, err := r.Database.TradeService().CreateTrade(trade)
 				if err != nil {
-					r.logger.Info("Error inserting trade: %v", err)
+					r.Logger.Info("Error inserting trade: %v", err)
+
+					select {
+					case r.err <- err:
+					default:
+					}
 				}
 			}(t)
 		}
+		r.Logger.Info("Trade goroutine done")
 	}()
 
 	go func() {
 		for err := range r.Listener.Error() {
-			r.logger.Info("Registrar received error: %v", err)
+			r.Logger.Info("Registrar received error: %v", err)
+
+			select {
+			case r.err <- err:
+			default:
+			}
 		}
+
+		r.Logger.Info("Error goroutine done")
 	}()
 }
+
+func (r *Registrar) Error() <-chan error {
+	return r.err
+}