[Update] Add task timeout #1
			
				
			
		
		
		
	| 
						 | 
					@ -1,5 +1,7 @@
 | 
				
			||||||
package queue
 | 
					package queue
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					import "time"
 | 
				
			||||||
 | 
					
 | 
				
			||||||
// Config ...
 | 
					// Config ...
 | 
				
			||||||
type Config struct {
 | 
					type Config struct {
 | 
				
			||||||
	// For message queue
 | 
						// For message queue
 | 
				
			||||||
| 
						 | 
					@ -10,6 +12,8 @@ type Config struct {
 | 
				
			||||||
	// https://github.com/hibiken/asynq/wiki/Queue-Priority
 | 
						// https://github.com/hibiken/asynq/wiki/Queue-Priority
 | 
				
			||||||
	Concurrency int
 | 
						Concurrency int
 | 
				
			||||||
	Priority    ConfigPriority
 | 
						Priority    ConfigPriority
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
						TaskTimeout time.Duration
 | 
				
			||||||
}
 | 
					}
 | 
				
			||||||
 | 
					
 | 
				
			||||||
// ConfigRedis ...
 | 
					// ConfigRedis ...
 | 
				
			||||||
| 
						 | 
					
 | 
				
			||||||
| 
						 | 
					@ -12,6 +12,8 @@ type Instance struct {
 | 
				
			||||||
	Client    *asynq.Client
 | 
						Client    *asynq.Client
 | 
				
			||||||
	Server    *asynq.ServeMux
 | 
						Server    *asynq.ServeMux
 | 
				
			||||||
	Scheduler *asynq.Scheduler
 | 
						Scheduler *asynq.Scheduler
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
						Configs Config
 | 
				
			||||||
}
 | 
					}
 | 
				
			||||||
 | 
					
 | 
				
			||||||
var instance Instance
 | 
					var instance Instance
 | 
				
			||||||
| 
						 | 
					@ -29,6 +31,7 @@ func NewInstance(cfg Config) Instance {
 | 
				
			||||||
	instance.Server = initServer(redisConn, cfg)
 | 
						instance.Server = initServer(redisConn, cfg)
 | 
				
			||||||
	instance.Scheduler = initScheduler(redisConn)
 | 
						instance.Scheduler = initScheduler(redisConn)
 | 
				
			||||||
	instance.Client = initClient(redisConn)
 | 
						instance.Client = initClient(redisConn)
 | 
				
			||||||
 | 
						instance.Configs = cfg
 | 
				
			||||||
 | 
					
 | 
				
			||||||
	// Return instance
 | 
						// Return instance
 | 
				
			||||||
	return instance
 | 
						return instance
 | 
				
			||||||
| 
						 | 
					
 | 
				
			||||||
							
								
								
									
										10
									
								
								task.go
								
								
								
								
							
							
						
						
									
										10
									
								
								task.go
								
								
								
								
							| 
						 | 
					@ -2,6 +2,11 @@ package queue
 | 
				
			||||||
 | 
					
 | 
				
			||||||
import (
 | 
					import (
 | 
				
			||||||
	"github.com/hibiken/asynq"
 | 
						"github.com/hibiken/asynq"
 | 
				
			||||||
 | 
						"time"
 | 
				
			||||||
 | 
					)
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
					const (
 | 
				
			||||||
 | 
						NoTimeout time.Duration = 0
 | 
				
			||||||
)
 | 
					)
 | 
				
			||||||
 | 
					
 | 
				
			||||||
// RunTask ...
 | 
					// RunTask ...
 | 
				
			||||||
| 
						 | 
					@ -22,6 +27,11 @@ func (i Instance) RunTask(typename string, payload []byte, priority string, retr
 | 
				
			||||||
	}
 | 
						}
 | 
				
			||||||
	options = append(options, asynq.MaxRetry(retryTimes))
 | 
						options = append(options, asynq.MaxRetry(retryTimes))
 | 
				
			||||||
 | 
					
 | 
				
			||||||
 | 
						// Task timeout
 | 
				
			||||||
 | 
						if i.Configs.TaskTimeout != 0 {
 | 
				
			||||||
 | 
							options = append(options, asynq.Timeout(i.Configs.TaskTimeout))
 | 
				
			||||||
 | 
						}
 | 
				
			||||||
 | 
					
 | 
				
			||||||
	// Enqueue task
 | 
						// Enqueue task
 | 
				
			||||||
	return i.Client.Enqueue(task, options...)
 | 
						return i.Client.Enqueue(task, options...)
 | 
				
			||||||
}
 | 
					}
 | 
				
			||||||
| 
						 | 
					
 | 
				
			||||||
		Loading…
	
		Reference in New Issue