business_admin.go
3.5 KB
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
108
109
110
package handle
import (
"encoding/json"
"gitlab.fjmaimaimai.com/allied-creation/performance/pkg/application/position/service"
"strconv"
"github.com/Shopify/sarama"
"gitlab.fjmaimaimai.com/allied-creation/performance/pkg/application/company"
"gitlab.fjmaimaimai.com/allied-creation/performance/pkg/application/department"
"gitlab.fjmaimaimai.com/allied-creation/performance/pkg/application/factory"
"gitlab.fjmaimaimai.com/allied-creation/performance/pkg/application/user"
"gitlab.fjmaimaimai.com/allied-creation/performance/pkg/domain"
"gitlab.fjmaimaimai.com/allied-creation/performance/pkg/log"
)
func SyncDataBusinessAdmin(msgData *sarama.ConsumerMessage) error {
var receivedMsg domain.ReceivedMessage
err := json.Unmarshal(msgData.Value, &receivedMsg)
if err != nil {
log.Logger.Error("解析ReceivedMessage 失败" + err.Error())
return err
}
log.Logger.Debug("解析message:", map[string]interface{}{"data": receivedMsg})
transactionContext, err := factory.CreateTransactionContext(nil)
if err != nil {
return err
}
_ = transactionContext.StartTransaction()
defer func() {
_ = transactionContext.RollbackTransaction()
}()
msgRepo := factory.CreateReceivedMessageRepository(map[string]interface{}{
"transactionContext": transactionContext,
})
oldMsg, err := msgRepo.FindMessage(receivedMsg.MessageId)
if err != nil {
log.Logger.Error(" 查询旧消息发生错误 " + err.Error())
log.Logger.Info("异常消息message_id=" + strconv.FormatInt(receivedMsg.MessageId, 10))
return err
}
if oldMsg.MessageId == receivedMsg.MessageId {
log.Logger.Info("消息重复，message_id=" + strconv.FormatInt(receivedMsg.MessageId, 10))
return nil
}
var msgBody domain.MessageBody
err = json.Unmarshal([]byte(receivedMsg.MessageBody), &msgBody)
if err != nil {
log.Logger.Error("解析messageBody发生错误" + err.Error())
return err
}
switch msgBody.Module {
case "company":
companySrv := company.SyncDataCompanyService{}
err = companySrv.FromBusinessAdmin(&msgBody)
if err != nil {
log.Logger.Error("处理company消息失败"+err.Error(), map[string]interface{}{
"module": msgBody.Module,
"action": msgBody.Action,
"data": string(msgBody.Data),
})
return err
}
case "department":
departmentSrv := department.SyncDataDepartmentService{}
err = departmentSrv.FromBusinessAdmin(&msgBody)
if err != nil {
log.Logger.Error("处理department消息失败"+err.Error(), map[string]interface{}{
"module": msgBody.Module,
"action": msgBody.Action,
"data": string(msgBody.Data),
})
return err
}
case "employee":
employeeSrv := user.SyncDataUserService{}
err = employeeSrv.FromBusinessAdmin(&msgBody)
if err != nil {
log.Logger.Error("处理employee消息失败"+err.Error(), map[string]interface{}{
"module": msgBody.Module,
"action": msgBody.Action,
"data": string(msgBody.Data),
})
return err
}
case "position":
positionSrv := service.SyncDataPositionService{}
err = positionSrv.FromBusinessAdmin(&msgBody)
if err != nil {
log.Logger.Error("处理position消息失败"+err.Error(), map[string]interface{}{
"module": msgBody.Module,
"action": msgBody.Action,
"data": string(msgBody.Data),
})
return err
}
}
err = msgRepo.SaveMessage(&receivedMsg)
if err != nil {
log.Logger.Error(" 保存新消息发生错误 " + err.Error())
log.Logger.Info("异常消息message_id=" + strconv.FormatInt(receivedMsg.MessageId, 10))
}
_ = transactionContext.CommitTransaction()
return nil
}