Improving structure, added End 2 End test and unified model for output

e2e_test.go

@@ -0,0 +1,56 @@
+package wsgateway_test
+
+import (
+	wsgateway "github.com/AlnsV/go-crypto-ws-gateway"
+	"github.com/AlnsV/go-crypto-ws-gateway/pkg/model"
+	"github.com/caarlos0/env"
+	"github.com/pkg/errors"
+	logger "github.com/sirupsen/logrus"
+	"testing"
+	"time"
+)
+
+type config struct {
+	FTXAPIKey    string `env:"FTX_API_KEY"`
+	FTXAPISecret string `env:"FTX_API_SECRET"`
+}
+
+func New() (*config, error) {
+	var cfg config
+
+	if err := env.Parse(&cfg); err != nil {
+		return nil, errors.Wrap(err, "error with initializing config")
+	}
+
+	return &cfg, nil
+}
+
+func TestGateway(t *testing.T) {
+	cfg, _ := New()
+	client, err := wsgateway.BuildWSClient(
+		"FTX",
+		cfg.FTXAPIKey,
+		cfg.FTXAPISecret,
+	)
+	if err != nil {
+		logger.Error(err)
+	}
+
+	err = client.Connect()
+	if err != nil {
+		logger.Error(err)
+	}
+
+	err = client.Listen(
+		[]string{"BTC-PERP", "SOL-PERP"},
+		func(trade *model.Trade) {
+			logger.Infoln(trade)
+		},
+	)
+	if err != nil {
+		logger.Error(err)
+	}
+
+	time.Sleep(2 * time.Second)
+	client.Close()
+}

wsgateway/exchanges/ftx.go → exchanges/ftx.go

@@ -5,29 +5,33 @@ import (
 	"crypto/sha256"
 	"encoding/hex"
 	"fmt"
-	"github.com/AlnsV/go-crypto-ws-gateway/wsgateway/internal"
-	"github.com/sirupsen/logrus"
+	"github.com/AlnsV/go-crypto-ws-gateway/internal"
+	"github.com/AlnsV/go-crypto-ws-gateway/pkg/model"
+	"github.com/AlnsV/go-crypto-ws-gateway/pkg/parse"
+	logger "github.com/sirupsen/logrus"
 	"net/http"
 	"time"
 )
 
 const address = "wss://ftx.com/ws/"
 
+var (
+	timezone, _ = time.LoadLocation("UTC")
+)
+
 type FTXWSClient struct {
-	APIKey    string
+	APIKey string
+
 	APISecret string
 
 	ws *internal.WebsocketClient
-
-	logger *logrus.Logger
 }
 
 func NewFTXWSClient(APIKey string, APISecret string) *FTXWSClient {
 	return &FTXWSClient{
 		APIKey:    APIKey,
 		APISecret: APISecret,
 		ws:        internal.NewWebsocketClient(address, true),
-		logger:    logrus.New(),
 	}
 }
 
