Function startQueueTask

queuetask.go:97–147

start queue task

(task *QueueTask)

//start queue task
func startQueueTask(task *QueueTask) {
	taskCtx := task.getTaskContext()
	handler := func() {
		defer func() {
			task.putTaskContext(taskCtx)
			if err := recover(); err != nil {
				task.CounterInfo().ErrorCounter.Inc(1)
				if task.taskService.ExceptionHandler != nil {
					task.taskService.ExceptionHandler(taskCtx, fmt.Errorf("%v", err))
				}
			}
		}()

		task.CounterInfo().RunCounter.Inc(1)
		//get value from message chan
		message := <-task.MessageChan
		taskCtx.Message = message

		if task.taskService != nil && task.taskService.OnBeforeHandler != nil {
			task.taskService.OnBeforeHandler(taskCtx)
		}

		var err error
		if !taskCtx.IsEnd {
			err = task.handler(taskCtx)
		}

		if err != nil {
			taskCtx.Error = err
			task.CounterInfo().ErrorCounter.Inc(1)
			if task.taskService != nil && task.taskService.ExceptionHandler != nil {
				task.taskService.ExceptionHandler(taskCtx, err)
			}
		}

		if task.taskService != nil && task.taskService.OnEndHandler != nil {
			task.taskService.OnEndHandler(taskCtx)
		}
	}
	dofunc := func() {
		task.TimeTicker = time.NewTicker(time.Duration(task.Interval) * time.Millisecond)
		handler()
		for {
			select {
			case <-task.TimeTicker.C:
				handler()
			}
		}
	}
	go dofunc()
}