Merge pull request #1923 from c9s/dboy/coinbase-stream-pr2

FEAT: [coinbase] add more message parsing, enhance stream functionality

pkg/exchange/coinbase/exchage.go

@@ -39,12 +39,21 @@ var log = logrus.WithField("exchange", ID)
 
 type Exchange struct {
 	client *api.RestAPIClient
+
+	// api keys
+	apiKey        string
+	apiSecret     string
+	apiPassphrase string
 }
 
 func New(key, secret, passphrase string, timeout time.Duration) *Exchange {
 	client := api.NewClient(key, secret, passphrase, timeout)
 	return &Exchange{
 		client: client,
+
+		apiKey:        key,
+		apiSecret:     secret,
+		apiPassphrase: passphrase,
 	}
 }
 
@@ -261,8 +270,7 @@ func (e *Exchange) CancelOrders(ctx context.Context, orders ...types.Order) erro
 
 // ExchangeMarketDataService
 func (e *Exchange) NewStream() types.Stream {
-	// TODO: implement stream
-	return nil
+	return NewStream(e, e.apiKey, e.apiPassphrase, e.apiSecret)
 }
 
 func (e *Exchange) QueryMarkets(ctx context.Context) (types.MarketMap, error) {

pkg/exchange/coinbase/parse.go

@@ -6,7 +6,7 @@ import (
 )
 
 // See https://docs.cdp.coinbase.com/exchange/docs/websocket-channels for message types
-func (s *Stream) parseMessage(data []byte) (interface{}, error) {
+func parseMessage(data []byte) (interface{}, error) {
 	var baseMsg messageBaseType
 	err := json.Unmarshal(data, &baseMsg)
 	if err != nil {
@@ -91,6 +91,27 @@ func (s *Stream) parseMessage(data []byte) (interface{}, error) {
 			return nil, err
 		}
 		return &activeMsg, nil
+	case "balance":
+		var balanceMsg BalanceMessage
+		err = json.Unmarshal(data, &balanceMsg)
+		if err != nil {
+			return nil, err
+		}
+		return &balanceMsg, nil
+	case "snapshot":
+		var snapshotMsg OrderBookSnapshotMessage
+		err = json.Unmarshal(data, &snapshotMsg)
+		if err != nil {
+			return nil, err
+		}
+		return &snapshotMsg, nil
+	case "l2update":
+		var updateMsg OrderBookUpdateMessage
+		err = json.Unmarshal(data, &updateMsg)
+		if err != nil {
+			return nil, err
+		}
+		return &updateMsg, nil
 	}
 	return nil, fmt.Errorf("unknown message type: %s", baseMsg.Type)
 }