transcollect/taskprocess.go

package main

import (
	"commonapi"
	"errors"
	"fmt"
	"time"
	_ "lite"
	_ "bitcoin"
	_ "bucash"
)

func sysStart() {

	go dataProcesser(CPUNUM)
	httpNetWorkServer()
}

type taskHandler struct {
	retcode  int
	data     []byte
	complete chan bool
}

func NewtaskHandler() *taskHandler {

	taskhdl := &taskHandler{}
	taskhdl.complete = make(chan bool, 1)
	return taskhdl
}

func (s *taskHandler) waitProcessData() ([]byte, error) {

	timeout := time.NewTicker(time.Duration(20) * time.Second)
	select {
	case <-timeout.C:
		log.Error("deal request timeout")
		return nil, errors.New("timeout")
	case <-s.complete:
	}

	return s.data, nil

}

func dataProcesser(processNum int) {
	for i := 0; i < processNum; i++ {
		go func(index int) {
			for {
				task := <-taskChan[index]
				go processer(task)
			}
		}(i)

	}

}

func processer(task *taskInfo) {
	c, err := commonapi.New(task.cointype)
	if err != nil {

		sendstr := fmt.Sprintf(`{"errcode:%v,"msg":"%v"}`, INIT_DRIVER_ERR, GetErrMsg(INIT_DRIVER_ERR, nil))
		log.Error("sendstr:%v", sendstr)
		task.resp.data = []byte(sendstr)
		task.resp.complete <- false
		return
	}
	log.Debug("urlpath:%v", task.urlpath)
	resp, err := commonapi.Handler[task.urlpath](c, task.jshandler)
	if err != nil {

		task.resp.data = []byte(err.Error())
		task.resp.complete <- false
		return
	}

	task.resp.retcode = OK
	task.resp.data = resp
	task.resp.complete <- true

	return

}