67 lines
2.1 KiB
Go
67 lines
2.1 KiB
Go
// Package usermq defines RocketMQ payloads owned by user-service.
|
|
package usermq
|
|
|
|
import (
|
|
"encoding/json"
|
|
"errors"
|
|
"strings"
|
|
)
|
|
|
|
const (
|
|
MessageTypeUserOutboxEvent = "user_outbox_event"
|
|
|
|
TagUserOutboxEvent = "user_outbox_event"
|
|
)
|
|
|
|
// UserOutboxMessage is the MQ representation of one committed user_outbox fact.
|
|
type UserOutboxMessage struct {
|
|
MessageType string `json:"message_type"`
|
|
AppCode string `json:"app_code"`
|
|
EventID string `json:"event_id"`
|
|
EventType string `json:"event_type"`
|
|
AggregateType string `json:"aggregate_type"`
|
|
AggregateID int64 `json:"aggregate_id"`
|
|
PayloadJSON string `json:"payload_json"`
|
|
OccurredAtMS int64 `json:"occurred_at_ms"`
|
|
}
|
|
|
|
// EncodeUserOutboxMessage serializes a user fact for fanout through RocketMQ.
|
|
func EncodeUserOutboxMessage(message UserOutboxMessage) ([]byte, error) {
|
|
message.MessageType = MessageTypeUserOutboxEvent
|
|
if strings.TrimSpace(message.AppCode) == "" ||
|
|
strings.TrimSpace(message.EventID) == "" ||
|
|
strings.TrimSpace(message.EventType) == "" ||
|
|
strings.TrimSpace(message.AggregateType) == "" ||
|
|
message.AggregateID <= 0 ||
|
|
message.OccurredAtMS <= 0 {
|
|
return nil, errors.New("user outbox message is incomplete")
|
|
}
|
|
if strings.TrimSpace(message.PayloadJSON) == "" {
|
|
message.PayloadJSON = "{}"
|
|
}
|
|
return json.Marshal(message)
|
|
}
|
|
|
|
// DecodeUserOutboxMessage validates the MQ body and restores the user fact.
|
|
func DecodeUserOutboxMessage(body []byte) (UserOutboxMessage, error) {
|
|
var message UserOutboxMessage
|
|
if err := json.Unmarshal(body, &message); err != nil {
|
|
return UserOutboxMessage{}, err
|
|
}
|
|
if message.MessageType != MessageTypeUserOutboxEvent {
|
|
return UserOutboxMessage{}, errors.New("unexpected user outbox message_type")
|
|
}
|
|
if strings.TrimSpace(message.AppCode) == "" ||
|
|
strings.TrimSpace(message.EventID) == "" ||
|
|
strings.TrimSpace(message.EventType) == "" ||
|
|
strings.TrimSpace(message.AggregateType) == "" ||
|
|
message.AggregateID <= 0 ||
|
|
message.OccurredAtMS <= 0 {
|
|
return UserOutboxMessage{}, errors.New("user outbox message is incomplete")
|
|
}
|
|
if strings.TrimSpace(message.PayloadJSON) == "" {
|
|
message.PayloadJSON = "{}"
|
|
}
|
|
return message, nil
|
|
}
|