/
config.go
107 lines (93 loc) · 4.06 KB
/
config.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
package config
import (
"crypto/tls"
"fmt"
"strings"
"time"
)
var (
// Start with sensible default values
defaultCnf = &Config{
Broker: "amqp://guest:guest@localhost:5672/",
DefaultQueue: "machinery_tasks",
ResultBackend: "amqp://guest:guest@localhost:5672/",
ResultsExpireIn: 3600,
AMQP: &AMQPConfig{
Exchange: "machinery_exchange",
ExchangeType: "direct",
BindingKey: "machinery_task",
PrefetchCount: 3,
},
Redis: &RedisConfig{
MaxIdle: 3,
IdleTimeout: 240,
ReadTimeout: 15,
WriteTimeout: 15,
ConnectTimeout: 15,
},
}
reloadDelay = time.Second * 10
)
// Config holds all configuration for our program
type Config struct {
Broker string `yaml:"broker" envconfig:"BROKER"`
Concurrency int `yaml:"concurrency" envconfig:"CONCURRENCY"`
Maxcpucores int `yaml:"maxcpucores" envconfig:"MAXCPUCORES"`
Maxthreads int `yaml:"maxthreads" envconfig:"MAXTHREADS"`
CeleryMessage int `yaml:"celery_message" envconfig:"CELERY_MESSAGE"`
DefaultQueue string `yaml:"default_queue" envconfig:"DEFAULT_QUEUE"`
ResultBackend string `yaml:"result_backend" envconfig:"RESULT_BACKEND"`
ResultsExpireIn int `yaml:"results_expire_in" envconfig:"RESULTS_EXPIRE_IN"`
AMQP *AMQPConfig `yaml:"amqp"`
Redis *RedisConfig `yaml:"redis"`
TLSConfig *tls.Config
//NoUnixSignals when set disables signal handling in machinery
NoUnixSignals bool `yaml:"no_unix_signals" envconfig:"NO_UNIX_SIGNALS"`
}
// QueueBindingArgs arguments which are used when binding to the exchange
type QueueBindingArgs map[string]interface{}
// AMQPConfig wraps RabbitMQ related configuration
type AMQPConfig struct {
Exchange string `yaml:"exchange" envconfig:"AMQP_EXCHANGE"`
ExchangeType string `yaml:"exchange_type" envconfig:"AMQP_EXCHANGE_TYPE"`
QueueBindingArgs QueueBindingArgs `yaml:"queue_binding_args" envconfig:"AMQP_QUEUE_BINDING_ARGS"`
BindingKey string `yaml:"binding_key" envconfig:"AMQP_BINDING_KEY"`
PrefetchCount int `yaml:"prefetch_count" envconfig:"AMQP_PREFETCH_COUNT"`
TaskRoutes map[string]string `yaml:"task_routes"`
}
type RedisConfig struct {
// Maximum number of idle connections in the pool.
MaxIdle int `yaml:"max_idle" envconfig:"REDIS_MAX_IDLE"`
// Maximum number of connections allocated by the pool at a given time.
// When zero, there is no limit on the number of connections in the pool.
MaxActive int `yaml:"max_active" envconfig:"REDIS_MAX_ACTIVE"`
// Close connections after remaining idle for this duration in seconds. If the value
// is zero, then idle connections are not closed. Applications should set
// the timeout to a value less than the server's timeout.
IdleTimeout int `yaml:"max_idle_timeout" envconfig:"REDIS_IDLE_TIMEOUT"`
// If Wait is true and the pool is at the MaxActive limit, then Get() waits
// for a connection to be returned to the pool before returning.
Wait bool `yaml:"wait" envconfig:"REDIS_WAIT"`
// ReadTimeout specifies the timeout in seconds for reading a single command reply.
ReadTimeout int `yaml:"read_timeout" envconfig:"REDIS_READ_TIMEOUT"`
// WriteTimeout specifies the timeout in seconds for writing a single command.
WriteTimeout int `yaml:"write_timeout" envconfig:"REDIS_WRITE_TIMEOUT"`
// ConnectTimeout specifies the timeout in seconds for connecting to the Redis server when
// no DialNetDial option is specified.
ConnectTimeout int `yaml:"connect_timeout" envconfig:"REDIS_CONNECT_TIMEOUT"`
}
// Decode from yaml to map (any field whose type or pointer-to-type implements
// envconfig.Decoder can control its own deserialization)
func (args *QueueBindingArgs) Decode(value string) error {
pairs := strings.Split(value, ",")
mp := make(map[string]interface{}, len(pairs))
for _, pair := range pairs {
kvpair := strings.Split(pair, ":")
if len(kvpair) != 2 {
return fmt.Errorf("invalid map item: %q", pair)
}
mp[kvpair[0]] = kvpair[1]
}
*args = QueueBindingArgs(mp)
return nil
}