更新

@@ -14,17 +14,17 @@ import (
 type SyncDataCompanyService struct {
 type SyncDataCompanyService struct {
 }
 }
 
 
-type BusinessAdminCommand struct {
-	//company:公司
-	Module string `json:"module"`
-	// add:添加,edit:编辑,setCompanyCharge：更改公司主管,changeAdmin换管理员
-	Action string `json:"action"`
-	// 具体的对象JSON数据
-	Datas json.RawMessage `json:"data"`
-}
+// type BusinessAdminCommand struct {
+// 	//company:公司
+// 	Module string `json:"module"`
+// 	// add:添加,edit:编辑,setCompanyCharge：更改公司主管,changeAdmin换管理员
+// 	Action string `json:"action"`
+// 	// 具体的对象JSON数据
+// 	Datas json.RawMessage `json:"data"`
+// }
 
 
 //从BusinessAdmins 接收消息，变更公司数据
 //从BusinessAdmins 接收消息，变更公司数据
-func (c SyncDataCompanyService) FromBusinessAdminCompany(param *BusinessAdminCommand) error {
+func (c SyncDataCompanyService) FromBusinessAdmin(param *domain.MessageBody) error {
 	action := param.Module + "/" + param.Action
 	action := param.Module + "/" + param.Action
 	var err error
 	var err error
 	switch action {
 	switch action {

@@ -12,16 +12,16 @@ import (
 
 
 type SyncDataDepartmentService struct{}
 type SyncDataDepartmentService struct{}
 
 
-type BusinessAdminCommand struct {
-	// department:部门
-	Module string `json:"module"`
-	// add:添加,edit:编辑,batchDelete:批量删除,import:导入部门
-	Action string `json:"action"`
-	// 具体的对象JSON数据
-	Datas json.RawMessage `json:"data"`
-}
+// type BusinessAdminCommand struct {
+// 	// department:部门
+// 	Module string `json:"module"`
+// 	// add:添加,edit:编辑,batchDelete:批量删除,import:导入部门
+// 	Action string `json:"action"`
+// 	// 具体的对象JSON数据
+// 	Datas json.RawMessage `json:"data"`
+// }
 
 
-func (srv SyncDataDepartmentService) FromBusinessAdminCompany(param BusinessAdminCommand) error {
+func (srv SyncDataDepartmentService) FromBusinessAdmin(param *domain.MessageBody) error {
 	action := param.Module + "/" + param.Action
 	action := param.Module + "/" + param.Action
 	var err error
 	var err error
 	switch action {
 	switch action {

@@ -35,3 +35,11 @@ func CreateDepartmentRepository(options map[string]interface{}) domain.Departmen
 	}
 	}
 	return repository.NewDepartmentRepository(transactionContext)
 	return repository.NewDepartmentRepository(transactionContext)
 }
 }
+
+func CreateReceivedMessageRepository(options map[string]interface{}) domain.ReceivedMessageRepository {
+	var transactionContext *pg.TransactionContext
+	if value, ok := options["transactionContext"]; ok {
+		transactionContext = value.(*pg.TransactionContext)
+	}
+	return repository.NewReceivedMessageRepository(transactionContext)
+}

@@ -14,16 +14,16 @@ import (
 
 
 type SyncDataUserService struct{}
 type SyncDataUserService struct{}
 
 
-type BusinessAdminCommand struct {
-	// employee:员工
-	Module string `json:"module"`
-	// add:添加,edit:编辑,batchDelete:批量删除,batchForbid:批量禁用用户，batchRemove：批量更改用户部门,import:导入用户
-	Action string `json:"action"`
-	// 具体的对象JSON数据
-	Datas json.RawMessage `json:"data"`
-}
+// type BusinessAdminCommand struct {
+// 	// employee:员工
+// 	Module string `json:"module"`
+// 	// add:添加,edit:编辑,batchDelete:批量删除,batchForbid:批量禁用用户，batchRemove：批量更改用户部门,import:导入用户
+// 	Action string `json:"action"`
+// 	// 具体的对象JSON数据
+// 	Datas json.RawMessage `json:"data"`
+// }
 
 
-func (srv SyncDataUserService) FromBusinessAdminCompany(param *BusinessAdminCommand) error {
+func (srv SyncDataUserService) FromBusinessAdmin(param *domain.MessageBody) error {
 	action := param.Module + "/" + param.Action
 	action := param.Module + "/" + param.Action
 	var err error
 	var err error
 	switch action {
 	switch action {

 package domain
 package domain
 
 
-import "time"
+import (
+	"encoding/json"
+	"time"
+)
 
 
 type ReceivedMessage struct {
 type ReceivedMessage struct {
 	MessageId   int64
 	MessageId   int64
@@ -10,6 +13,12 @@ type ReceivedMessage struct {
 	CreateAt    time.Time
 	CreateAt    time.Time
 }
 }
 
 
+type MessageBody struct {
+	Module string          `json:"module"`
+	Action string          `json:"action"`
+	Datas  json.RawMessage `json:"data"` // 具体的对象JSON数据
+}
+
 type ReceivedMessageRepository interface {
 type ReceivedMessageRepository interface {
 	SaveMessage(param *ReceivedMessage) error
 	SaveMessage(param *ReceivedMessage) error
 	FindMessage(id int64) (*ReceivedMessage, error)
 	FindMessage(id int64) (*ReceivedMessage, error)

@@ -30,6 +30,7 @@ func init() {
 			&models.Company{},
 			&models.Company{},
 			&models.Department{},
 			&models.Department{},
 			&models.User{},
 			&models.User{},
+			&models.ReceivedMessage{},
 		}
 		}
 		for _, model := range tables {
 		for _, model := range tables {
 			err := DB.Model(model).CreateTable(&orm.CreateTableOptions{
 			err := DB.Model(model).CreateTable(&orm.CreateTableOptions{

@@ -3,6 +3,7 @@ package repository
 import (
 import (
 	"time"
 	"time"
 
 
+	"github.com/go-pg/pg/v10"
 	pgTransaction "github.com/linmadan/egglib-go/transaction/pg"
 	pgTransaction "github.com/linmadan/egglib-go/transaction/pg"
 	"gitlab.fjmaimaimai.com/allied-creation/performance/pkg/domain"
 	"gitlab.fjmaimaimai.com/allied-creation/performance/pkg/domain"
 	"gitlab.fjmaimaimai.com/allied-creation/performance/pkg/infrastructure/pg/models"
 	"gitlab.fjmaimaimai.com/allied-creation/performance/pkg/infrastructure/pg/models"
@@ -14,6 +15,12 @@ type ReceivedMessageRepository struct {
 
 
 var _ domain.ReceivedMessageRepository = (*ReceivedMessageRepository)(nil)
 var _ domain.ReceivedMessageRepository = (*ReceivedMessageRepository)(nil)
 
 
+func NewReceivedMessageRepository(tx *pgTransaction.TransactionContext) *ReceivedMessageRepository {
+	return &ReceivedMessageRepository{
+		transactionContext: tx,
+	}
+}
+
 func (repo *ReceivedMessageRepository) SaveMessage(param *domain.ReceivedMessage) error {
 func (repo *ReceivedMessageRepository) SaveMessage(param *domain.ReceivedMessage) error {
 	message := &models.ReceivedMessage{
 	message := &models.ReceivedMessage{
 		MessageId:   param.MessageId,
 		MessageId:   param.MessageId,
@@ -31,9 +38,10 @@ func (repo *ReceivedMessageRepository) SaveMessage(param *domain.ReceivedMessage
 func (repo *ReceivedMessageRepository) FindMessage(messageId int64) (*domain.ReceivedMessage, error) {
 func (repo *ReceivedMessageRepository) FindMessage(messageId int64) (*domain.ReceivedMessage, error) {
 	tx := repo.transactionContext.PgTx
 	tx := repo.transactionContext.PgTx
 	receivedMessageModel := new(models.ReceivedMessage)
 	receivedMessageModel := new(models.ReceivedMessage)
-	query := tx.Model(receivedMessageModel).Where("message.id = ?", messageId)
+	query := tx.Model(receivedMessageModel).
+		Where("message_id = ?", messageId)
 	if err := query.First(); err != nil {
 	if err := query.First(); err != nil {
-		if err.Error() != "pg: no rows in result set" {
+		if err != pg.ErrNoRows {
 			return nil, err
 			return nil, err
 		}
 		}
 	}
 	}

+package handle
+
+import (
+	"encoding/json"
+	"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 {
+		return err
+	}
+	transactionContext, err := factory.CreateTransactionContext(nil)
+	if err != nil {
+		return err
+	}
+	transactionContext.StartTransaction()
+	var transactionIsSucceed bool
+	defer func() {
+		if transactionIsSucceed {
+			transactionContext.CommitTransaction()
+		} else {
+			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{}{
+				"data": msgBody,
+			})
+			return err
+		}
+	case "department":
+		departmentSrv := department.SyncDataDepartmentService{}
+		err = departmentSrv.FromBusinessAdmin(&msgBody)
+		if err != nil {
+			log.Logger.Error("处理department消息失败"+err.Error(), map[string]interface{}{
+				"data": msgBody,
+			})
+			return err
+		}
+	case "employee":
+		employeeSrv := user.SyncDataUserService{}
+		err = employeeSrv.FromBusinessAdmin(&msgBody)
+		if err != nil {
+			log.Logger.Error("处理employee消息失败"+err.Error(), map[string]interface{}{
+				"data": msgBody,
+			})
+			return err
+		}
+	}
+	err = msgRepo.SaveMessage(&receivedMsg)
+	if err != nil {
+		log.Logger.Error(" 保存新消息发生错误  " + err.Error())
+		log.Logger.Info("异常消息message_id=" + strconv.FormatInt(receivedMsg.MessageId, 10))
+	}
+	return nil
+}

-package handle

@@ -5,12 +5,14 @@ import (
 	saramaConsumer "github.com/linmadan/egglib-go/mom/kafka/sarama"
 	saramaConsumer "github.com/linmadan/egglib-go/mom/kafka/sarama"
 	"gitlab.fjmaimaimai.com/allied-creation/performance/pkg/constant"
 	"gitlab.fjmaimaimai.com/allied-creation/performance/pkg/constant"
 	"gitlab.fjmaimaimai.com/allied-creation/performance/pkg/log"
 	"gitlab.fjmaimaimai.com/allied-creation/performance/pkg/log"
+	"gitlab.fjmaimaimai.com/allied-creation/performance/pkg/port/consumer/handle"
 )
 )
 
 
 func Run() {
 func Run() {
 	messageHandlerMap := make(map[string]func(message *sarama.ConsumerMessage) error)
 	messageHandlerMap := make(map[string]func(message *sarama.ConsumerMessage) error)
 	messageHandlerMap["demo-v1"] = Demo
 	messageHandlerMap["demo-v1"] = Demo
-
+	//"指定topic" => 对应的处理方法
+	messageHandlerMap["mmm-business-admin-dev"] = handle.SyncDataBusinessAdmin
 	err := saramaConsumer.StartConsume(constant.KAFKA_HOSTS, constant.SERVICE_NAME, messageHandlerMap, log.Logger)
 	err := saramaConsumer.StartConsume(constant.KAFKA_HOSTS, constant.SERVICE_NAME, messageHandlerMap, log.Logger)
 	log.Logger.Error(err.Error())
 	log.Logger.Error(err.Error())
 }
 }