Only send an error if there is a receiver

Makefile

@@ -23,13 +23,13 @@ watcher: $(WATCHER)
 webapp: $(WEBAPP)
 
 get:
-	go get $(GETFLAGS) git.kevincotugno.com/kcotugno/tacitus
+	go get $(GETFLAGS) github.com/kcotugno/tacitus
-	go get $(GETFLAGS) git.kevincotugno.com/kcotugno/tacitus/cmd/interval
+	go get $(GETFLAGS) github.com/kcotugno/tacitus/cmd/interval
-	go get $(GETFLAGS) git.kevincotugno.com/kcotugno/tacitus/cmd/watcher
+	go get $(GETFLAGS) github.com/kcotugno/tacitus/cmd/watcher
-	go get $(GETFLAGS) git.kevincotugno.com/kcotugno/tacitus/cmd/webapp
+	go get $(GETFLAGS) github.com/kcotugno/tacitus/cmd/webapp
-	go get $(GETFLAGS) git.kevincotugno.com/kcotugno/tacitus/gdax
+	go get $(GETFLAGS) github.com/kcotugno/tacitus/gdax
-	go get $(GETFLAGS) git.kevincotugno.com/kcotugno/tacitus/gdax/websocket
+	go get $(GETFLAGS) github.com/kcotugno/tacitus/gdax/websocket
-	go get $(GETFLAGS) git.kevincotugno.com/kcotugno/tacitus/postgres
+	go get $(GETFLAGS) github.com/kcotugno/tacitus/postgres
 
 fmt: $(TACITUS_FILES) $(OSUTIL_FILES) $(POSTGRES_FILES) $(OPS_FILES) \
 	$(GDAX_FILES) $(GDAX_WEBSOCKET_FILES) $(HTTP_FILES)

cmd/interval/main.go

