feat: added poller and streamer components

Author: achannarasappa

internal/common/common.go

@@ -88,10 +88,10 @@ type Monitors struct {
 }
 
 type Monitor interface {
-	Connect(url string) error
+	Start() error
 	GetAssetQuotes(useCache ...bool) []AssetQuote
 	SetSymbols(symbols []string)
-	Close()
+	Stop() error
 }
 
 // Lot represents a cost basis lot

internal/monitor/coinbase/monitor.go

@@ -0,0 +1,141 @@
+package monitorCoinbase
+
+import (
+	"context"
+	"slices"
+	"sync"
+	"time"
+
+	"github.com/achannarasappa/ticker/v4/internal/common"
+	poller "github.com/achannarasappa/ticker/v4/internal/monitor/coinbase/poller"
+	streamer "github.com/achannarasappa/ticker/v4/internal/monitor/coinbase/streamer"
+	unary "github.com/achannarasappa/ticker/v4/internal/monitor/coinbase/unary"
+	resty "github.com/go-resty/resty/v2"
+)
+
+type MonitorCoinbase struct {
+	unaryAPI    *unary.UnaryAPI
+	streamer    *streamer.Streamer
+	poller      *poller.Poller
+	unary       *unary.UnaryAPI
+	input       input
+	symbols     []string
+	assetQuotes []common.AssetQuote
+	mu          sync.Mutex
+	ctx         context.Context
+	cancel      context.CancelFunc
+}
+
+type unaryAPI struct {
+	symbols []string
+	client  resty.Client
+}
+
+type input struct {
+	symbols           []string
+	symbolsUnderlying []string
+}
+
+// Config contains the required configuration for the Coinbase monitor
+type Config struct {
+	Client   resty.Client
+	OnUpdate func()
+}
+
+// Option defines an option for configuring the monitor
+type Option func(*MonitorCoinbase)
+
+func NewMonitorCoinbase(config Config, opts ...Option) *MonitorCoinbase {
+
+	ctx, cancel := context.WithCancel(context.Background())
+
+	unaryAPI := unary.NewUnaryAPI(config.Client)
+
+	monitor := &MonitorCoinbase{
+		streamer: streamer.NewStreamer(ctx),
+		poller:   poller.NewPoller(ctx, unaryAPI),
+		unaryAPI: unaryAPI,
+		ctx:      ctx,
+		cancel:   cancel,
+	}
+
+	for _, opt := range opts {
+		opt(monitor)
+	}
+
+	return monitor
+}
+
+// WithSymbolsUnderlying sets the underlying symbols for the monitor
+func WithSymbolsUnderlying(symbols []string) Option {
+	return func(m *MonitorCoinbase) {
+		m.input.symbolsUnderlying = symbols
+	}
+}
+
+// WithStreamingURL sets the streaming URL for the monitor
+func WithStreamingURL(url string) Option {
+	return func(m *MonitorCoinbase) {
+		m.streamer.SetURL(url)
+	}
+}
+
+// WithRefreshInterval sets the refresh interval for the monitor
+func WithRefreshInterval(interval time.Duration) Option {
+	return func(m *MonitorCoinbase) {
+		m.poller.SetRefreshInterval(interval)
+	}
+}
+
+func (m *MonitorCoinbase) GetAssetQuotes(ignoreCache ...bool) []common.AssetQuote {
+
+	if len(ignoreCache) > 0 && ignoreCache[0] {
+		return m.unaryAPI.GetAssetQuotes(m.symbols).AssetQuotes
+	}
+
+	return m.assetQuotes
+}
+
+func (m *MonitorCoinbase) SetSymbols(symbols []string) {
+
+	symbolsMerged := make([]string, 0)
+	symbolsMerged = append(symbolsMerged, m.input.symbolsUnderlying...)
+	symbolsMerged = append(symbolsMerged, symbols...)
+	slices.Sort(symbolsMerged)
+	symbolsMerged = slices.Compact(symbolsMerged)
+
+	m.input.symbols = symbols
+	m.symbols = symbolsMerged
+
+	// Execute one unary API call to get data not sent by streaming API and set initial prices
+	assetQuotesIndexed := m.unaryAPI.GetAssetQuotes(m.symbols) // TODO: update to return and handle error
+	m.assetQuotes = assetQuotesIndexed.AssetQuotes             // TODO: Add saving indexed quotes
+
+	// Coinbase steaming API for CBE (spot) only and not CDE (futures)
+	m.streamer.SetSymbolsAndUpdateSubscriptions(symbols) // TODO: update to return and handle error
+
+}
+
+// Start the monitor
+func (m *MonitorCoinbase) Start() error {
+
+	var err error
+
+	err = m.streamer.Start()
+	if err != nil {
+		return err
+	}
+
+	err = m.poller.Start()
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func (m *MonitorCoinbase) Stop() error {
+
+	m.cancel()
+	return nil
+}

internal/monitor/coinbase/monitor_suite_test.go

@@ -0,0 +1,29 @@
+package monitorCoinbase_test
+
+import (
+	"testing"
+
+	"github.com/go-resty/resty/v2"
+	"github.com/jarcoal/httpmock"
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+)
+
+var client = resty.New()
+
+var _ = BeforeSuite(func() {
+	httpmock.ActivateNonDefault(client.GetClient())
+})
+
+var _ = BeforeEach(func() {
+	httpmock.Reset()
+})
+
+var _ = AfterSuite(func() {
+	httpmock.DeactivateAndReset()
+})
+
+func TestCoinbase(t *testing.T) {
+	RegisterFailHandler(Fail)
+	RunSpecs(t, "Coinbase Suite")
+}

internal/monitor/coinbase/monitor_test.go

@@ -0,0 +1,31 @@
+package monitorCoinbase_test
+
+// import (
+// 	"time"
+
+// 	"github.com/go-resty/resty/v2"
+// 	. "github.com/onsi/ginkgo/v2"
+// 	. "github.com/onsi/gomega"
+
+// 	monitorCoinbase "github.com/achannarasappa/ticker/v4/internal/monitor/coinbase"
+// )
+
+// var _ = Describe("Monitor Coinbase", func() {
+
+// 	Describe("NewMonitorCoinbase", func() {
+// 		PIt("should return a new MonitorCoinbase", func() {
+// 			monitor := monitorCoinbase.NewMonitorCoinbase(time.Second, *resty.New(), []string{"BTC-USD"}, func() {})
+
+// 			Expect(monitor).NotTo(BeNil())
+// 		})
+// 	})
+
+// 	Describe("GetAssetQuotes", func() {
+// 		PIt("should return the asset quotes", func() {
+// 			monitor := monitorCoinbase.NewMonitorCoinbase(time.Second, *resty.New(), []string{"BTC-USD"}, func() {})
+
+// 			quotes := monitor.GetAssetQuotes()
+// 			Expect(quotes).NotTo(BeEmpty())
+// 		})
+// 	})
+// })

internal/monitor/coinbase/poller/poller.go

@@ -0,0 +1,80 @@
+package monitorCoinbase
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/achannarasappa/ticker/v4/internal/monitor/coinbase/unary"
+)
+
+type Poller struct {
+	refreshInterval time.Duration
+	symbols         []string
+	isStarted       bool
+	ctx             context.Context
+	cancel          context.CancelFunc
+	unaryAPI        *unary.UnaryAPI
+}
+
+func NewPoller(ctx context.Context, unaryAPI *unary.UnaryAPI) *Poller {
+	ctx, cancel := context.WithCancel(ctx)
+
+	return &Poller{
+		refreshInterval: 0,
+		isStarted:       false,
+		ctx:             ctx,
+		cancel:          cancel,
+		unaryAPI:        unaryAPI,
+	}
+}
+
+func (p *Poller) SetSymbols(symbols []string) {
+	p.symbols = symbols
+}
+
+func (p *Poller) SetRefreshInterval(interval time.Duration) error {
+
+	if p.isStarted {
+		return fmt.Errorf("cannot set refresh interval while poller is started")
+	}
+
+	p.refreshInterval = interval
+	return nil
+}
+
+func (p *Poller) Start() error {
+	if p.isStarted {
+		return fmt.Errorf("poller already started")
+	}
+
+	if p.refreshInterval <= 0 {
+		return fmt.Errorf("refresh interval is not set")
+	}
+
+	if len(p.symbols) == 0 {
+		return fmt.Errorf("symbols are not set")
+	}
+
+	p.isStarted = true
+
+	// Start polling goroutine
+	go func() {
+		ticker := time.NewTicker(p.refreshInterval)
+		defer ticker.Stop()
+
+		// Initial poll
+		p.unaryAPI.GetAssetQuotes(p.symbols)
+
+		for {
+			select {
+			case <-ticker.C:
+				p.unaryAPI.GetAssetQuotes(p.symbols)
+			case <-p.ctx.Done():
+				return
+			}
+		}
+	}()
+
+	return nil
+}