allied-creation / performance · 提交

转到一个项目

作者 郑周
提交 02c8b63b0dca407e60f1047c4237878340682358 2 个父辈 52f18dfb ff61d49b

Merge remote-tracking branch 'origin/master' 

# Conflicts:
#	pkg/application/factory/reposetory.go
内嵌 并排对比
正在显示 12 个修改的文件 包含 420 行增加 和 35 行删除
  1 +{
  2 + // 使用 IntelliSense 了解相关属性。
  3 + // 悬停以查看现有属性的描述。
  4 + // 欲了解更多信息，请访问: https://go.microsoft.com/fwlink/?linkid=830387
  5 + "version": "0.2.0",
  6 + "configurations": [
  7 +
  8 + {
  9 + "name": "Launch file",
  10 + "type": "go",
  11 + "request": "launch",
  12 + "mode": "debug",
  13 + "buildFlags": "--tags=local",
  14 + "program": "./main.go"
  15 + },
  16 + ]
  17 +}
1 package company 1 package company
2 2
3 import ( 3 import (
  4 + "encoding/json"
4 "time" 5 "time"
5 6
6 "github.com/linmadan/egglib-go/core/application" 7 "github.com/linmadan/egglib-go/core/application"
7 "gitlab.fjmaimaimai.com/allied-creation/performance/pkg/application/company/command" 8 "gitlab.fjmaimaimai.com/allied-creation/performance/pkg/application/company/command"
8 "gitlab.fjmaimaimai.com/allied-creation/performance/pkg/application/factory" 9 "gitlab.fjmaimaimai.com/allied-creation/performance/pkg/application/factory"
9 "gitlab.fjmaimaimai.com/allied-creation/performance/pkg/domain" 10 "gitlab.fjmaimaimai.com/allied-creation/performance/pkg/domain"
  11 + "gitlab.fjmaimaimai.com/allied-creation/performance/pkg/log"
10 ) 12 )
11 13
12 -type CompanyServices struct { 14 +type SyncDataCompanyService struct {
13 } 15 }
14 16
  17 +// type BusinessAdminCommand struct {
  18 +// //company:公司
  19 +// Module string `json:"module"`
  20 +// // add:添加,edit:编辑,setCompanyCharge：更改公司主管,changeAdmin换管理员
  21 +// Action string `json:"action"`
  22 +// // 具体的对象JSON数据
  23 +// Datas json.RawMessage `json:"data"`
  24 +// }
  25 +
