100 lines
3.3 KiB
Go
100 lines
3.3 KiB
Go
package app
|
||
|
||
import (
|
||
"context"
|
||
|
||
"hyapp/pkg/appcode"
|
||
"hyapp/pkg/rocketmqx"
|
||
servicemq "hyapp/pkg/servicekit/mq"
|
||
"hyapp/pkg/walletmq"
|
||
"hyapp/services/wallet-service/internal/config"
|
||
walletservice "hyapp/services/wallet-service/internal/service/wallet"
|
||
)
|
||
|
||
func (a *App) startMQ() error {
|
||
if err := servicemq.StartProducers([]*rocketmqx.Producer{a.outboxProducer, a.realtimeOutboxProducer}); err != nil {
|
||
return err
|
||
}
|
||
if err := servicemq.StartConsumers([]*rocketmqx.Consumer{a.projectionConsumer}); err != nil {
|
||
servicemq.ShutdownProducers([]*rocketmqx.Producer{a.outboxProducer, a.realtimeOutboxProducer})
|
||
return err
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func (a *App) shutdownMQ() {
|
||
servicemq.ShutdownConsumers([]*rocketmqx.Consumer{a.projectionConsumer})
|
||
servicemq.ShutdownProducers([]*rocketmqx.Producer{a.outboxProducer, a.realtimeOutboxProducer})
|
||
}
|
||
|
||
func newWalletProjectionConsumer(cfg config.Config, svc *walletservice.Service) (*rocketmqx.Consumer, error) {
|
||
if svc == nil || !cfg.ProjectionWorker.Enabled {
|
||
return nil, nil
|
||
}
|
||
// 钱包自有投影只关心礼物墙和徽章发放事实；兼容 selector 同时保留
|
||
// legacy Tag，待旧消息排空后可在后续阶段只留下这三个 typed Tag。
|
||
tagExpression, err := walletmq.LegacyCompatibleTagExpression(
|
||
walletmq.EventTypeWalletGiftDebited,
|
||
walletmq.EventTypeResourceGranted,
|
||
walletmq.EventTypeResourceGroupGranted,
|
||
)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
consumer, err := rocketmqx.NewConsumer(rocketMQProjectionConsumerConfig(cfg.RocketMQ, cfg.ProjectionWorker))
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
workerID := "wallet-projection-" + cfg.NodeID
|
||
if err := consumer.Subscribe(cfg.RocketMQ.WalletOutbox.Topic, tagExpression, func(ctx context.Context, message rocketmqx.ConsumedMessage) error {
|
||
walletMessage, err := walletmq.DecodeWalletOutboxMessage(message.Body)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
_, err = svc.ProcessWalletProjectionMessage(appcode.WithContext(ctx, walletMessage.AppCode), workerID, walletMessage, cfg.ProjectionWorker.LockTTL)
|
||
return err
|
||
}); err != nil {
|
||
_ = consumer.Shutdown()
|
||
return nil, err
|
||
}
|
||
return consumer, nil
|
||
}
|
||
|
||
func rocketMQProjectionConsumerConfig(cfg config.RocketMQConfig, projection config.ProjectionWorkerConfig) rocketmqx.ConsumerConfig {
|
||
return rocketmqx.ConsumerConfig{
|
||
EndpointConfig: rocketmqx.EndpointConfig{
|
||
NameServers: cfg.NameServers,
|
||
NameServerDomain: cfg.NameServerDomain,
|
||
AccessKey: cfg.AccessKey,
|
||
SecretKey: cfg.SecretKey,
|
||
SecurityToken: cfg.SecurityToken,
|
||
Namespace: cfg.Namespace,
|
||
VIPChannel: cfg.VIPChannel,
|
||
},
|
||
GroupName: projection.ConsumerGroup,
|
||
MaxReconsumeTimes: projection.ConsumerMaxReconsumeTimes,
|
||
ConsumePullBatch: int32(projection.BatchSize),
|
||
}
|
||
}
|
||
|
||
func rocketMQProducerConfig(cfg config.RocketMQConfig, groupName string) rocketmqx.ProducerConfig {
|
||
return rocketmqx.ProducerConfig{
|
||
EndpointConfig: rocketmqx.EndpointConfig{
|
||
NameServers: cfg.NameServers,
|
||
NameServerDomain: cfg.NameServerDomain,
|
||
AccessKey: cfg.AccessKey,
|
||
SecretKey: cfg.SecretKey,
|
||
SecurityToken: cfg.SecurityToken,
|
||
Namespace: cfg.Namespace,
|
||
VIPChannel: cfg.VIPChannel,
|
||
},
|
||
GroupName: groupName,
|
||
SendTimeout: cfg.SendTimeout,
|
||
Retry: cfg.Retry,
|
||
}
|
||
}
|
||
|
||
func shutdownProducers(producers ...*rocketmqx.Producer) {
|
||
servicemq.ShutdownProducers(producers)
|
||
}
|