@@ -9,7 +9,7 @@ import (
 	"strconv"
 	"time"
 
-	"git.kevincotugno.com/kcotugno/tacitus/postgres"
+	"github.com/kcotugno/tacitus/postgres"
 )
 
 func main() {

cmd/watcher/main.go

@@ -1,91 +1,27 @@
 package main
 
 import (
-	"git.kevincotugno.com/kcotugno/tacitus"
+	"github.com/kcotugno/tacitus/gdax"
-	"git.kevincotugno.com/kcotugno/tacitus/gdax"
+	"github.com/kcotugno/tacitus/gdax/websocket"
-	"git.kevincotugno.com/kcotugno/tacitus/gdax/websocket"
+	"github.com/kcotugno/tacitus/ops"
-	"git.kevincotugno.com/kcotugno/tacitus/ops"
+	"github.com/kcotugno/tacitus/osutil"
-	"git.kevincotugno.com/kcotugno/tacitus/osutil"
+	"github.com/kcotugno/tacitus/postgres"
-	"git.kevincotugno.com/kcotugno/tacitus/postgres"
 
-	"flag"
-	"fmt"
-	"os"
-	"os/signal"
 	"time"
 )
 
-type Parameters struct {
-	Products []string
-
-	Host     string
-	Port     int
-	User     string
-	Name     string
-	Password string
-	SSL      string
-}
-
-type StringList []string
-
-func parseFlags() Parameters {
-	var p Parameters
-	p.Products = make([]string, 0)
-
-	host := flag.String("host", "localhost", "Database host")
-	port := flag.Int("port", 5432, "Database port")
-	user := flag.String("user", "gdax", "Database user")
-	db := flag.String("name", "gdax", "Database name")
-	password := flag.String("password", `""`, "Database password")
-	ssl := flag.String("ssl", "disable", "Database ssl")
-
-	products := make(StringList, 0)
-	flag.Var(&products, "product", "Product to watch. This option can "+
-		"be specified multiple time")
-
-	flag.Parse()
-
-	for _, v := range products {
-		if !tacitus.ValidProduct(v) {
-			fmt.Printf("error: %v is not a valid product", v)
-			os.Exit(128)
-		}
-	}
-
-	p.Products = []string(products)
-	p.Host = *host
-	p.Port = *port
-	p.User = *user
-	p.Name = *db
-	p.Password = *password
-	p.SSL = *ssl
-
-	return p
-}
-
 func main() {
-	params := parseFlags()
-	if len(params.Products) == 0 {
-		flag.PrintDefaults()
-		os.Exit(1)
-	}
-
 	logger := osutil.NewLogger()
 
 	db := postgres.NewClient()
-	db.Name = params.Name
+	db.Name = "gdax"
-	db.User = params.User
+	db.User = "gdax"
-	db.Password = params.Password
-	db.Host = params.Host
-	db.Port = params.Port
-	db.SslMode = params.SSL
 	db.SetLogger(logger)
 	err := db.Open()
 	if err != nil {
 		logger.Info("Error openning database connection: %v", err)
 		return
 	}
-	defer db.Close()
 
 	ws := websocket.NewClient()
 	ls := &gdax.ListenerService{}
@@ -95,50 +31,31 @@ func main() {
 	r := ops.Registrar{}
 	r.Database = db
 	r.Listener = ls
-	r.SetLogger(logger)
+	r.Logger = logger
-	if err := r.Record(params.Products...); err != nil {
+	if err := r.Record("ETH-USD", "BTC-USD"); err != nil {
 		logger.Info("Error: %v", err)
-		os.Exit(3)
+		return
 	}
 
-	defer ls.Close()
-
 	a := ops.Aggregator{}
 	a.Database = db
 	a.Logger = logger
-	a.Products = params.Products
+	a.Products = []string{"ETH-USD", "BTC-USD"}
 	a.Start(30 * time.Second)
 
 	v := ops.Validator{}
 	v.Database = db
 	v.Logger = logger
-	v.Products = params.Products
+	v.Products = []string{"ETH-USD", "BTC-USD"}
 	v.Start(30 * time.Second)
 
 	av := ops.AggregationValidator{}
 	av.Database = db
 	av.Logger = logger
-	av.Products = params.Products
+	av.Products = []string{"ETH-USD", "BTC-USD"}
 	av.Intervals = []time.Duration{30 * time.Second}
 	av.Start(30 * time.Second)
 
-	c := make(chan os.Signal)
+	t := make(chan bool)
-	signal.Notify(c, os.Interrupt)
+	<-t
-
-	var done bool
-	for !done {
-		select {
-		case <-c:
-			done = true
-		}
-	}
-}
-
-func (l *StringList) String() string {
-	return "List of strings"
-}
-
-func (l *StringList) Set(value string) error {
-	*l = append(*l, value)
-	return nil
 }

cmd/webapp/main.go

@@ -1,9 +1,9 @@
 package main
 
 import (
-	"git.kevincotugno.com/kcotugno/tacitus/http"
+	"github.com/kcotugno/tacitus/http"
-	"git.kevincotugno.com/kcotugno/tacitus/osutil"
+	"github.com/kcotugno/tacitus/osutil"
-	"git.kevincotugno.com/kcotugno/tacitus/postgres"
+	"github.com/kcotugno/tacitus/postgres"
 	"log"
 )
 

gdax/listener_service.go

@@ -1,12 +1,13 @@
 package gdax
 
 import (
-	"git.kevincotugno.com/kcotugno/tacitus"
+	"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
+	dirty bool
-
-	closed bool
-
-	shouldRestart bool
-	restMu        sync.Mutex
-
-	errorsMu   sync.Mutex
-	sendErrors 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() {
+			select {
-				s.err <- err
+			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
-}

gdax/public_client.go

@@ -1,7 +1,7 @@
 package gdax
 
 import (
-	"git.kevincotugno.com/kcotugno/tacitus"
+	"github.com/kcotugno/tacitus"
 
 	"encoding/json"
 	"net/http"

gdax/websocket/client.go

@@ -2,7 +2,7 @@ package websocket
 
 import (
 	"github.com/gorilla/websocket"
-	"git.kevincotugno.com/kcotugno/tacitus/gdax"
+	"github.com/kcotugno/tacitus/gdax"
 
 	"sync"
 	"time"

http/mux.go

@@ -1,7 +1,7 @@
 package http
 
 import (
-	"git.kevincotugno.com/kcotugno/tacitus"
+	"github.com/kcotugno/tacitus"
 
 	"net/http"
 )

http/server.go

@@ -4,7 +4,7 @@ import (
 	"net"
 	"net/http"
 
-	"git.kevincotugno.com/kcotugno/tacitus"
+	"github.com/kcotugno/tacitus"
 )
 
 const DefaultAddr = ":8080"

http/trade_handler.go

@@ -6,7 +6,7 @@ import (
 	"strconv"
 	"strings"
 
-	"git.kevincotugno.com/kcotugno/tacitus"
+	"github.com/kcotugno/tacitus"
 )
 
 type TradeHandler struct {

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/aggregation_validator.go

@@ -1,7 +1,7 @@
 package ops
 
 import (
-	"git.kevincotugno.com/kcotugno/tacitus"
+	"github.com/kcotugno/tacitus"
 
 	"sort"
 	"time"

ops/aggregator.go

@@ -1,7 +1,7 @@
 package ops
 
 import (
-	"git.kevincotugno.com/kcotugno/tacitus"
+	"github.com/kcotugno/tacitus"
 	"github.com/shopspring/decimal"
 
 	"sync"

ops/registrar.go

@@ -1,14 +1,15 @@
 package ops
 
 import (
-	"git.kevincotugno.com/kcotugno/tacitus"
+	"github.com/kcotugno/tacitus"
 )
 
 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
+}

ops/validator.go

@@ -1,8 +1,8 @@
 package ops
 
 import (
-	"git.kevincotugno.com/kcotugno/tacitus"
+	"github.com/kcotugno/tacitus"
-	"git.kevincotugno.com/kcotugno/tacitus/gdax"
+	"github.com/kcotugno/tacitus/gdax"
 
 	"sort"
 	"time"

postgres/aggregation_results.go

@@ -1,7 +1,7 @@
 package postgres
 
 import (
-	"git.kevincotugno.com/kcotugno/tacitus"
+	"github.com/kcotugno/tacitus"
 	"github.com/shopspring/decimal"
 
 	"database/sql"

postgres/aggregation_service.go

@@ -1,7 +1,7 @@
 package postgres
 
 import (
-	"git.kevincotugno.com/kcotugno/tacitus"
+	"github.com/kcotugno/tacitus"
 
 	"strings"
 	"time"

postgres/client.go

@@ -7,19 +7,15 @@ import (
 	"database/sql"
 	"text/template"
 
-	"git.kevincotugno.com/kcotugno/tacitus"
+	"github.com/kcotugno/tacitus"
 )
 
-const connStr = `host={{.Host}} dbname={{.Name}} port={{.Port}} user={{.User}} ` +
+const connStr = `user={{.User}} dbname={{.Name}} sslmode={{.SslMode}}`
-	`password={{.Password}} sslmode={{.SslMode}}`
 
 type Client struct {
-	Host     string
-	Port     int
 	Name    string
 	User    string
 	SslMode string
-	Password string
 
 	tradeService        TradeService
 	confirmationService ConfirmationService
@@ -34,12 +30,6 @@ func NewClient() *Client {
 	c.tradeService.client = &c
 	c.confirmationService.client = &c
 	c.aggregationService.client = &c
-
-	c.Host = "localhost"
-	c.Port = 5432
-	c.Name = "postgres"
-	c.User = "postgres"
-	c.Password = `""`
 	c.SslMode = "disable"
 
 	return &c

postgres/confirmation_service.go

@@ -1,7 +1,7 @@
 package postgres
 
 import (
-	"git.kevincotugno.com/kcotugno/tacitus"
+	"github.com/kcotugno/tacitus"
 )
 
 const (

postgres/trade_result.go

@@ -1,7 +1,7 @@
 package postgres
 
 import (
-	"git.kevincotugno.com/kcotugno/tacitus"
+	"github.com/kcotugno/tacitus"
 
 	"database/sql"
 	"errors"

postgres/trade_service.go

@@ -1,7 +1,7 @@
 package postgres
 
 import (
-	"git.kevincotugno.com/kcotugno/tacitus"
+	"github.com/kcotugno/tacitus"
 
 	"database/sql"
 	"strings"

product.go

@@ -1,37 +0,0 @@
-package tacitus
-
-import (
-	"strings"
-)
-
-type Product string
-
-const (
-	BtcUsd = "BTC-USD"
-	BtcEur = "BTC-EUR"
-	BtcGbp = "BTC-GBP"
-
-	BchUsd = "BCH-USD"
-
-	EthUsd = "ETH-USD"
-	EthBtc = "ETH-BTC"
-	EthEur = "ETH-EUR"
-
-	LtcUsd = "LTC-USD"
-	LtcBtc = "LTC-BTC"
-	LtcEur = "LTC-EUR"
-)
-
-var products = [...]string{BtcUsd, BtcEur, BtcGbp, BchUsd, EthUsd, EthBtc,
-	EthEur, LtcUsd, LtcBtc, LtcEur}
-
-func ValidProduct(prod string) bool {
-	prod = strings.ToUpper(prod)
-	for _, p := range products {
-		if p == prod {
-			return true
-		}
-	}
-
-	return false
-}

trade.go

@@ -3,9 +3,12 @@ package tacitus
 import (
 	"github.com/shopspring/decimal"
 
+	"strings"
 	"time"
 )
 
+var PRODUCTS = [...]string{"BTC-USD", "ETH-USD"}
+
 type Trade struct {
 	Id        int             `json:"-"`
 	TradeId   int             `json:"trade_id,"`
@@ -32,3 +35,14 @@ type TradeService interface {
 	TradesAfterAll(product string, id int) (Results, error)
 	TradesWhereResults(sql string, params ...interface{}) (Results, error)
 }
+
+func ValidProduct(prod string) bool {
+	prod = strings.ToUpper(prod)
+	for _, p := range PRODUCTS {
+		if p == prod {
+			return true
+		}
+	}
+
+	return false
+}