mirror of
https://github.com/d0zingcat/gocryptotrader.git
synced 2026-05-13 23:16:45 +00:00
b949388994
* Add exchange manager to engine
* Several improvements for engine and friends
1) New file.Exists func
2) gRPC TLS cert expiration date check and regeneration
3) New donation var for use across the codebase
4) Use Go log package until the logger is initialised
* Add cert tests and create dir tree if it doesn't exist for file.Write
* Link up donation address to documentation tool plus minor adjustments
* Fix remaining donation addrs
* Move non-needed reload exchange funcs
* Revert accidental config_example.json changes 🕯️
* Use go logger for logging until the logger has initiliased, otherwise no output will be seen
* Link up portfolio delay val and other fixes
* Run go mod tidy after dependabot PR
* Address nitterinos
603 lines
18 KiB
Go
603 lines
18 KiB
Go
package engine
|
|
|
|
import (
|
|
"errors"
|
|
"strings"
|
|
"sync/atomic"
|
|
"time"
|
|
|
|
"github.com/thrasher-corp/gocryptotrader/currency"
|
|
"github.com/thrasher-corp/gocryptotrader/exchanges/asset"
|
|
"github.com/thrasher-corp/gocryptotrader/exchanges/ticker"
|
|
log "github.com/thrasher-corp/gocryptotrader/logger"
|
|
)
|
|
|
|
// const holds the sync item types
|
|
const (
|
|
SyncItemTicker = iota
|
|
SyncItemOrderbook
|
|
SyncItemTrade
|
|
|
|
DefaultSyncerWorkers = 15
|
|
DefaultSyncerTimeout = time.Second * 15
|
|
)
|
|
|
|
var (
|
|
createdCounter = 0
|
|
removedCounter = 0
|
|
)
|
|
|
|
// NewCurrencyPairSyncer starts a new CurrencyPairSyncer
|
|
func NewCurrencyPairSyncer(c CurrencyPairSyncerConfig) (*ExchangeCurrencyPairSyncer, error) {
|
|
if !c.SyncOrderbook && !c.SyncTicker && !c.SyncTrades {
|
|
return nil, errors.New("no sync items enabled")
|
|
}
|
|
|
|
if c.NumWorkers <= 0 {
|
|
c.NumWorkers = DefaultSyncerWorkers
|
|
}
|
|
|
|
if c.SyncTimeout <= time.Duration(0) {
|
|
c.SyncTimeout = DefaultSyncerTimeout
|
|
}
|
|
|
|
s := ExchangeCurrencyPairSyncer{
|
|
Cfg: CurrencyPairSyncerConfig{
|
|
SyncTicker: c.SyncTicker,
|
|
SyncOrderbook: c.SyncOrderbook,
|
|
SyncTrades: c.SyncTrades,
|
|
SyncContinuously: c.SyncContinuously,
|
|
SyncTimeout: c.SyncTimeout,
|
|
NumWorkers: c.NumWorkers,
|
|
},
|
|
}
|
|
|
|
s.tickerBatchLastRequested = make(map[string]time.Time)
|
|
|
|
log.Debugf(log.SyncMgr,
|
|
"Exchange currency pair syncer config: continuous: %v ticker: %v"+
|
|
" orderbook: %v trades: %v workers: %v verbose: %v timeout: %v\n",
|
|
s.Cfg.SyncContinuously, s.Cfg.SyncTicker, s.Cfg.SyncOrderbook,
|
|
s.Cfg.SyncTrades, s.Cfg.NumWorkers, s.Cfg.Verbose, s.Cfg.SyncTimeout)
|
|
return &s, nil
|
|
}
|
|
|
|
func (e *ExchangeCurrencyPairSyncer) get(exchangeName string, p currency.Pair, a asset.Item) (*CurrencyPairSyncAgent, error) {
|
|
e.mux.Lock()
|
|
defer e.mux.Unlock()
|
|
|
|
for x := range e.CurrencyPairs {
|
|
if e.CurrencyPairs[x].Exchange == exchangeName &&
|
|
e.CurrencyPairs[x].Pair.Equal(p) &&
|
|
e.CurrencyPairs[x].AssetType == a {
|
|
return &e.CurrencyPairs[x], nil
|
|
}
|
|
}
|
|
|
|
return nil, errors.New("exchange currency pair syncer not found")
|
|
}
|
|
|
|
func (e *ExchangeCurrencyPairSyncer) exists(exchangeName string, p currency.Pair, a asset.Item) bool {
|
|
e.mux.Lock()
|
|
defer e.mux.Unlock()
|
|
|
|
for x := range e.CurrencyPairs {
|
|
if e.CurrencyPairs[x].Exchange == exchangeName &&
|
|
e.CurrencyPairs[x].Pair.Equal(p) &&
|
|
e.CurrencyPairs[x].AssetType == a {
|
|
return true
|
|
}
|
|
}
|
|
return false
|
|
}
|
|
|
|
func (e *ExchangeCurrencyPairSyncer) add(c *CurrencyPairSyncAgent) {
|
|
e.mux.Lock()
|
|
defer e.mux.Unlock()
|
|
|
|
if e.Cfg.SyncTicker {
|
|
if e.Cfg.Verbose {
|
|
log.Debugf(log.SyncMgr,
|
|
"%s: Added ticker sync item %v: using websocket: %v using REST: %v\n",
|
|
c.Exchange, FormatCurrency(c.Pair).String(), c.Ticker.IsUsingWebsocket,
|
|
c.Ticker.IsUsingREST)
|
|
}
|
|
if atomic.LoadInt32(&e.initSyncCompleted) != 1 {
|
|
e.initSyncWG.Add(1)
|
|
createdCounter++
|
|
}
|
|
}
|
|
|
|
if e.Cfg.SyncOrderbook {
|
|
if e.Cfg.Verbose {
|
|
log.Debugf(log.SyncMgr,
|
|
"%s: Added orderbook sync item %v: using websocket: %v using REST: %v\n",
|
|
c.Exchange, FormatCurrency(c.Pair).String(), c.Orderbook.IsUsingWebsocket,
|
|
c.Orderbook.IsUsingREST)
|
|
}
|
|
if atomic.LoadInt32(&e.initSyncCompleted) != 1 {
|
|
e.initSyncWG.Add(1)
|
|
createdCounter++
|
|
}
|
|
}
|
|
|
|
if e.Cfg.SyncTrades {
|
|
if e.Cfg.Verbose {
|
|
log.Debugf(log.SyncMgr,
|
|
"%s: Added trade sync item %v: using websocket: %v using REST: %v\n",
|
|
c.Exchange, FormatCurrency(c.Pair).String(), c.Trade.IsUsingWebsocket,
|
|
c.Trade.IsUsingREST)
|
|
}
|
|
if atomic.LoadInt32(&e.initSyncCompleted) != 1 {
|
|
e.initSyncWG.Add(1)
|
|
createdCounter++
|
|
}
|
|
}
|
|
|
|
c.Created = time.Now()
|
|
e.CurrencyPairs = append(e.CurrencyPairs, *c)
|
|
}
|
|
|
|
func (e *ExchangeCurrencyPairSyncer) remove(c *CurrencyPairSyncAgent) {
|
|
e.mux.Lock()
|
|
defer e.mux.Unlock()
|
|
|
|
for x := range e.CurrencyPairs {
|
|
if e.CurrencyPairs[x].Exchange == c.Exchange &&
|
|
e.CurrencyPairs[x].Pair.Equal(c.Pair) &&
|
|
e.CurrencyPairs[x].AssetType == c.AssetType {
|
|
e.CurrencyPairs = append(e.CurrencyPairs[:x], e.CurrencyPairs[x+1:]...)
|
|
return
|
|
}
|
|
}
|
|
}
|
|
|
|
func (e *ExchangeCurrencyPairSyncer) isProcessing(exchangeName string, p currency.Pair, a asset.Item, syncType int) bool {
|
|
e.mux.Lock()
|
|
defer e.mux.Unlock()
|
|
|
|
for x := range e.CurrencyPairs {
|
|
if e.CurrencyPairs[x].Exchange == exchangeName &&
|
|
e.CurrencyPairs[x].Pair.Equal(p) &&
|
|
e.CurrencyPairs[x].AssetType == a {
|
|
switch syncType {
|
|
case SyncItemTicker:
|
|
return e.CurrencyPairs[x].Ticker.IsProcessing
|
|
case SyncItemOrderbook:
|
|
return e.CurrencyPairs[x].Orderbook.IsProcessing
|
|
case SyncItemTrade:
|
|
return e.CurrencyPairs[x].Trade.IsProcessing
|
|
}
|
|
}
|
|
}
|
|
|
|
return false
|
|
}
|
|
|
|
func (e *ExchangeCurrencyPairSyncer) setProcessing(exchangeName string, p currency.Pair, a asset.Item, syncType int, processing bool) {
|
|
e.mux.Lock()
|
|
defer e.mux.Unlock()
|
|
|
|
for x := range e.CurrencyPairs {
|
|
if e.CurrencyPairs[x].Exchange == exchangeName &&
|
|
e.CurrencyPairs[x].Pair.Equal(p) &&
|
|
e.CurrencyPairs[x].AssetType == a {
|
|
switch syncType {
|
|
case SyncItemTicker:
|
|
e.CurrencyPairs[x].Ticker.IsProcessing = processing
|
|
case SyncItemOrderbook:
|
|
e.CurrencyPairs[x].Orderbook.IsProcessing = processing
|
|
case SyncItemTrade:
|
|
e.CurrencyPairs[x].Trade.IsProcessing = processing
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
func (e *ExchangeCurrencyPairSyncer) update(exchangeName string, p currency.Pair, a asset.Item, syncType int, err error) {
|
|
if atomic.LoadInt32(&e.initSyncStarted) != 1 {
|
|
return
|
|
}
|
|
|
|
switch syncType {
|
|
case SyncItemOrderbook, SyncItemTrade, SyncItemTicker:
|
|
if !e.Cfg.SyncOrderbook && syncType == SyncItemOrderbook {
|
|
return
|
|
}
|
|
|
|
if !e.Cfg.SyncTicker && syncType == SyncItemTicker {
|
|
return
|
|
}
|
|
|
|
if !e.Cfg.SyncTrades && syncType == SyncItemTrade {
|
|
return
|
|
}
|
|
default:
|
|
log.Warnf(log.SyncMgr, "ExchangeCurrencyPairSyncer: unknown sync item %v\n", syncType)
|
|
return
|
|
}
|
|
|
|
e.mux.Lock()
|
|
defer e.mux.Unlock()
|
|
|
|
for x := range e.CurrencyPairs {
|
|
if e.CurrencyPairs[x].Exchange == exchangeName &&
|
|
e.CurrencyPairs[x].Pair.Equal(p) &&
|
|
e.CurrencyPairs[x].AssetType == a {
|
|
switch syncType {
|
|
case SyncItemTicker:
|
|
origHadData := e.CurrencyPairs[x].Ticker.HaveData
|
|
e.CurrencyPairs[x].Ticker.LastUpdated = time.Now()
|
|
if err != nil {
|
|
e.CurrencyPairs[x].Ticker.NumErrors++
|
|
}
|
|
e.CurrencyPairs[x].Ticker.HaveData = true
|
|
e.CurrencyPairs[x].Ticker.IsProcessing = false
|
|
if atomic.LoadInt32(&e.initSyncCompleted) != 1 && !origHadData {
|
|
removedCounter++
|
|
log.Debugf(log.SyncMgr, "%s ticker sync complete %v [%d/%d].\n",
|
|
exchangeName, FormatCurrency(p).String(), removedCounter, createdCounter)
|
|
e.initSyncWG.Done()
|
|
}
|
|
|
|
case SyncItemOrderbook:
|
|
origHadData := e.CurrencyPairs[x].Orderbook.HaveData
|
|
e.CurrencyPairs[x].Orderbook.LastUpdated = time.Now()
|
|
if err != nil {
|
|
e.CurrencyPairs[x].Orderbook.NumErrors++
|
|
}
|
|
e.CurrencyPairs[x].Orderbook.HaveData = true
|
|
e.CurrencyPairs[x].Orderbook.IsProcessing = false
|
|
if atomic.LoadInt32(&e.initSyncCompleted) != 1 && !origHadData {
|
|
removedCounter++
|
|
log.Debugf(log.SyncMgr, "%s orderbook sync complete %v [%d/%d].\n",
|
|
exchangeName, FormatCurrency(p).String(), removedCounter, createdCounter)
|
|
e.initSyncWG.Done()
|
|
}
|
|
|
|
case SyncItemTrade:
|
|
origHadData := e.CurrencyPairs[x].Trade.HaveData
|
|
e.CurrencyPairs[x].Trade.LastUpdated = time.Now()
|
|
if err != nil {
|
|
e.CurrencyPairs[x].Trade.NumErrors++
|
|
}
|
|
e.CurrencyPairs[x].Trade.HaveData = true
|
|
e.CurrencyPairs[x].Trade.IsProcessing = false
|
|
if atomic.LoadInt32(&e.initSyncCompleted) != 1 && !origHadData {
|
|
removedCounter++
|
|
log.Debugf(log.SyncMgr, "%s trade sync complete %v [%d/%d].\n",
|
|
exchangeName, FormatCurrency(p).String(), removedCounter, createdCounter)
|
|
e.initSyncWG.Done()
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
func (e *ExchangeCurrencyPairSyncer) worker() {
|
|
cleanup := func() {
|
|
log.Debugln(log.SyncMgr, "Exchange CurrencyPairSyncer worker shutting down.")
|
|
}
|
|
defer cleanup()
|
|
|
|
for atomic.LoadInt32(&e.shutdown) != 1 {
|
|
exchanges := GetExchanges()
|
|
for x := range exchanges {
|
|
exchangeName := exchanges[x].GetName()
|
|
assetTypes := exchanges[x].GetAssetTypes()
|
|
supportsREST := exchanges[x].SupportsREST()
|
|
supportsRESTTickerBatching := exchanges[x].SupportsRESTTickerBatchUpdates()
|
|
var usingREST bool
|
|
var usingWebsocket bool
|
|
var switchedToRest bool
|
|
if exchanges[x].SupportsWebsocket() && exchanges[x].IsWebsocketEnabled() {
|
|
ws, err := exchanges[x].GetWebsocket()
|
|
if err != nil {
|
|
log.Errorf(log.SyncMgr, "%s unable to get websocket pointer. Err: %s\n",
|
|
exchangeName, err)
|
|
usingREST = true
|
|
}
|
|
|
|
if ws.IsConnected() {
|
|
usingWebsocket = true
|
|
} else {
|
|
usingREST = true
|
|
}
|
|
} else if supportsREST {
|
|
usingREST = true
|
|
}
|
|
|
|
for y := range assetTypes {
|
|
enabledPairs := exchanges[x].GetEnabledPairs(assetTypes[y])
|
|
for i := range enabledPairs {
|
|
if atomic.LoadInt32(&e.shutdown) == 1 {
|
|
return
|
|
}
|
|
|
|
if !e.exists(exchangeName, enabledPairs[i], assetTypes[y]) {
|
|
c := CurrencyPairSyncAgent{
|
|
AssetType: assetTypes[y],
|
|
Exchange: exchangeName,
|
|
Pair: enabledPairs[i],
|
|
}
|
|
|
|
if e.Cfg.SyncTicker {
|
|
c.Ticker = SyncBase{
|
|
IsUsingREST: usingREST,
|
|
IsUsingWebsocket: usingWebsocket,
|
|
}
|
|
}
|
|
|
|
if e.Cfg.SyncOrderbook {
|
|
c.Orderbook = SyncBase{
|
|
IsUsingREST: usingREST,
|
|
IsUsingWebsocket: usingWebsocket,
|
|
}
|
|
}
|
|
|
|
if e.Cfg.SyncTrades {
|
|
c.Trade = SyncBase{
|
|
IsUsingREST: usingREST,
|
|
IsUsingWebsocket: usingWebsocket,
|
|
}
|
|
}
|
|
|
|
e.add(&c)
|
|
}
|
|
|
|
c, err := e.get(exchangeName, enabledPairs[i], assetTypes[y])
|
|
if err != nil {
|
|
log.Errorf(log.SyncMgr, "failed to get item. Err: %s\n", err)
|
|
continue
|
|
}
|
|
if switchedToRest && usingWebsocket {
|
|
log.Infof(log.SyncMgr,
|
|
"%s %s: Websocket re-enabled, switching from rest to websocket\n",
|
|
c.Exchange, FormatCurrency(enabledPairs[i]).String())
|
|
switchedToRest = false
|
|
}
|
|
if e.Cfg.SyncTicker {
|
|
if !e.isProcessing(exchangeName, c.Pair, c.AssetType, SyncItemTicker) {
|
|
if c.Ticker.LastUpdated.IsZero() || time.Since(c.Ticker.LastUpdated) > e.Cfg.SyncTimeout {
|
|
if c.Ticker.IsUsingWebsocket {
|
|
if time.Since(c.Created) < e.Cfg.SyncTimeout {
|
|
continue
|
|
}
|
|
|
|
if supportsREST {
|
|
e.setProcessing(c.Exchange, c.Pair, c.AssetType, SyncItemTicker, true)
|
|
c.Ticker.IsUsingWebsocket = false
|
|
c.Ticker.IsUsingREST = true
|
|
log.Warnf(log.SyncMgr,
|
|
"%s %s %s: No ticker update after %s, switching from websocket to rest\n",
|
|
c.Exchange,
|
|
FormatCurrency(enabledPairs[i]).String(),
|
|
strings.ToUpper(c.AssetType.String()),
|
|
e.Cfg.SyncTimeout,
|
|
)
|
|
switchedToRest = true
|
|
e.setProcessing(c.Exchange, c.Pair, c.AssetType, SyncItemTicker, false)
|
|
}
|
|
}
|
|
|
|
if c.Ticker.IsUsingREST {
|
|
e.setProcessing(c.Exchange, c.Pair, c.AssetType, SyncItemTicker, true)
|
|
var result *ticker.Price
|
|
var err error
|
|
|
|
if supportsRESTTickerBatching {
|
|
e.mux.Lock()
|
|
batchLastDone, ok := e.tickerBatchLastRequested[exchangeName]
|
|
if !ok {
|
|
e.tickerBatchLastRequested[exchangeName] = time.Time{}
|
|
}
|
|
e.mux.Unlock()
|
|
|
|
if batchLastDone.IsZero() || time.Since(batchLastDone) > e.Cfg.SyncTimeout {
|
|
e.mux.Lock()
|
|
if e.Cfg.Verbose {
|
|
log.Debugf(log.SyncMgr, "%s Init'ing REST ticker batching\n", exchangeName)
|
|
}
|
|
result, err = exchanges[x].UpdateTicker(c.Pair, c.AssetType)
|
|
e.tickerBatchLastRequested[exchangeName] = time.Now()
|
|
e.mux.Unlock()
|
|
} else {
|
|
if e.Cfg.Verbose {
|
|
log.Debugf(log.SyncMgr, "%s Using recent batching cache\n", exchangeName)
|
|
}
|
|
result, err = exchanges[x].FetchTicker(c.Pair, c.AssetType)
|
|
}
|
|
} else {
|
|
result, err = exchanges[x].UpdateTicker(c.Pair, c.AssetType)
|
|
}
|
|
printTickerSummary(result, c.Pair, c.AssetType, exchangeName, "REST", err)
|
|
if err == nil {
|
|
//nolint:gocritic Bot.CommsRelayer.StageTickerData(exchangeName, c.AssetType, result)
|
|
if Bot.Config.RemoteControl.WebsocketRPC.Enabled {
|
|
relayWebsocketEvent(result, "ticker_update", c.AssetType.String(), exchangeName)
|
|
}
|
|
}
|
|
e.update(c.Exchange, c.Pair, c.AssetType, SyncItemTicker, err)
|
|
}
|
|
} else {
|
|
time.Sleep(time.Millisecond * 50)
|
|
}
|
|
}
|
|
}
|
|
|
|
if e.Cfg.SyncOrderbook {
|
|
if !e.isProcessing(exchangeName, c.Pair, c.AssetType, SyncItemOrderbook) {
|
|
if c.Orderbook.LastUpdated.IsZero() || time.Since(c.Orderbook.LastUpdated) > e.Cfg.SyncTimeout {
|
|
if c.Orderbook.IsUsingWebsocket {
|
|
if time.Since(c.Created) < e.Cfg.SyncTimeout {
|
|
continue
|
|
}
|
|
if supportsREST {
|
|
e.setProcessing(c.Exchange, c.Pair, c.AssetType, SyncItemOrderbook, true)
|
|
c.Orderbook.IsUsingWebsocket = false
|
|
c.Orderbook.IsUsingREST = true
|
|
log.Warnf(log.SyncMgr,
|
|
"%s %s %s: No orderbook update after %s, switching from websocket to rest\n",
|
|
c.Exchange,
|
|
FormatCurrency(c.Pair).String(),
|
|
strings.ToUpper(c.AssetType.String()),
|
|
e.Cfg.SyncTimeout,
|
|
)
|
|
switchedToRest = true
|
|
e.setProcessing(c.Exchange, c.Pair, c.AssetType, SyncItemOrderbook, false)
|
|
}
|
|
}
|
|
|
|
e.setProcessing(c.Exchange, c.Pair, c.AssetType, SyncItemOrderbook, true)
|
|
result, err := exchanges[x].UpdateOrderbook(c.Pair, c.AssetType)
|
|
printOrderbookSummary(result, c.Pair, c.AssetType, exchangeName, "REST", err)
|
|
if err == nil {
|
|
//nolint:gocritic Bot.CommsRelayer.StageOrderbookData(exchangeName, c.AssetType, result)
|
|
if Bot.Config.RemoteControl.WebsocketRPC.Enabled {
|
|
relayWebsocketEvent(result, "orderbook_update", c.AssetType.String(), exchangeName)
|
|
}
|
|
}
|
|
e.update(c.Exchange, c.Pair, c.AssetType, SyncItemOrderbook, err)
|
|
} else {
|
|
time.Sleep(time.Millisecond * 50)
|
|
}
|
|
}
|
|
if e.Cfg.SyncTrades {
|
|
if !e.isProcessing(exchangeName, c.Pair, c.AssetType, SyncItemTrade) {
|
|
if c.Trade.LastUpdated.IsZero() || time.Since(c.Trade.LastUpdated) > e.Cfg.SyncTimeout {
|
|
e.setProcessing(c.Exchange, c.Pair, c.AssetType, SyncItemTrade, true)
|
|
e.update(c.Exchange, c.Pair, c.AssetType, SyncItemTrade, nil)
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
// Start starts an exchange currency pair syncer
|
|
func (e *ExchangeCurrencyPairSyncer) Start() {
|
|
log.Debugln(log.SyncMgr, "Exchange CurrencyPairSyncer started.")
|
|
exchanges := GetExchanges()
|
|
for x := range exchanges {
|
|
exchangeName := exchanges[x].GetName()
|
|
supportsWebsocket := exchanges[x].SupportsWebsocket()
|
|
assetTypes := exchanges[x].GetAssetTypes()
|
|
supportsREST := exchanges[x].SupportsREST()
|
|
|
|
if !supportsREST && !supportsWebsocket {
|
|
log.Warnf(log.SyncMgr,
|
|
"Loaded exchange %s does not support REST or Websocket.\n",
|
|
exchangeName)
|
|
continue
|
|
}
|
|
|
|
var usingWebsocket bool
|
|
var usingREST bool
|
|
if supportsWebsocket && exchanges[x].IsWebsocketEnabled() {
|
|
ws, err := exchanges[x].GetWebsocket()
|
|
if err != nil {
|
|
log.Errorf(log.SyncMgr, "%s failed to get websocket. Err: %s\n",
|
|
exchangeName, err)
|
|
usingREST = true
|
|
}
|
|
|
|
if !ws.IsConnected() && !ws.IsConnecting() {
|
|
go WebsocketDataHandler(ws)
|
|
|
|
err = ws.Connect()
|
|
if err != nil {
|
|
log.Errorf(log.SyncMgr, "%s websocket failed to connect. Err: %s\n",
|
|
exchangeName, err)
|
|
usingREST = true
|
|
} else {
|
|
usingWebsocket = true
|
|
}
|
|
} else {
|
|
usingWebsocket = true
|
|
}
|
|
} else if supportsREST {
|
|
usingREST = true
|
|
}
|
|
|
|
for y := range assetTypes {
|
|
enabledPairs := exchanges[x].GetEnabledPairs(assetTypes[y])
|
|
for i := range enabledPairs {
|
|
if e.exists(exchangeName, enabledPairs[i], assetTypes[y]) {
|
|
continue
|
|
}
|
|
|
|
c := CurrencyPairSyncAgent{
|
|
AssetType: assetTypes[y],
|
|
Exchange: exchangeName,
|
|
Pair: enabledPairs[i],
|
|
}
|
|
|
|
if e.Cfg.SyncTicker {
|
|
c.Ticker = SyncBase{
|
|
IsUsingREST: usingREST,
|
|
IsUsingWebsocket: usingWebsocket,
|
|
}
|
|
}
|
|
|
|
if e.Cfg.SyncOrderbook {
|
|
c.Orderbook = SyncBase{
|
|
IsUsingREST: usingREST,
|
|
IsUsingWebsocket: usingWebsocket,
|
|
}
|
|
}
|
|
|
|
if e.Cfg.SyncTrades {
|
|
c.Trade = SyncBase{
|
|
IsUsingREST: usingREST,
|
|
IsUsingWebsocket: usingWebsocket,
|
|
}
|
|
}
|
|
|
|
e.add(&c)
|
|
}
|
|
}
|
|
}
|
|
|
|
if atomic.CompareAndSwapInt32(&e.initSyncStarted, 0, 1) {
|
|
log.Debugf(log.SyncMgr,
|
|
"Exchange CurrencyPairSyncer initial sync started. %d items to process.\n",
|
|
createdCounter)
|
|
e.initSyncStartTime = time.Now()
|
|
}
|
|
|
|
go func() {
|
|
e.initSyncWG.Wait()
|
|
if atomic.CompareAndSwapInt32(&e.initSyncCompleted, 0, 1) {
|
|
log.Debugf(log.SyncMgr, "Exchange CurrencyPairSyncer initial sync is complete.\n")
|
|
completedTime := time.Now()
|
|
log.Debugf(log.SyncMgr, "Exchange CurrencyPairSyncer initial sync took %v [%v sync items].\n",
|
|
completedTime.Sub(e.initSyncStartTime), createdCounter)
|
|
|
|
if !e.Cfg.SyncContinuously {
|
|
log.Debugln(log.SyncMgr, "Exchange CurrencyPairSyncer stopping.")
|
|
e.Stop()
|
|
return
|
|
}
|
|
}
|
|
}()
|
|
|
|
if atomic.LoadInt32(&e.initSyncCompleted) == 1 && !e.Cfg.SyncContinuously {
|
|
return
|
|
}
|
|
|
|
for i := 0; i < e.Cfg.NumWorkers; i++ {
|
|
go e.worker()
|
|
}
|
|
}
|
|
|
|
// Stop shuts down the exchange currency pair syncer
|
|
func (e *ExchangeCurrencyPairSyncer) Stop() {
|
|
stopped := atomic.CompareAndSwapInt32(&e.shutdown, 0, 1)
|
|
if stopped {
|
|
log.Debugln(log.SyncMgr, "Exchange CurrencyPairSyncer stopped.")
|
|
}
|
|
}
|