Fix: config overwritten by default values

app.go

@@ -166,7 +166,7 @@ func GetConfigByPrefix(config *viper.Viper, prefix string, trimPrefix bool) *vip
 		if trimPrefix {
 			key = k[len(prefix):]
 		}
-		subConfig.SetDefault(key, v)
+		subConfig.Set(key, v)
 	}
 	return subConfig
 }

extensions/busext/amqp.go

@@ -169,7 +169,7 @@ func (b *BusExt) Push(exchange, routingKey string, data amqp.Publishing) error {
 
 func (b *BusExt) doPush(ctx context.Context, exchange, routingKey string, data amqp.Publishing, result chan error) {
 	log.Printf("Trying to publish: %s\n", data.Headers["id"])
-	for i := 0; i <= b.publishRetry; i++ {
+	for i := 0; i < b.publishRetry; i++ {
 		// clear staled confirmnation
 		// 有可能在超时之后才收到 confirm，会堵塞 channel，最终造成死锁
 		select {

extensions/busext/amqp_test.go

@@ -72,6 +72,12 @@ func TestPushConsume(t *testing.T) {
 	time.Sleep(2 * time.Second)
 	assert.Len(t, result, 100)
 
+	// check config works
+	assert.NotEqual(t, defaultPublishRetry, bus.publishRetry)
+	assert.Equal(t, 5, bus.publishRetry)
+	assert.NotEqual(t, defaultPushTimeout, bus.pushTimeout)
+	assert.Equal(t, time.Second * 3, bus.pushTimeout)
+
 	// mock amqp 的 publish, 使其 sleep 一个远比设置的 pushTimeout 长的时间, 模拟其卡死的情况
 	bus.publishFunc = func(exchange, key string, mandatory, immediate bool, msg amqp.Publishing) error {
 		dur := 100 * bus.pushTimeout

testdata/config.yaml

@@ -7,7 +7,8 @@ defaults: &defaults
   bus_queues:
     - test
   bus_resend_delay: "1s"
-  bus_publish_retry: 3
+  bus_publish_retry: 5
+  bus_push_timeout: "3s"
   bus_prefetch: 10
   bus_quit_consumer_on_empty_queue: false
   bus_bindings: