update processor

exchain/exmonitor/klinegenerator.go

@@ -25,7 +25,7 @@ type KLineGeneratorJob struct {
 	ProcessorFactory CoinProcessorFactory
 }
 
-func NewKLineGeneratorJob(factory CoinProcessorFactory, service MarketService) *KLineGeneratorJob {
+func NewKLineGeneratorJob(factory CoinProcessorFactory) *KLineGeneratorJob {
 	return &KLineGeneratorJob{
 		ProcessorFactory: factory,
 	}

exchain/exmonitor/markethandle.go

@@ -14,7 +14,7 @@ type mongoMarketHandler struct {
 
 // HandleTrade inserts an ExchangeTrade into the corresponding collection
 func (h *mongoMarketHandler) HandleTrade(symbol string, trade *ExchangeTrade) error {
-	collection := h.db.Collection("exchange_trade_" + symbol)
+	collection := h.db.Collection(tradePrefix + symbol)
 	_, err := collection.InsertOne(context.Background(), trade)
 	if err != nil {
 		return fmt.Errorf("failed to insert trade: %v", err)
@@ -24,7 +24,7 @@ func (h *mongoMarketHandler) HandleTrade(symbol string, trade *ExchangeTrade) er
 
 // HandleKLine inserts a KLine into the corresponding collection
 func (h *mongoMarketHandler) HandleKLine(symbol string, kline *KLine) error {
-	collection := h.db.Collection("exchange_kline_" + symbol + "_" + kline.Period)
+	collection := h.db.Collection(klinePrefix + symbol + "_" + kline.Period)
 	_, err := collection.InsertOne(context.Background(), kline)
 	if err != nil {
 		return fmt.Errorf("failed to insert KLine: %v", err)

exchain/exmonitor/monitor.go

@@ -9,7 +9,7 @@ type Monitor struct {
 
 func NewMonitor(conf *config.Config) *Monitor {
 	service := NewMarketService(conf)
-	kline := NewKLineGeneratorJob(&coinProcessorFactory{}, service)
+	kline := NewKLineGeneratorJob(NewCoinProcessorFactory(service))
 
 	return &Monitor{
 		kline:   kline,

exchain/exmonitor/processor.go

@@ -41,7 +41,7 @@ type ExchangeTrade struct {
 
 type MarketHandler interface {
 	HandleTrade(symbol string, trade *ExchangeTrade) error
-	HandleKLine(symbol string, kline *KLine)
+	HandleKLine(symbol string, kline *KLine) error
 }
 
 type MarketService interface {
@@ -50,6 +50,7 @@ type MarketService interface {
 	FindTradeByTimeRange(symbol string, start, end int64) []*ExchangeTrade
 	FindAllKLineByTimeRange(symbol string, fromTime, toTime int64, period string) []*KLine
 	SaveKLine(symbol string, kline *KLine)
+	NewHandler(symbol string) MarketHandler
 }
 
 type DefaultCoinProcessor struct {
@@ -67,20 +68,30 @@ type DefaultCoinProcessor struct {
 
 type coinProcessorFactory struct {
 	mux            sync.Mutex
+	service        MarketService
 	coinProcessors map[string]*DefaultCoinProcessor
 }
 
+func NewCoinProcessorFactory(service MarketService) CoinProcessorFactory {
+	return &coinProcessorFactory{
+		service:        service,
+		coinProcessors: make(map[string]*DefaultCoinProcessor),
+	}
+
+}
+
 func (cpf *coinProcessorFactory) GetCoinProcessor(symbol, baseCoin string) *DefaultCoinProcessor {
 	cpf.mux.Lock()
 	defer cpf.mux.Unlock()
 	if processor, exists := cpf.coinProcessors[symbol]; exists {
 		return processor
 	} else {
+		// todo: initialize the processor with the service.
 		processor = &DefaultCoinProcessor{
 			symbol:       symbol,
 			baseCoin:     baseCoin,
 			currentKLine: createNewKLine(),
-			handlers:     []MarketHandler{},
+			handlers:     make([]MarketHandler, 0),
 			coinThumb:    &CoinThumb{},
 			isHalt:       true,
 			stopKLine:    false,