加入Currency&Market自动更新机制

liomen · Sep 8, 2019 · 2c34d32 · 2c34d32
1 parent fca7bce
commit 2c34d32
Show file tree

Hide file tree

Showing 10 changed files with 76 additions and 2 deletions.
diff --git a/api/api.go b/api/api.go
@@ -51,6 +51,7 @@ func main() {
 	<-quit
 	fmt.Println("accepted signal")
 	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	initializers.DeleteListeQueue()
 	defer cancel()
 	if err := e.Shutdown(ctx); err != nil {
 		fmt.Println("shutting down failed, err:" + err.Error())
@@ -78,6 +79,7 @@ func initialize() {
 
 	initializers.LoadInterfaces()
 	initializers.InitI18n()
+	initializers.LoadCacheData()
 
 	err := ioutil.WriteFile("pids/api.pid", []byte(strconv.Itoa(os.Getpid())), 0644)
 	if err != nil {

diff --git a/config/amqp.yml.example b/config/amqp.yml.example
@@ -18,6 +18,8 @@ exchange:
   cancel:
     key: goDCE.order.cancel
     type: direct
+  fanout:
+    name: goDCE.fanout
 
 queue:
   matching:

diff --git a/initializers/cacheData.go b/initializers/cacheData.go
@@ -0,0 +1,48 @@
+package initializers
+
+import (
+	"fmt"
+
+	. "github.com/oldfritter/goDCE/models"
+	"github.com/oldfritter/goDCE/utils"
+)
+
+var QueueName string
+
+func InitCacheData() {
+	db := utils.MainDbBegin()
+	defer db.DbRollback()
+	InitAllCurrencies(db)
+	InitAllMarkets(db)
+}
+
+func LoadCacheData() {
+	go func() {
+		channel, err := utils.RabbitMqConnect.Channel()
+		if err != nil {
+			fmt.Errorf("Channel: %s", err)
+		}
+		channel.ExchangeDeclare(utils.AmqpGlobalConfig.Exchange.Fanout["name"], "fanout", true, false, false, false, nil)
+		queue, err := channel.QueueDeclare("", true, false, false, false, nil)
+		if err != nil {
+			return
+		}
+		QueueName = queue.Name
+		channel.QueueBind(queue.Name, QueueName, utils.AmqpGlobalConfig.Exchange.Fanout["name"], false, nil)
+		msgs, _ := channel.Consume(queue.Name, "", true, false, false, false, nil)
+		for _ = range msgs {
+			InitCacheData()
+		}
+		return
+	}()
+
+}
+
+func DeleteListeQueue() {
+	channel, err := utils.RabbitMqConnect.Channel()
+	if err != nil {
+		fmt.Errorf("Channel: %s", err)
+	}
+	channel.QueueDelete(QueueName, false, false, false)
+
+}
diff --git a/models/currency.go b/models/currency.go
@@ -1,7 +1,8 @@
 package models
 
 import (
-// "regexp"
+	// "regexp"
+	"github.com/oldfritter/goDCE/utils"
 )
 
 type Currency struct {
@@ -18,6 +19,11 @@ type Currency struct {
 	Depositable bool   `json:"depositable"`
 }
 
+var AllCurrencies []Currency
+
+func InitAllCurrencies(db *utils.GormDB) {
+	db.Where("visible = ?", true).Find(&AllCurrencies)
+}
 func (currency *Currency) IsEthereum() (result bool) {
 	if currency.Code == "eth" || currency.Erc20 || currency.Erc23 {
 		result = true

diff --git a/models/market.go b/models/market.go
@@ -39,6 +39,12 @@ type Market struct {
 	Running         bool   `json:"-" sql:"-"`
 }
 
+var Markets []Market
+
+func InitAllMarkets(db *utils.GormDB) {
+	db.Where("visible = ?", true).Find(&Markets)
+}
+
 // Exchange
 func (assignment *Market) MatchingExchange() string {
 	return utils.AmqpGlobalConfig.Exchange.Matching["key"]

diff --git a/order/cancel.go b/order/cancel.go
@@ -8,6 +8,7 @@ import (
 	"strconv"
 
 	envConfig "github.com/oldfritter/goDCE/config"
+	"github.com/oldfritter/goDCE/initializers"
 	"github.com/oldfritter/goDCE/models"
 	"github.com/oldfritter/goDCE/order/cancel"
 	"github.com/oldfritter/goDCE/utils"
@@ -20,6 +21,7 @@ func main() {
 	quit := make(chan os.Signal)
 	signal.Notify(quit, os.Interrupt)
 	<-quit
+	initializers.DeleteListeQueue()
 	closeResource()
 }
 
@@ -30,6 +32,7 @@ func initialize() {
 	models.AutoMigrations()
 	utils.InitRedisPools()
 	utils.InitializeAmqpConfig()
+	initializers.LoadCacheData()
 
 	err := ioutil.WriteFile("pids/cancel.pid", []byte(strconv.Itoa(os.Getpid())), 0644)
 	if err != nil {

diff --git a/trade/matching.go b/trade/matching.go
@@ -8,6 +8,7 @@ import (
 	"strconv"
 
 	envConfig "github.com/oldfritter/goDCE/config"
+	"github.com/oldfritter/goDCE/initializers"
 	"github.com/oldfritter/goDCE/models"
 	"github.com/oldfritter/goDCE/trade/matching"
 	"github.com/oldfritter/goDCE/utils"
@@ -20,6 +21,7 @@ func main() {
 	quit := make(chan os.Signal)
 	signal.Notify(quit, os.Interrupt)
 	<-quit
+	initializers.DeleteListeQueue()
 	closeResource()
 }
 
@@ -30,6 +32,7 @@ func initialize() {
 	models.AutoMigrations()
 	utils.InitRedisPools()
 	utils.InitializeAmqpConfig()
+	initializers.LoadCacheData()
 
 	err := ioutil.WriteFile("pids/matching.pid", []byte(strconv.Itoa(os.Getpid())), 0644)
 	if err != nil {

diff --git a/trade/treat.go b/trade/treat.go
@@ -8,6 +8,7 @@ import (
 	"strconv"
 
 	envConfig "github.com/oldfritter/goDCE/config"
+	"github.com/oldfritter/goDCE/initializers"
 	"github.com/oldfritter/goDCE/models"
 	"github.com/oldfritter/goDCE/trade/treat"
 	"github.com/oldfritter/goDCE/utils"
@@ -21,6 +22,7 @@ func main() {
 	quit := make(chan os.Signal)
 	signal.Notify(quit, os.Interrupt)
 	<-quit
+	initializers.DeleteListeQueue()
 	closeResource()
 }
 
@@ -32,6 +34,7 @@ func initialize() {
 	utils.InitRedisPools()
 	utils.InitializeAmqpConfig()
 	sneakerWorkers.InitWorkers()
+	initializers.LoadCacheData()
 
 	err := ioutil.WriteFile("pids/treat.pid", []byte(strconv.Itoa(os.Getpid())), 0644)
 	if err != nil {

diff --git a/utils/rabbitmq.go b/utils/rabbitmq.go
@@ -24,6 +24,7 @@ type Amqp struct {
 		Matching map[string]string `yaml:"matching"`
 		Trade    map[string]string `yaml:"trade"`
 		Cancel   map[string]string `yaml:"cancel"`
+		Fanout   map[string]string `yaml:"fanout"`
 	} `yaml:"exchange"`
 
 	Queue struct {

diff --git a/workers/sneakerWorkers/tickerWorker.go b/workers/sneakerWorkers/tickerWorker.go
@@ -28,7 +28,7 @@ func buildTicker(marketId int) {
 
 	now := time.Now()
 	begin := now.Add(-time.Hour * 24)
-	ticker := Ticker{MarketId: marketId, Name: market.Name, Code: market.Code}
+	ticker := Ticker{MarketId: marketId, Name: market.Name}
 	mainDB.Model(Trade{}).Order("id ASC").Where("market_id = ?", marketId).Where("? <= created_at AND created_at < ?", begin, now).Select("min(price) as low").Scan(&ticker.TickerAspect)
 	mainDB.Model(Trade{}).Order("id ASC").Where("market_id = ?", marketId).Where("? <= created_at AND created_at < ?", begin, now).Select("max(price) as high").Scan(&ticker.TickerAspect)
 	mainDB.Model(Trade{}).Order("id ASC").Where("market_id = ?", marketId).Where("? <= created_at AND created_at < ?", begin, now).Select("last(price) as last").Scan(&ticker.TickerAspect)