wip

node/pkg/fetcher/fetcher.go

@@ -100,6 +100,8 @@ func (f *Fetcher) fetch(chainHelpers map[string]ChainHelper, proxies []Proxy) ([
 					errChan <- fetchErr
 					return
 				}
+			case *definition.Type == "websocket":
+				return
 			default:
 				errChan <- errorSentinel.ErrFetcherInvalidType
 			}

node/pkg/wfetcher/binance/binance.go

@@ -0,0 +1,54 @@
+package binance
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+
+	"bisonai.com/orakl/node/pkg/wss"
+)
+
+func Test() error {
+	ctx := context.Background()
+	conn, err := wss.NewConnection(ctx, wss.WithEndpoint(URL))
+	if err != nil {
+		return err
+	}
+
+	sub := subscription{
+		Method: "SUBSCRIBE",
+		Params: []Stream{
+			"ethusdt@miniTicker",
+		},
+		Id: 1,
+	}
+
+	err = conn.Write(ctx, sub)
+	if err != nil {
+		return err
+	}
+
+	ch := conn.ReadV2(ctx)
+	for {
+		select {
+		case msg := <-ch:
+			rawData, ok := msg.(map[string]interface{})
+			if !ok {
+				return fmt.Errorf("received data is not a map[string]interface{}")
+			}
+			jsonData, err := json.Marshal(rawData)
+			if err != nil {
+				return err
+			}
+
+			var miniTicker MiniTicker
+			err = json.Unmarshal(jsonData, &miniTicker)
+			if err != nil {
+				return err
+			}
+
+			fmt.Println(miniTicker)
+
+		}
+	}
+}

node/pkg/wfetcher/binance/type.go

@@ -0,0 +1,34 @@
+package binance
+
+import "fmt"
+
+const (
+	URL = "wss://stream.binance.com:443/ws"
+)
+
+type Stream string
+
+type MiniTicker struct {
+	EventType   string `json:"e"`
+	EventTime   int64  `json:"E"`
+	Symbol      string `json:"s"`
+	Price       string `json:"c"`
+	OpenPrice   string `json:"o"`
+	HighPrice   string `json:"h"`
+	LowPrice    string `json:"l"`
+	Volume      string `json:"v"`
+	QuoteVolume string `json:"q"`
+}
+
+func (m MiniTicker) String() string {
+	return fmt.Sprintf(
+		"EventType: %s, EventTime: %d, Symbol: %s, Price: %s, OpenPrice: %s, HighPrice: %s, LowPrice: %s, Volume: %s, QuoteVolume: %s\n",
+		m.EventType, m.EventTime, m.Symbol, m.Price, m.OpenPrice, m.HighPrice, m.LowPrice, m.Volume, m.QuoteVolume,
+	)
+}
+
+type subscription struct {
+	Method string   `json:"method"`
+	Params []Stream `json:"params"`
+	Id     uint32   `json:"id"`
+}

node/pkg/wss/utils.go

@@ -92,6 +92,23 @@ func (ws *wsConn) Read(ctx context.Context, ch chan interface{}) {
 	}
 }
 
+func (ws *wsConn) ReadV2(ctx context.Context) chan interface{} {
+	ch := make(chan interface{})
+	go func() {
+		defer close(ch)
+		for {
+			var t interface{}
+			err := wsjson.Read(ctx, ws.Conn, &t)
+			if err != nil {
+				log.Error().Err(err).Msg("error reading from websocket")
+				break
+			}
+			ch <- t
+		}
+	}()
+	return ch
+}
+
 func (ws *wsConn) Close() error {
 	err := ws.Conn.Close(websocket.StatusNormalClosure, "")
 	if err != nil {

node/script/test_websocket/main.go

@@ -0,0 +1,15 @@
+package main
+
+import (
+	"sync"
+
+	"bisonai.com/orakl/node/pkg/wfetcher/binance"
+)
+
+func main() {
+	var wg sync.WaitGroup
+	wg.Add(1)
+	binance.Test()
+	wg.Wait()
+
+}