liucaijun
/
voucher
2
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

cmb

This commit is contained in:
李子铭 2025-03-04 16:08:07 +08:00
parent e96e3dc777
commit effe78a62a
5 changed files with 51 additions and 50 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
internal/biz/cmb.go
View File

@ -16,7 +16,7 @@ func (v *VoucherBiz) CmbOrder(ctx context.Context, req *bo.OrderCreateReqBo) (or
return err
}
return nil
return v.PushOrderMQ(ctx, orderNo)
})
return

7
internal/biz/cmb/cmb.go
View File

@ -4,35 +4,28 @@ import (
"voucher/internal/biz/mixrepos"
"voucher/internal/biz/repo"
"voucher/internal/biz/wechatrepo"
"voucher/internal/conf"
"voucher/internal/data"
)
type Cmb struct {
bc *conf.Bootstrap
rdb *data.Rdb
OrderRepo repo.OrderRepo
ProductRepo repo.ProductRepo
MqSendMixRepo mixrepos.MQSendMixRepo
WechatCpnRepo wechatrepo.WechatCpnRepo
GenerateMixRepo mixrepos.GenerateMixRepo
}
func NewCmb(
bc *conf.Bootstrap,
rdb *data.Rdb,
orderRepo repo.OrderRepo,
ProductRepo repo.ProductRepo,
MqSendMixRepo mixrepos.MQSendMixRepo,
WechatCpnRepo wechatrepo.WechatCpnRepo,
GenerateMixRepo mixrepos.GenerateMixRepo,
) *Cmb {
return &Cmb{
bc: bc,
rdb: rdb,
OrderRepo: orderRepo,
ProductRepo: ProductRepo,
MqSendMixRepo: MqSendMixRepo,
WechatCpnRepo: WechatCpnRepo,
GenerateMixRepo: GenerateMixRepo,
}

34
internal/biz/cmb/consume.go
View File

@ -2,47 +2,13 @@ package cmb
import (
"context"
"fmt"
"go.opentelemetry.io/otel/trace"
"voucher/internal/biz/bo"
"voucher/internal/pkg/mq"
)
func (v *Cmb) PushOrderMQ(ctx context.Context, orderNo string) error {
eventMap := v.bc.RocketMQ.EventMap["order"]
sendOption := []mq.SendOption{
mq.WithSendShardingKeysOption(fmt.Sprintf("%s", orderNo)),
mq.WithOpenTelemetryOption(trace.SpanFromContext(ctx).SpanContext().TraceID().String()),
}
if err := v.MqSendMixRepo.SendSync(ctx, eventMap.Topic, []byte("{}"), sendOption...); err != nil {
return fmt.Errorf("收单成功，消费队列投递失败[%v]", err)
}
return nil
}
func (v *Cmb) OrderConsume(ctx context.Context, order *bo.OrderBo) (err error) {
return
}
func (v *Cmb) PushQueryDelayMQ(ctx context.Context, orderNo string) error {
eventMap := v.bc.RocketMQ.EventMap["query"]
sendOption := []mq.SendOption{
mq.WithSendShardingKeysOption(orderNo),
mq.WithOpenTelemetryOption(trace.SpanFromContext(ctx).SpanContext().TraceID().String()),
mq.WithSendDelayLevelOption(300),
}
if err := v.MqSendMixRepo.SendSync(ctx, eventMap.Topic, []byte("{}"), sendOption...); err != nil {
return fmt.Errorf("查询入队投递失败[%v]", err)
}
return nil
}
func (v *Cmb) QueryConsume(ctx context.Context, order *bo.OrderBo) (err error) {
return
}

3
internal/biz/cmb/voucher.go
View File

@ -19,6 +19,7 @@ func (v *Cmb) Order(ctx context.Context, req *bo.OrderCreateReqBo) (orderNo stri
func (v *Cmb) order(ctx context.Context, req *bo.OrderCreateReqBo) (orderNo string, err error) {
order, err := v.OrderRepo.GetByOutBizNo(ctx, req.OutBizNo)
if err != nil && !errors.Is(err, gorm.ErrRecordNotFound) {
return
}
@ -56,8 +57,6 @@ func (v *Cmb) order(ctx context.Context, req *bo.OrderCreateReqBo) (orderNo stri
return
}
err = v.PushOrderMQ(ctx, orderNo)
return
}

43
internal/biz/voucher.go
View File

@ -1,25 +1,68 @@
package biz
import (
"context"
"fmt"
"go.opentelemetry.io/otel/trace"
"voucher/internal/biz/cmb"
"voucher/internal/biz/mixrepos"
"voucher/internal/biz/repo"
"voucher/internal/conf"
"voucher/internal/data"
"voucher/internal/pkg/mq"
)
type VoucherBiz struct {
bc *conf.Bootstrap
rdb *data.Rdb
Cmb *cmb.Cmb
OrderRepo repo.OrderRepo
MqSendMixRepo mixrepos.MQSendMixRepo
}
func NewVoucherBiz(
bc *conf.Bootstrap,
rdb *data.Rdb,
Cmb *cmb.Cmb,
OrderRepo repo.OrderRepo,
MqSendMixRepo mixrepos.MQSendMixRepo,
) *VoucherBiz {
return &VoucherBiz{
bc: bc,
rdb: rdb,
Cmb: Cmb,
OrderRepo: OrderRepo,
MqSendMixRepo: MqSendMixRepo,
}
}
func (v *VoucherBiz) PushOrderMQ(ctx context.Context, orderNo string) error {
eventMap := v.bc.RocketMQ.EventMap["order"]
sendOption := []mq.SendOption{
mq.WithSendShardingKeysOption(fmt.Sprintf("%s", orderNo)),
mq.WithOpenTelemetryOption(trace.SpanFromContext(ctx).SpanContext().TraceID().String()),
}
if err := v.MqSendMixRepo.SendSync(ctx, eventMap.Topic, []byte("{}"), sendOption...); err != nil {
return fmt.Errorf("收单成功，消费队列投递失败[%v]", err)
}
return nil
}
func (v *VoucherBiz) PushQueryDelayMQ(ctx context.Context, orderNo string) error {
eventMap := v.bc.RocketMQ.EventMap["query"]
sendOption := []mq.SendOption{
mq.WithSendShardingKeysOption(orderNo),
mq.WithOpenTelemetryOption(trace.SpanFromContext(ctx).SpanContext().TraceID().String()),
mq.WithSendDelayLevelOption(300),
}
if err := v.MqSendMixRepo.SendSync(ctx, eventMap.Topic, []byte("{}"), sendOption...); err != nil {
return fmt.Errorf("查询入队投递失败[%v]", err)
}
return nil
}