feat(订阅配置): 支持环境变量展开并校验订阅连接地址
Motivation: 订阅连接地址中的敏感凭据不应硬编码到配置文件,改为通过环境变量注入,并自动跳过未配置或非 AMQP 协议的订阅,避免无效连接导致启动失败。 Changes: * 订阅 URL 加载时展开环境变量并去除首尾空白 * 新增 AMQP 协议校验,仅 amqp/amqps 地址视为有效连接 * 非 AMQP 协议的订阅(如未配置的占位符)不再参与激活 * docker-compose 环境变量改用映射格式并加引号,规避特殊字符解析问题 * 补充环境变量展开与占位符跳过相关的单元测试
This commit is contained in:
+3
-3
@@ -6,9 +6,9 @@ services:
|
|||||||
ports:
|
ports:
|
||||||
- "8080:8080"
|
- "8080:8080"
|
||||||
environment:
|
environment:
|
||||||
- SMTP_PASSWORD=mssp.92QdSYS.3z0vklo982xl7qrx.tSlSpXs
|
SMTP_PASSWORD: "mssp.92QdSYS.3z0vklo982xl7qrx.tSlSpXs"
|
||||||
- DB_PASSWORD=7Qay8mksnwrCffGi
|
DB_PASSWORD: "7Qay8mksnwrCffGi"
|
||||||
- RABBITMQ_URL=amqps://gfzknmdk:BXoIOszWGpokmyP3FeQ64LqIldw8kf2v@gerbil.rmq.cloudamqp.com:5671/gfzknmdk
|
RABBITMQ_URL: "amqps://gfzknmdk:BXoIOszWGpokmyP3FeQ64LqIldw8kf2v@gerbil.rmq.cloudamqp.com:5671/gfzknmdk"
|
||||||
restart: unless-stopped
|
restart: unless-stopped
|
||||||
networks:
|
networks:
|
||||||
- 1panel-network
|
- 1panel-network
|
||||||
|
|||||||
@@ -66,10 +66,17 @@ func (o StrategyOverride) QuantityMultiplierFor(action string) float64 {
|
|||||||
return v
|
return v
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func IsAMQPURL(u string) bool {
|
||||||
|
u = strings.TrimSpace(u)
|
||||||
|
return strings.HasPrefix(u, "amqp://") || strings.HasPrefix(u, "amqps://")
|
||||||
|
}
|
||||||
|
|
||||||
func (c *Config) NormalizeSubscriptions() error {
|
func (c *Config) NormalizeSubscriptions() error {
|
||||||
for i := range c.Subscriptions {
|
for i := range c.Subscriptions {
|
||||||
s := &c.Subscriptions[i]
|
s := &c.Subscriptions[i]
|
||||||
if s.URL == "" {
|
s.URL = strings.TrimSpace(expandEnv(s.URL))
|
||||||
|
if !IsAMQPURL(s.URL) {
|
||||||
|
s.URL = ""
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
if s.Queue == "" {
|
if s.Queue == "" {
|
||||||
@@ -100,7 +107,7 @@ func (c *Config) NormalizeSubscriptions() error {
|
|||||||
func (c *Config) ActiveSubscriptions() []SubscriptionConfig {
|
func (c *Config) ActiveSubscriptions() []SubscriptionConfig {
|
||||||
out := make([]SubscriptionConfig, 0, len(c.Subscriptions))
|
out := make([]SubscriptionConfig, 0, len(c.Subscriptions))
|
||||||
for _, s := range c.Subscriptions {
|
for _, s := range c.Subscriptions {
|
||||||
if s.URL != "" {
|
if IsAMQPURL(s.URL) {
|
||||||
out = append(out, s)
|
out = append(out, s)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -1,6 +1,10 @@
|
|||||||
package config
|
package config
|
||||||
|
|
||||||
import "testing"
|
import (
|
||||||
|
"os"
|
||||||
|
"path/filepath"
|
||||||
|
"testing"
|
||||||
|
)
|
||||||
|
|
||||||
func TestNormalizeSubscriptionDefaults(t *testing.T) {
|
func TestNormalizeSubscriptionDefaults(t *testing.T) {
|
||||||
cfg := &Config{Subscriptions: []SubscriptionConfig{{
|
cfg := &Config{Subscriptions: []SubscriptionConfig{{
|
||||||
@@ -47,6 +51,47 @@ func TestNormalizeUnknownFormatter(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestLoadExpandsRabbitMQURL(t *testing.T) {
|
||||||
|
dir := t.TempDir()
|
||||||
|
path := filepath.Join(dir, "config.yaml")
|
||||||
|
yaml := []byte(`
|
||||||
|
server:
|
||||||
|
port: 8080
|
||||||
|
admin_key: test
|
||||||
|
subscriptions:
|
||||||
|
- name: trade-signal
|
||||||
|
url: "${RABBITMQ_URL}"
|
||||||
|
queue: q
|
||||||
|
source: trade-signal
|
||||||
|
`)
|
||||||
|
if err := os.WriteFile(path, yaml, 0o644); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
t.Setenv("RABBITMQ_URL", "amqps://user:pass@example.invalid:5671/vhost")
|
||||||
|
cfg, err := Load(path)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if len(cfg.Subscriptions) != 1 {
|
||||||
|
t.Fatalf("subs=%d", len(cfg.Subscriptions))
|
||||||
|
}
|
||||||
|
if cfg.Subscriptions[0].URL != "amqps://user:pass@example.invalid:5671/vhost" {
|
||||||
|
t.Fatalf("url=%q", cfg.Subscriptions[0].URL)
|
||||||
|
}
|
||||||
|
if n := len(cfg.ActiveSubscriptions()); n != 1 {
|
||||||
|
t.Fatalf("active=%d", n)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestActiveSubscriptionsSkipsPlaceholderURL(t *testing.T) {
|
||||||
|
cfg := &Config{Subscriptions: []SubscriptionConfig{{
|
||||||
|
URL: "${RABBITMQ_URL}", Queue: "q", Source: "s", Name: "trade-signal",
|
||||||
|
}}}
|
||||||
|
if n := len(cfg.ActiveSubscriptions()); n != 0 {
|
||||||
|
t.Fatalf("placeholder should not be active, n=%d", n)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
func TestQuantityMultiplierFor(t *testing.T) {
|
func TestQuantityMultiplierFor(t *testing.T) {
|
||||||
o := StrategyOverride{QuantityMultipliers: QuantityMultipliers{Open: 100, Add: 0}}
|
o := StrategyOverride{QuantityMultipliers: QuantityMultipliers{Open: 100, Add: 0}}
|
||||||
if o.QuantityMultiplierFor("OPEN") != 100 {
|
if o.QuantityMultiplierFor("OPEN") != 100 {
|
||||||
|
|||||||
Reference in New Issue
Block a user