@@ -43,7 +47,7 @@ func (f FTXWSClient) Connect() error {
 	if err != nil {
 		return err
 	}
-	f.logger.Infoln(response)
+	logger.Infoln(response)
 
 	ts := time.Now().Second()
 	loginMsg := map[string]interface{}{
@@ -57,7 +61,7 @@ func (f FTXWSClient) Connect() error {
 		},
 	}
 
-	f.logger.Infoln(loginMsg)
+	logger.Infoln(loginMsg)
 
 	err = f.ws.SendMessageJSON(loginMsg)
 	if err != nil {
@@ -75,7 +79,7 @@ func (f FTXWSClient) subscribe(pairs []string) error {
 			"market":  pair,
 		}
 
-		f.logger.Infoln(loginMsg)
+		logger.Infoln(loginMsg)
 
 		err = f.ws.SendMessageJSON(loginMsg)
 		if err != nil {
@@ -85,7 +89,7 @@ func (f FTXWSClient) subscribe(pairs []string) error {
 	return err
 }
 
-func (f FTXWSClient) Listen(instruments []string, receiver func(map[string]interface{})) error {
+func (f FTXWSClient) Listen(instruments []string, receiver func(trade *model.Trade)) error {
 	err := f.subscribe(instruments)
 	if err != nil {
 		return err
@@ -97,9 +101,23 @@ func (f FTXWSClient) Listen(instruments []string, receiver func(map[string]inter
 		for {
 			select {
 			case msg := <-messageContainer:
-				f.logger.Infoln(msg)
-				// TODO(JV): Unify msg format
-				receiver(msg)
+				logger.Infoln(msg)
+
+				if trades, ok := msg["data"]; ok {
+					for _, rawTrade := range trades.([]interface{}) {
+						trade := rawTrade.(map[string]interface{})
+						timestamp, _ := parse.ParseTimestamp(trade["time"].(string), timezone)
+
+						newTrade := &model.Trade{
+							Price:     trade["price"].(float64),
+							Side:      trade["side"].(string),
+							Size:      trade["size"].(float64),
+							Timestamp: timestamp,
+							Market:    msg["market"].(string),
+						}
+						receiver(newTrade)
+					}
+				}
 			}
 		}
 	}()
@@ -108,6 +126,5 @@ func (f FTXWSClient) Listen(instruments []string, receiver func(map[string]inter
 }
 
 func (f FTXWSClient) Close() {
-	//TODO implement me
-	panic("implement me")
+	f.ws.Close()
 }

wsgateway/factory.go → factory.go

@@ -2,18 +2,19 @@ package wsgateway
 
 import (
 	"fmt"
-	"github.com/AlnsV/go-crypto-ws-gateway/wsgateway/exchanges"
+	"github.com/AlnsV/go-crypto-ws-gateway/exchanges"
+	"github.com/AlnsV/go-crypto-ws-gateway/pkg/model"
 )
 
 type ExchangeWSClient interface {
 	Connect() error
-	Listen([]string, func(map[string]interface{})) error
+	Listen([]string, func(trade *model.Trade)) error
 	Close()
 }
 
 func BuildWSClient(exchange, APIKey, APISecret string) (ExchangeWSClient, error) {
 	if exchange == "FTX" {
 		return exchanges.NewFTXWSClient(APIKey, APISecret), nil
 	}
-	return nil, fmt.Errorf("exchange: %s doesn't exists")
+	return nil, fmt.Errorf("exchange: %s doesn't exists", exchange)
 }

go.mod

@@ -1,10 +1,11 @@
 module github.com/AlnsV/go-crypto-ws-gateway
 
-
 go 1.17
 
 require (
+	github.com/caarlos0/env v3.5.0+incompatible
 	github.com/gorilla/websocket v1.4.2
+	github.com/pkg/errors v0.9.1
 	github.com/sirupsen/logrus v1.8.1
 )
 

go.sum

@@ -1,7 +1,11 @@
+github.com/caarlos0/env v3.5.0+incompatible h1:Yy0UN8o9Wtr/jGHZDpCBLpNrzcFLLM2yixi/rBrKyJs=
+github.com/caarlos0/env v3.5.0+incompatible/go.mod h1:tdCsowwCzMLdkqRYDlHpZCp2UooDD3MspDBjZ2AD02Y=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/gorilla/websocket v1.4.2 h1:+/TMaTYc4QFitKJxsQ7Yye35DkWvkdLcvGKqM+x0Ufc=
 github.com/gorilla/websocket v1.4.2/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
+github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
+github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
 github.com/sirupsen/logrus v1.8.1 h1:dJKuHgqk1NNQlqoA6BTlM1Wf9DOH3NBjQyu0h9+AZZE=

wsgateway/internal/ws.go → internal/ws.go

@@ -2,8 +2,8 @@ package internal
 
 import (
 	"encoding/json"
-	"fmt"
 	"github.com/gorilla/websocket"
+	logger "github.com/sirupsen/logrus"
 	"log"
 	"net/http"
 	"net/url"
@@ -65,9 +65,13 @@ func (ws *WebsocketClient) Listen(messageBuffer chan<- map[string]interface{}) {
 		var msg map[string]interface{}
 		err = json.Unmarshal(message, &msg)
 		if err != nil {
-			log.Fatal(fmt.Errorf("parsing failure in: %s, err: %s", message, err))
+			logger.Warningf("parsing failure in: %s, err: %s", message, err)
 		} else {
 			messageBuffer <- msg
 		}
 	}
 }
+
+func (ws *WebsocketClient) Close() {
+	ws.Close()
+}

pkg/model/trade.go

@@ -0,0 +1,19 @@
+package model
+
+import "time"
+
+type Trade struct {
+	//Price of the settled trade
+	Price float64
+
+	// Side of trade, may be Buy or Sell
+	Side string
+
+	// Size Amount of currency traded
+	Size float64
+
+	Timestamp time.Time
+
+	// Market base-quote pair
+	Market string
+}

pkg/parse/timestamp.go

@@ -0,0 +1,31 @@
+package parse
+
+import (
+	"strings"
+	"time"
+)
+
+func ParseTimestamp(timestamp string, tz *time.Location) (time.Time, error) {
+	date, err := time.ParseInLocation(time.RFC3339Nano, timestamp, tz)
+	if err != nil {
+		date, errTwo := time.ParseInLocation(
+			time.RFC3339Nano,
+			strings.Replace(timestamp, " ", "T", 1),
+			tz,
+		)
+		if errTwo != nil {
+			dateStr := strings.Replace(timestamp, " ", "T", 1)
+			date, errThree := time.ParseInLocation(
+				time.RFC3339Nano,
+				dateStr+"+00:00",
+				tz,
+			)
+			if errThree != nil {
+				return time.Now().UTC(), errThree
+			}
+			return date, nil
+		}
+		return date, nil
+	}
+	return date, nil
+}

pkg/parse/timestamp_test.go

@@ -0,0 +1,33 @@
+package parse
+
+import (
+	"log"
+	"testing"
+	"time"
+)
+
+var (
+	timezone, _ = time.LoadLocation("UTC")
+)
+
+func TestParseTimestamp(t *testing.T) {
+	date, err := ParseTimestamp("2020-03-14T12:02:05.000234", timezone)
+	if err != nil {
+		t.Log(err)
+	}
+	log.Print(date)
+	dateTwo, errTwo := ParseTimestamp("2020-03-14 12:02:05.000234", timezone)
+	if errTwo != nil {
+		t.Log(errTwo)
+	}
+	log.Print(dateTwo)
+	dateThree, errThree := ParseTimestamp("2020-03-14T12:02:05.000234Z", timezone)
+	if errThree != nil {
+		t.Log(errThree)
+	}
+	log.Print(dateThree)
+	_, errFour := ParseTimestamp("2020-0314 12:005.000234Z", timezone)
+	if errFour == nil {
+		t.Log("timestamp should not be parsed")
+	}
+}