tickserver/server/tick/ds.go

// Copyright 2013-2014 Fuzamei tech Ltd. All rights reserved.

package tick

// 本文件实现DataSource数据源的tick数据获取下载和保存

import (
	"log"
	"sync"
	"time"
)

var Debug bool = false // for debug

func SetDebug(debug bool) {
	Debug = debug
}

// 数据源接口
// lmax, easyforex, oanda, ctp以及未来的macoin等均实现此接口
type DataSource interface {
	Name() string
	Run()
	GetInstrument() []Instrument
	GetMarket() chan *Market
}

type DSBase struct {
	conf       *DsConf //配置
	chM        chan *Market
	insMap     map[int64]*Instrument
	countTotal int64
	countThrow int64
	mu         sync.Mutex
}

func NewDsBase(confIn *DsConf) *DSBase {
	dsb := &DSBase{
		chM:    make(chan *Market, 10240),
		insMap: make(map[int64]*Instrument),
		conf:   confIn,
	}
	return dsb
}

func (dsb *DSBase) Save(m *Market) {
	dsb.countTotal++
	select {
	case dsb.chM <- m:
	default:
		dsb.countThrow++
		//if dsb.countThrow%1000 == 0 {
		//log.Println("@@@:Save:", DataTypeName(int(m.Type)), m, m.InsId, m.LastPrice, dsb.countTotal, dsb.countThrow)
		//}
		//if m.Type == IntTdx && m.InsId == 1 {
		//log.Println("[DS.Save]data trace")
		//}
	}
}

func (dsb *DSBase) GetInstrument() []Instrument {
	dsb.mu.Lock()
	defer dsb.mu.Unlock()
	var ret []Instrument
	for _, in := range dsb.insMap {
		ret = append(ret, *in)
	}
	return ret
}

func (dsb *DSBase) GetMarket() chan *Market {
	return dsb.chM
}

func debugDelay(prefix, insId string, ts int64) {
	// for debug delay
	now := time.Now()
	d := int64(float64(now.UnixNano())*1e-6) - ts
	if Debug && d > 1000 {
		//log.Println(prefix, "delay > 1000ms", insId, d, getTime(ts))
	}
}

func printDelay(prefix, insId string, ts int64) {
	// for debug delay
	now := time.Now()
	d := int64(float64(now.UnixNano())*1e-6) - ts
	log.Println(prefix, "delay ", d, insId, getTime(ts))
}

func getTime(ts int64) time.Time {
	if ts < 0 {
		return time.Now()
	}
	return time.Unix(ts/1000, (ts%1000)*1e6)
}