15 //从BusinessAdmins 接收消息，变更公司数据 26 //从BusinessAdmins 接收消息，变更公司数据
16 -//  
17 -func (c CompanyServices) BusinessAdminCompany() error {  
18 - return nil 27 +func (c SyncDataCompanyService) FromBusinessAdmin(param *domain.MessageBody) error {
  28 + action := param.Module + "/" + param.Action
  29 + var err error
  30 + switch action {
  31 + case "company/add":
  32 + var param1 command.SaveCompanyCommand
  33 + err = json.Unmarshal(param.Datas, &param1)
  34 + if err != nil {
  35 + return application.ThrowError(application.INTERNAL_SERVER_ERROR, err.Error())
  36 + }
  37 + err = c.addCompany(&param1)
  38 + case "company/edit":
  39 + var param2 command.SaveCompanyCommand
  40 + err = json.Unmarshal(param.Datas, &param2)
  41 + if err != nil {
  42 + return application.ThrowError(application.INTERNAL_SERVER_ERROR, err.Error())
  43 + }
  44 + err = c.editCompany(&param2)
  45 + case "company/setCompanyCharge":
  46 + var param3 command.SetCompanyCharge
  47 + err = json.Unmarshal(param.Datas, &param3)
  48 + if err != nil {
  49 + return application.ThrowError(application.INTERNAL_SERVER_ERROR, err.Error())
  50 + }
  51 + err = c.setCompanyCharge(&param3)
  52 + case "company/changeAdmin":
  53 + var param3 command.ChangeAdminCommand
  54 + err = json.Unmarshal(param.Datas, &param3)
  55 + if err != nil {
  56 + return application.ThrowError(application.INTERNAL_SERVER_ERROR, err.Error())
  57 + }
  58 + err = c.changeAdmin(&param3)
  59 + default:
  60 + log.Logger.Error("action err:" + action)
  61 + }
  62 +
  63 + return err
19 } 64 }
20 65
21 //addCompany 66 //addCompany
22 //从BusinessAdmins 接收消息 添加公司 67 //从BusinessAdmins 接收消息 添加公司
23 -func (c CompanyServices) addCompany(param *command.SaveCompanyCommand) error { 68 +//module="company" action="add"
  69 +func (c SyncDataCompanyService) addCompany(param *command.SaveCompanyCommand) error {
24 transactionContext, err := factory.CreateTransactionContext(nil) 70 transactionContext, err := factory.CreateTransactionContext(nil)
25 if err != nil { 71 if err != nil {
26 return application.ThrowError(application.TRANSACTION_ERROR, err.Error()) 72 return application.ThrowError(application.TRANSACTION_ERROR, err.Error())
@@ -63,11 +109,11 @@ func (c CompanyServices) addCompany(param *command.SaveCompanyCommand) error { @@ -63,11 +109,11 @@ func (c CompanyServices) addCompany(param *command.SaveCompanyCommand) error {
63 }) 109 })
64 _, err = companyRepo.Insert(&newCompany) 110 _, err = companyRepo.Insert(&newCompany)
65 if err != nil { 111 if err != nil {
66 - return err 112 + return application.ThrowError(application.INTERNAL_SERVER_ERROR, err.Error())
67 } 113 }
68 _, err = userRepo.Insert(&newUser) 114 _, err = userRepo.Insert(&newUser)
69 if err != nil { 115 if err != nil {
70 - return err 116 + return application.ThrowError(application.INTERNAL_SERVER_ERROR, err.Error())
71 } 117 }
72 if err := transactionContext.CommitTransaction(); err != nil { 118 if err := transactionContext.CommitTransaction(); err != nil {
73 return application.ThrowError(application.TRANSACTION_ERROR, err.Error()) 119 return application.ThrowError(application.TRANSACTION_ERROR, err.Error())
@@ -77,7 +123,8 @@ func (c CompanyServices) addCompany(param *command.SaveCompanyCommand) error { @@ -77,7 +123,8 @@ func (c CompanyServices) addCompany(param *command.SaveCompanyCommand) error {
77 123
78 //editCompany 124 //editCompany
79 //从BusinessAdmins 接收消息 更新公司 125 //从BusinessAdmins 接收消息 更新公司
80 -func (c CompanyServices) editCompany(param *command.SaveCompanyCommand) error { 126 +//module="company" action="edit"
  127 +func (c SyncDataCompanyService) editCompany(param *command.SaveCompanyCommand) error {
81 transactionContext, err := factory.CreateTransactionContext(nil) 128 transactionContext, err := factory.CreateTransactionContext(nil)
82 if err != nil { 129 if err != nil {
83 return application.ThrowError(application.TRANSACTION_ERROR, err.Error()) 130 return application.ThrowError(application.TRANSACTION_ERROR, err.Error())
@@ -100,14 +147,14 @@ func (c CompanyServices) editCompany(param *command.SaveCompanyCommand) error { @@ -100,14 +147,14 @@ func (c CompanyServices) editCompany(param *command.SaveCompanyCommand) error {
100 "id": param.Comapany.Id, 147 "id": param.Comapany.Id,
101 }) 148 })
102 if err != nil { 149 if err != nil {
103 - return err 150 + return application.ThrowError(application.INTERNAL_SERVER_ERROR, err.Error())
104 } 151 }
105 _, userList, err := userRepo.Find(map[string]interface{}{ 152 _, userList, err := userRepo.Find(map[string]interface{}{
106 "limit": 1, 153 "limit": 1,
107 "id": param.User.Id, 154 "id": param.User.Id,
108 }) 155 })
109 if err != nil { 156 if err != nil {
110 - return err 157 + return application.ThrowError(application.INTERNAL_SERVER_ERROR, err.Error())
111 } 158 }
112 var ( 159 var (
113 newCompany *domain.Company 160 newCompany *domain.Company
@@ -146,32 +193,33 @@ func (c CompanyServices) editCompany(param *command.SaveCompanyCommand) error { @@ -146,32 +193,33 @@ func (c CompanyServices) editCompany(param *command.SaveCompanyCommand) error {
146 if len(companyList) > 0 { 193 if len(companyList) > 0 {
147 _, err = companyRepo.Update(newCompany) 194 _, err = companyRepo.Update(newCompany)
148 if err != nil { 195 if err != nil {
149 - return err 196 + return application.ThrowError(application.INTERNAL_SERVER_ERROR, err.Error())
150 } 197 }
151 } else { 198 } else {
152 _, err = companyRepo.Insert(newCompany) 199 _, err = companyRepo.Insert(newCompany)
153 if err != nil { 200 if err != nil {
154 - return err 201 + return application.ThrowError(application.INTERNAL_SERVER_ERROR, err.Error())
155 } 202 }
156 } 203 }
157 if len(userList) > 0 { 204 if len(userList) > 0 {
158 _, err = userRepo.Update(newUser) 205 _, err = userRepo.Update(newUser)
159 if err != nil { 206 if err != nil {
160 - return err 207 + return application.ThrowError(application.INTERNAL_SERVER_ERROR, err.Error())
161 } 208 }
162 } else { 209 } else {
163 _, err = userRepo.Insert(newUser) 210 _, err = userRepo.Insert(newUser)
164 if err != nil { 211 if err != nil {
165 - return err 212 + return application.ThrowError(application.INTERNAL_SERVER_ERROR, err.Error())
166 } 213 }
167 } 214 }
168 if err := transactionContext.CommitTransaction(); err != nil { 215 if err := transactionContext.CommitTransaction(); err != nil {
169 - return err 216 + return application.ThrowError(application.INTERNAL_SERVER_ERROR, err.Error())
170 } 217 }
171 return nil 218 return nil
172 } 219 }
173 220
174 -func (srv CompanyServices) setCompanyCharge(param *command.SetCompanyCharge) error { 221 +//module="company" action="setCompanyCharge"
  222 +func (srv SyncDataCompanyService) setCompanyCharge(param *command.SetCompanyCharge) error {
175 transactionContext, err := factory.CreateTransactionContext(nil) 223 transactionContext, err := factory.CreateTransactionContext(nil)
176 if err != nil { 224 if err != nil {
177 return application.ThrowError(application.TRANSACTION_ERROR, err.Error()) 225 return application.ThrowError(application.TRANSACTION_ERROR, err.Error())
@@ -193,14 +241,14 @@ func (srv CompanyServices) setCompanyCharge(param *command.SetCompanyCharge) err @@ -193,14 +241,14 @@ func (srv CompanyServices) setCompanyCharge(param *command.SetCompanyCharge) err
193 }) 241 })
194 242
195 if err != nil { 243 if err != nil {
196 - return err 244 + return application.ThrowError(application.INTERNAL_SERVER_ERROR, err.Error())
197 } 245 }
198 for i := range companyList { 246 for i := range companyList {
199 companyList[i].ChargeUserIds = param.ChargeUserIds 247 companyList[i].ChargeUserIds = param.ChargeUserIds
200 companyList[i].UpdateAt = time.Now() 248 companyList[i].UpdateAt = time.Now()
201 _, err = companyRepo.Update(companyList[i]) 249 _, err = companyRepo.Update(companyList[i])
202 if err != nil { 250 if err != nil {
203 - return err 251 + return application.ThrowError(application.INTERNAL_SERVER_ERROR, err.Error())
204 } 252 }
205 } 253 }
206 if err := transactionContext.CommitTransaction(); err != nil { 254 if err := transactionContext.CommitTransaction(); err != nil {
@@ -211,7 +259,8 @@ func (srv CompanyServices) setCompanyCharge(param *command.SetCompanyCharge) err @@ -211,7 +259,8 @@ func (srv CompanyServices) setCompanyCharge(param *command.SetCompanyCharge) err
211 259
212 //changeAdmin 260 //changeAdmin
213 //从BusinessAdmins 接收消息 变更主管 261 //从BusinessAdmins 接收消息 变更主管
214 -func (srv CompanyServices) changeAdmin(param *command.ChangeAdminCommand) error { 262 +//module="company" action="changeAdmin"
  263 +func (srv SyncDataCompanyService) changeAdmin(param *command.ChangeAdminCommand) error {
215 transactionContext, err := factory.CreateTransactionContext(nil) 264 transactionContext, err := factory.CreateTransactionContext(nil)
216 if err != nil { 265 if err != nil {
217 return application.ThrowError(application.TRANSACTION_ERROR, err.Error()) 266 return application.ThrowError(application.TRANSACTION_ERROR, err.Error())
@@ -233,7 +282,7 @@ func (srv CompanyServices) changeAdmin(param *command.ChangeAdminCommand) error @@ -233,7 +282,7 @@ func (srv CompanyServices) changeAdmin(param *command.ChangeAdminCommand) error
233 "adminType": domain.UserTypeManager, 282 "adminType": domain.UserTypeManager,
234 }) 283 })
235 if err != nil { 284 if err != nil {
236 - return err 285 + return application.ThrowError(application.INTERNAL_SERVER_ERROR, err.Error())
237 } 286 }
238 //修改旧管理员 为普通用户 287 //修改旧管理员 为普通用户
239 for i := range userList { 288 for i := range userList {
@@ -241,7 +290,7 @@ func (srv CompanyServices) changeAdmin(param *command.ChangeAdminCommand) error @@ -241,7 +290,7 @@ func (srv CompanyServices) changeAdmin(param *command.ChangeAdminCommand) error
241 userList[i].UpdateAt = time.Now() 290 userList[i].UpdateAt = time.Now()
242 _, err := userRepo.Update(userList[i]) 291 _, err := userRepo.Update(userList[i])
243 if err != nil { 292 if err != nil {
244 - return err 293 + return application.ThrowError(application.INTERNAL_SERVER_ERROR, err.Error())
245 } 294 }
246 } 295 }
247 //获取新管理员 296 //获取新管理员
@@ -251,7 +300,7 @@ func (srv CompanyServices) changeAdmin(param *command.ChangeAdminCommand) error @@ -251,7 +300,7 @@ func (srv CompanyServices) changeAdmin(param *command.ChangeAdminCommand) error
251 "account": param.UserAccount, 300 "account": param.UserAccount,
252 }) 301 })
253 if err != nil { 302 if err != nil {
254 - return err 303 + return application.ThrowError(application.INTERNAL_SERVER_ERROR, err.Error())
255 } 304 }
256 //修改为管理员用户 305 //修改为管理员用户
257 for i := range userList2 { 306 for i := range userList2 {
@@ -259,7 +308,7 @@ func (srv CompanyServices) changeAdmin(param *command.ChangeAdminCommand) error @@ -259,7 +308,7 @@ func (srv CompanyServices) changeAdmin(param *command.ChangeAdminCommand) error
259 userList[i].UpdateAt = time.Now() 308 userList[i].UpdateAt = time.Now()
260 _, err := userRepo.Update(userList[i]) 309 _, err := userRepo.Update(userList[i])
261 if err != nil { 310 if err != nil {
262 - return err 311 + return application.ThrowError(application.INTERNAL_SERVER_ERROR, err.Error())
263 } 312 }
264 } 313 }
265 if err := transactionContext.CommitTransaction(); err != nil { 314 if err := transactionContext.CommitTransaction(); err != nil {
1 package department 1 package department
2 2
3 import ( 3 import (
  4 + "encoding/json"
4 "time" 5 "time"
5 6
6 "github.com/linmadan/egglib-go/core/application" 7 "github.com/linmadan/egglib-go/core/application"
@@ -11,10 +12,55 @@ import ( @@ -11,10 +12,55 @@ import (
11 12
12 type SyncDataDepartmentService struct{} 13 type SyncDataDepartmentService struct{}
13 14
  15 +// type BusinessAdminCommand struct {
  16 +// // department:部门
  17 +// Module string `json:"module"`
  18 +// // add:添加,edit:编辑,batchDelete:批量删除,import:导入部门
  19 +// Action string `json:"action"`
  20 +// // 具体的对象JSON数据
  21 +// Datas json.RawMessage `json:"data"`
  22 +// }
  23 +
  24 +func (srv SyncDataDepartmentService) FromBusinessAdmin(param *domain.MessageBody) error {
  25 + action := param.Module + "/" + param.Action
  26 + var err error
  27 + switch action {
  28 + case "department/add":
  29 + var param1 command.AddDepartmentCommand
  30 + err = json.Unmarshal(param.Datas, &param1)
  31 + if err != nil {
  32 + return application.ThrowError(application.INTERNAL_SERVER_ERROR, err.Error())
  33 + }
  34 + err = srv.addDepartment(&param1)
  35 + case "department/edit":
  36 + var param1 command.EditDepartmentCommand
  37 + err = json.Unmarshal(param.Datas, &param1)
  38 + if err != nil {
  39 + return application.ThrowError(application.INTERNAL_SERVER_ERROR, err.Error())
  40 + }
  41 + err = srv.editDepartment(&param1)
  42 + case "department/batchDelete":
  43 + var param1 command.BatchDeleteCommand
  44 + err = json.Unmarshal(param.Datas, &param1)
  45 + if err != nil {
  46 + return application.ThrowError(application.INTERNAL_SERVER_ERROR, err.Error())
  47 + }
  48 + err = srv.batchDeleteDepartment(&param1)
  49 + case "department/import":
  50 + var param1 []command.ImportDepartmentCommand
  51 + err = json.Unmarshal(param.Datas, &param1)
  52 + if err != nil {
  53 + return application.ThrowError(application.INTERNAL_SERVER_ERROR, err.Error())
  54 + }
  55 + err = srv.importDepartment(param1)
  56 + }
  57 + return err
  58 +}
  59 +
14 //AddDepartment 60 //AddDepartment
15 //从BusinessAdmins 接收消息 添加部门 61 //从BusinessAdmins 接收消息 添加部门
16 //module="department" action="add" 62 //module="department" action="add"
17 -func (srv SyncDataDepartmentService) addDepartment(param command.AddDepartmentCommand) error { 63 +func (srv SyncDataDepartmentService) addDepartment(param *command.AddDepartmentCommand) error {
18 transactionContext, err := factory.CreateTransactionContext(nil) 64 transactionContext, err := factory.CreateTransactionContext(nil)
19 if err != nil { 65 if err != nil {
20 return application.ThrowError(application.TRANSACTION_ERROR, err.Error()) 66 return application.ThrowError(application.TRANSACTION_ERROR, err.Error())
@@ -56,7 +102,7 @@ func (srv SyncDataDepartmentService) addDepartment(param command.AddDepartmentCo @@ -56,7 +102,7 @@ func (srv SyncDataDepartmentService) addDepartment(param command.AddDepartmentCo
56 //EditDepartment 102 //EditDepartment
57 //从BusinessAdmins 接收消息 编辑部门 103 //从BusinessAdmins 接收消息 编辑部门
58 //module="department" action="edit" 104 //module="department" action="edit"
59 -func (srv SyncDataDepartmentService) editDepartment(param command.EditDepartmentCommand) error { 105 +func (srv SyncDataDepartmentService) editDepartment(param *command.EditDepartmentCommand) error {
60 transactionContext, err := factory.CreateTransactionContext(nil) 106 transactionContext, err := factory.CreateTransactionContext(nil)
61 if err != nil { 107 if err != nil {
62 return application.ThrowError(application.TRANSACTION_ERROR, err.Error()) 108 return application.ThrowError(application.TRANSACTION_ERROR, err.Error())
@@ -115,7 +161,7 @@ func (srv SyncDataDepartmentService) editDepartment(param command.EditDepartment @@ -115,7 +161,7 @@ func (srv SyncDataDepartmentService) editDepartment(param command.EditDepartment
115 //batchDelete 161 //batchDelete
116 //从BusinessAdmins 接收消息 删除部门 162 //从BusinessAdmins 接收消息 删除部门
117 //module="department" action="batchDelete" 163 //module="department" action="batchDelete"
118 -func (srv SyncDataDepartmentService) batchDeleteDepartment(param command.BatchDeleteCommand) error { 164 +func (srv SyncDataDepartmentService) batchDeleteDepartment(param *command.BatchDeleteCommand) error {
119 if len(param.Ids) == 0 { 165 if len(param.Ids) == 0 {
120 return nil 166 return nil
121 } 167 }
@@ -72,3 +72,11 @@ func CreateRoleUserRepository(options map[string]interface{}) domain.RoleUserRep @@ -72,3 +72,11 @@ func CreateRoleUserRepository(options map[string]interface{}) domain.RoleUserRep
72 } 72 }
73 return repository.NewRoleUserRepository(transactionContext) 73 return repository.NewRoleUserRepository(transactionContext)
74 } 74 }
  75 +
  76 +func CreateReceivedMessageRepository(options map[string]interface{}) domain.ReceivedMessageRepository {
  77 + var transactionContext *pg.TransactionContext
  78 + if value, ok := options["transactionContext"]; ok {
  79 + transactionContext = value.(*pg.TransactionContext)
  80 + }
  81 + return repository.NewReceivedMessageRepository(transactionContext)
  82 +}
1 package user 1 package user
2 2
3 import ( 3 import (
  4 + "encoding/json"
4 "time" 5 "time"
5 6
  7 + "gitlab.fjmaimaimai.com/allied-creation/performance/pkg/log"
  8 +
6 "github.com/linmadan/egglib-go/core/application" 9 "github.com/linmadan/egglib-go/core/application"
7 "gitlab.fjmaimaimai.com/allied-creation/performance/pkg/application/factory" 10 "gitlab.fjmaimaimai.com/allied-creation/performance/pkg/application/factory"
8 "gitlab.fjmaimaimai.com/allied-creation/performance/pkg/application/user/command" 11 "gitlab.fjmaimaimai.com/allied-creation/performance/pkg/application/user/command"
@@ -11,10 +14,65 @@ import ( @@ -11,10 +14,65 @@ import (
11 14
12 type SyncDataUserService struct{} 15 type SyncDataUserService struct{}
13 16
  17 +// type BusinessAdminCommand struct {
  18 +// // employee:员工
  19 +// Module string `json:"module"`
  20 +// // add:添加,edit:编辑,batchDelete:批量删除,batchForbid:批量禁用用户，batchRemove：批量更改用户部门,import:导入用户
  21 +// Action string `json:"action"`
  22 +// // 具体的对象JSON数据
  23 +// Datas json.RawMessage `json:"data"`
  24 +// }
  25 +
  26 +func (srv SyncDataUserService) FromBusinessAdmin(param *domain.MessageBody) error {
  27 + action := param.Module + "/" + param.Action
  28 + var err error
  29 + switch action {
  30 + case "employee/add":
  31 + var param1 command.SaveUserCommand
  32 + err = json.Unmarshal(param.Datas, &param1)
  33 + if err != nil {
  34 + return application.ThrowError(application.INTERNAL_SERVER_ERROR, err.Error())
  35 + }
  36 + err = srv.AddUser(&param1)
  37 + case "employee/edit":
  38 + var param2 command.SaveUserCommand
  39 + err = json.Unmarshal(param.Datas, &param2)
  40 + if err != nil {
  41 + return application.ThrowError(application.INTERNAL_SERVER_ERROR, err.Error())
  42 + }
  43 + err = srv.UpdateUser(&param2)
  44 + case "employee/batchDelete":
  45 + var param3 command.BatchDeleteCommand
  46 + err = json.Unmarshal(param.Datas, &param3)
  47 + if err != nil {
  48 + return application.ThrowError(application.INTERNAL_SERVER_ERROR, err.Error())
  49 + }
  50 + err = srv.batchDelete(&param3)
  51 + case "company/batchForbid":
  52 + var param4 command.BatchForbidCommand
  53 + err = json.Unmarshal(param.Datas, &param4)
  54 + if err != nil {
  55 + return application.ThrowError(application.INTERNAL_SERVER_ERROR, err.Error())
  56 + }
  57 + err = srv.batchForbid(&param4)
  58 + case "company/import":
  59 + var param4 command.ImportUserCommand
  60 + err = json.Unmarshal(param.Datas, &param4)
  61 + if err != nil {
  62 + return application.ThrowError(application.INTERNAL_SERVER_ERROR, err.Error())
  63 + }
  64 + err = srv.importUser(&param4)
  65 + default:
  66 + log.Logger.Error("action err:" + action)
  67 + }
  68 +
  69 + return err
  70 +}
  71 +
14 //AddUser 72 //AddUser
15 //从BusinessAdmins 接收消息 添加用户 73 //从BusinessAdmins 接收消息 添加用户
16 //module="employee" action="add" 74 //module="employee" action="add"
17 -func (srv SyncDataUserService) AddUser(param command.SaveUserCommand) error { 75 +func (srv SyncDataUserService) AddUser(param *command.SaveUserCommand) error {
18 transactionContext, err := factory.CreateTransactionContext(nil) 76 transactionContext, err := factory.CreateTransactionContext(nil)
19 if err != nil { 77 if err != nil {
20 return application.ThrowError(application.TRANSACTION_ERROR, err.Error()) 78 return application.ThrowError(application.TRANSACTION_ERROR, err.Error())
@@ -59,7 +117,7 @@ func (srv SyncDataUserService) AddUser(param command.SaveUserCommand) error { @@ -59,7 +117,7 @@ func (srv SyncDataUserService) AddUser(param command.SaveUserCommand) error {
59 //UpdateUser 117 //UpdateUser
60 //从BusinessAdmins 接收消息 更新用户 118 //从BusinessAdmins 接收消息 更新用户
61 //module="employee" action="edit" 119 //module="employee" action="edit"
62 -func (srv SyncDataUserService) UpdateUser(param command.SaveUserCommand) error { 120 +func (srv SyncDataUserService) UpdateUser(param *command.SaveUserCommand) error {
63 transactionContext, err := factory.CreateTransactionContext(nil) 121 transactionContext, err := factory.CreateTransactionContext(nil)
64 if err != nil { 122 if err != nil {
65 return application.ThrowError(application.TRANSACTION_ERROR, err.Error()) 123 return application.ThrowError(application.TRANSACTION_ERROR, err.Error())
@@ -119,7 +177,7 @@ func (srv SyncDataUserService) UpdateUser(param command.SaveUserCommand) error { @@ -119,7 +177,7 @@ func (srv SyncDataUserService) UpdateUser(param command.SaveUserCommand) error {
119 //batchDelete 177 //batchDelete
120 //从BusinessAdmins 接收消息 删除用户 178 //从BusinessAdmins 接收消息 删除用户
121 //module="employee" action="batchDelete" 179 //module="employee" action="batchDelete"
122 -func (srv SyncDataUserService) batchDelete(param command.BatchDeleteCommand) error { 180 +func (srv SyncDataUserService) batchDelete(param *command.BatchDeleteCommand) error {
123 if len(param.Uids) == 0 { 181 if len(param.Uids) == 0 {
124 return nil 182 return nil
125 } 183 }
@@ -150,7 +208,7 @@ func (srv SyncDataUserService) batchDelete(param command.BatchDeleteCommand) err @@ -150,7 +208,7 @@ func (srv SyncDataUserService) batchDelete(param command.BatchDeleteCommand) err
150 //batchForbid 208 //batchForbid
151 //从BusinessAdmins 接收消息 禁用,启用用户 209 //从BusinessAdmins 接收消息 禁用,启用用户
152 //module="employee" action="batchForbid" 210 //module="employee" action="batchForbid"
153 -func (srv SyncDataUserService) batchForbid(param command.BatchForbidCommand) error { 211 +func (srv SyncDataUserService) batchForbid(param *command.BatchForbidCommand) error {
154 if len(param.Uids) == 0 { 212 if len(param.Uids) == 0 {
155 return nil 213 return nil
156 } 214 }
@@ -190,7 +248,7 @@ func (srv SyncDataUserService) batchForbid(param command.BatchForbidCommand) err @@ -190,7 +248,7 @@ func (srv SyncDataUserService) batchForbid(param command.BatchForbidCommand) err
190 //importUser 248 //importUser
191 //从BusinessAdmins 接收消息 导入用户数据 249 //从BusinessAdmins 接收消息 导入用户数据
192 //module="employee" action="import" 250 //module="employee" action="import"
193 -func (srv SyncDataUserService) importUser(param command.ImportUserCommand) error { 251 +func (srv SyncDataUserService) importUser(param *command.ImportUserCommand) error {
194 transactionContext, err := factory.CreateTransactionContext(nil) 252 transactionContext, err := factory.CreateTransactionContext(nil)
195 if err != nil { 253 if err != nil {
196 return application.ThrowError(application.TRANSACTION_ERROR, err.Error()) 254 return application.ThrowError(application.TRANSACTION_ERROR, err.Error())
  1 +package domain
  2 +
  3 +import (
  4 + "encoding/json"
  5 + "time"
  6 +)
  7 +
  8 +type ReceivedMessage struct {
  9 + MessageId int64 `json:"MessageId"`
  10 + MessageType string `json:"MessageType"`
  11 + MessageBody string `json:"MessageBody"`
  12 + OccurredOn time.Time `json:"OccurredOn"`
  13 + CreateAt time.Time `json:"-"`
  14 +}
  15 +
  16 +type MessageBody struct {
  17 + Module string `json:"module"`
  18 + Action string `json:"action"`
  19 + Datas json.RawMessage `json:"datas"` // 具体的对象JSON数据
  20 +}
  21 +
  22 +type ReceivedMessageRepository interface {
  23 + SaveMessage(param *ReceivedMessage) error
  24 + FindMessage(id int64) (*ReceivedMessage, error)
  25 +}
@@ -6,11 +6,12 @@ import ( @@ -6,11 +6,12 @@ import (
6 "github.com/linmadan/egglib-go/persistent/pg/comment" 6 "github.com/linmadan/egglib-go/persistent/pg/comment"
7 "gitlab.fjmaimaimai.com/allied-creation/performance/pkg/infrastructure/pg/models" 7 "gitlab.fjmaimaimai.com/allied-creation/performance/pkg/infrastructure/pg/models"
8 8
9 - "gitlab.fjmaimaimai.com/allied-creation/performance/pkg/log"  
10 -  
11 "github.com/go-pg/pg/v10" 9 "github.com/go-pg/pg/v10"
  10 + "github.com/go-pg/pg/v10/orm"
12 "github.com/linmadan/egglib-go/persistent/pg/hooks" 11 "github.com/linmadan/egglib-go/persistent/pg/hooks"
13 "gitlab.fjmaimaimai.com/allied-creation/performance/pkg/constant" 12 "gitlab.fjmaimaimai.com/allied-creation/performance/pkg/constant"
  13 + "gitlab.fjmaimaimai.com/allied-creation/performance/pkg/infrastructure/pg/models"
  14 + "gitlab.fjmaimaimai.com/allied-creation/performance/pkg/log"
14 ) 15 )
15 16
16 var DB *pg.DB 17 var DB *pg.DB
@@ -27,6 +28,24 @@ func init() { @@ -27,6 +28,24 @@ func init() {
27 Logger: log.Logger, 28 Logger: log.Logger,
28 }) 29 })
29 } 30 }
  31 + if !constant.DISABLE_CREATE_TABLE {
  32 + tables := []interface{}{
  33 + &models.Company{},
  34 + &models.Department{},
  35 + &models.User{},
  36 + &models.ReceivedMessage{},
  37 + }
  38 + for _, model := range tables {
  39 + err := DB.Model(model).CreateTable(&orm.CreateTableOptions{
  40 + Temp: false,
  41 + IfNotExists: true,
  42 + FKConstraints: true,
  43 + })
  44 + if err != nil {
  45 + panic(err)
  46 + }
  47 + }
  48 + }
30 49
31 models := []interface{}{ 50 models := []interface{}{
32 &models.User{}, 51 &models.User{},
  1 +package models
  2 +
  3 +import "time"
  4 +
  5 +type ReceivedMessage struct {
  6 + tableName struct{} `pg:"received_message"`
  7 + MessageId int64 `pg:"pk:message_id"`
  8 + MessageType string
  9 + MessageBody string
  10 + OccurredOn time.Time
  11 + CreateAt time.Time
  12 +}
  1 +package repository
  2 +
  3 +import (
  4 + "time"
  5 +
  6 + "github.com/go-pg/pg/v10"
  7 + pgTransaction "github.com/linmadan/egglib-go/transaction/pg"
  8 + "gitlab.fjmaimaimai.com/allied-creation/performance/pkg/domain"
  9 + "gitlab.fjmaimaimai.com/allied-creation/performance/pkg/infrastructure/pg/models"
  10 +)
  11 +
  12 +type ReceivedMessageRepository struct {
  13 + transactionContext *pgTransaction.TransactionContext
  14 +}
  15 +
  16 +var _ domain.ReceivedMessageRepository = (*ReceivedMessageRepository)(nil)
  17 +
  18 +func NewReceivedMessageRepository(tx *pgTransaction.TransactionContext) *ReceivedMessageRepository {
  19 + return &ReceivedMessageRepository{
  20 + transactionContext: tx,
  21 + }
  22 +}
  23 +
  24 +func (repo *ReceivedMessageRepository) SaveMessage(param *domain.ReceivedMessage) error {
  25 + message := &models.ReceivedMessage{
  26 + MessageId: param.MessageId,
  27 + MessageType: param.MessageType,
  28 + MessageBody: param.MessageBody,
  29 + OccurredOn: param.OccurredOn,
  30 + CreateAt: time.Now(),
  31 + }
  32 + tx := repo.transactionContext.PgTx
  33 + _, err := tx.Model(message).
  34 + Insert()
  35 + return err
  36 +}
  37 +
  38 +func (repo *ReceivedMessageRepository) FindMessage(messageId int64) (*domain.ReceivedMessage, error) {
  39 + tx := repo.transactionContext.PgTx
  40 + receivedMessageModel := new(models.ReceivedMessage)
  41 + query := tx.Model(receivedMessageModel).
  42 + Where("message_id = ?", messageId)
  43 + if err := query.First(); err != nil {
  44 + if err != pg.ErrNoRows {
  45 + return nil, err
  46 + }
  47 + }
  48 + message := &domain.ReceivedMessage{
  49 + MessageId: receivedMessageModel.MessageId,
  50 + MessageType: receivedMessageModel.MessageType,
  51 + MessageBody: receivedMessageModel.MessageBody,
  52 + OccurredOn: receivedMessageModel.OccurredOn,
  53 + CreateAt: receivedMessageModel.CreateAt,
  54 + }
  55 + return message, nil
  56 +
  57 +}
  1 +package handle
  2 +
  3 +import (
  4 + "encoding/json"
  5 + "strconv"
  6 +
  7 + "github.com/Shopify/sarama"
  8 + "gitlab.fjmaimaimai.com/allied-creation/performance/pkg/application/company"
  9 + "gitlab.fjmaimaimai.com/allied-creation/performance/pkg/application/department"
  10 + "gitlab.fjmaimaimai.com/allied-creation/performance/pkg/application/factory"
  11 + "gitlab.fjmaimaimai.com/allied-creation/performance/pkg/application/user"
  12 + "gitlab.fjmaimaimai.com/allied-creation/performance/pkg/domain"
  13 + "gitlab.fjmaimaimai.com/allied-creation/performance/pkg/log"
  14 +)
  15 +
  16 +func SyncDataBusinessAdmin(msgData *sarama.ConsumerMessage) error {
  17 + var receivedMsg domain.ReceivedMessage
  18 +
  19 + log.Logger.Debug(string(msgData.Value))
  20 + err := json.Unmarshal(msgData.Value, &receivedMsg)
  21 + if err != nil {
  22 + log.Logger.Error("解析ReceivedMessage 失败" + err.Error())
  23 + return err
  24 + }
  25 + transactionContext, err := factory.CreateTransactionContext(nil)
  26 + if err != nil {
  27 + return err
  28 + }
  29 + _ = transactionContext.StartTransaction()
  30 +
  31 + defer func() {
  32 +
  33 + _ = transactionContext.RollbackTransaction()
  34 +
  35 + }()
  36 +
  37 + msgRepo := factory.CreateReceivedMessageRepository(map[string]interface{}{
  38 + "transactionContext": transactionContext,
  39 + })
  40 + oldMsg, err := msgRepo.FindMessage(receivedMsg.MessageId)
  41 + if err != nil {
  42 + log.Logger.Error(" 查询旧消息发生错误 " + err.Error())
  43 + log.Logger.Info("异常消息message_id=" + strconv.FormatInt(receivedMsg.MessageId, 10))
  44 + return err
  45 + }
  46 + if oldMsg.MessageId == receivedMsg.MessageId {
  47 + log.Logger.Info("消息重复，message_id=" + strconv.FormatInt(receivedMsg.MessageId, 10))
  48 + return nil
  49 + }
  50 + var msgBody domain.MessageBody
  51 + err = json.Unmarshal([]byte(receivedMsg.MessageBody), &msgBody)
  52 + if err != nil {
  53 + log.Logger.Error("解析messageBody发生错误" + err.Error())
  54 + return err
  55 + }
  56 +
  57 + switch msgBody.Module {
  58 + case "company":
  59 + companySrv := company.SyncDataCompanyService{}
  60 + err = companySrv.FromBusinessAdmin(&msgBody)
  61 + if err != nil {
  62 + log.Logger.Error("处理company消息失败"+err.Error(), map[string]interface{}{
  63 + "data": msgBody,
  64 + })
  65 + return err
  66 + }
  67 + case "department":
  68 + departmentSrv := department.SyncDataDepartmentService{}
  69 + err = departmentSrv.FromBusinessAdmin(&msgBody)
  70 + if err != nil {
  71 + log.Logger.Error("处理department消息失败"+err.Error(), map[string]interface{}{
  72 + "data": msgBody,
  73 + })
  74 + return err
  75 + }
  76 + case "employee":
  77 + employeeSrv := user.SyncDataUserService{}
  78 + err = employeeSrv.FromBusinessAdmin(&msgBody)
  79 + if err != nil {
  80 + log.Logger.Error("处理employee消息失败"+err.Error(), map[string]interface{}{
  81 + "data": msgBody,
  82 + })
  83 + return err
  84 + }
  85 + }
  86 + err = msgRepo.SaveMessage(&receivedMsg)
  87 + if err != nil {
  88 + log.Logger.Error(" 保存新消息发生错误 " + err.Error())
  89 + log.Logger.Info("异常消息message_id=" + strconv.FormatInt(receivedMsg.MessageId, 10))
  90 + }
  91 + _ = transactionContext.CommitTransaction()
  92 + return nil
  93 +}
@@ -5,12 +5,14 @@ import ( @@ -5,12 +5,14 @@ import (
5 saramaConsumer "github.com/linmadan/egglib-go/mom/kafka/sarama" 5 saramaConsumer "github.com/linmadan/egglib-go/mom/kafka/sarama"
6 "gitlab.fjmaimaimai.com/allied-creation/performance/pkg/constant" 6 "gitlab.fjmaimaimai.com/allied-creation/performance/pkg/constant"
7 "gitlab.fjmaimaimai.com/allied-creation/performance/pkg/log" 7 "gitlab.fjmaimaimai.com/allied-creation/performance/pkg/log"
  8 + "gitlab.fjmaimaimai.com/allied-creation/performance/pkg/port/consumer/handle"
8 ) 9 )
9 10
10 func Run() { 11 func Run() {
11 messageHandlerMap := make(map[string]func(message *sarama.ConsumerMessage) error) 12 messageHandlerMap := make(map[string]func(message *sarama.ConsumerMessage) error)
12 messageHandlerMap["demo-v1"] = Demo 13 messageHandlerMap["demo-v1"] = Demo
13 - 14 + //"指定topic" => 对应的处理方法
  15 + messageHandlerMap["mmm-business-admin-dev"] = handle.SyncDataBusinessAdmin
14 err := saramaConsumer.StartConsume(constant.KAFKA_HOSTS, constant.SERVICE_NAME, messageHandlerMap, log.Logger) 16 err := saramaConsumer.StartConsume(constant.KAFKA_HOSTS, constant.SERVICE_NAME, messageHandlerMap, log.Logger)
15 log.Logger.Error(err.Error()) 17 log.Logger.Error(err.Error())
16 } 18 }