107 lines
3.9 KiB
Go
107 lines
3.9 KiB
Go
// Package roommq defines RocketMQ payloads owned by room-service.
|
|
package roommq
|
|
|
|
import (
|
|
"encoding/json"
|
|
"errors"
|
|
|
|
"google.golang.org/protobuf/proto"
|
|
roomeventsv1 "hyapp.local/api/proto/events/room/v1"
|
|
)
|
|
|
|
const (
|
|
MessageTypeRoomOutboxEvent = "room_outbox_event"
|
|
MessageTypeRoomRocketLaunchDue = "room_rocket_launch_due"
|
|
|
|
TagRoomOutboxEvent = "room_outbox_event"
|
|
TagRoomRocketLaunchDue = "room_rocket_launch_due"
|
|
)
|
|
|
|
// RoomOutboxMessage is the MQ representation of one durable room_outbox fact.
|
|
type RoomOutboxMessage struct {
|
|
MessageType string `json:"message_type"`
|
|
AppCode string `json:"app_code"`
|
|
EventID string `json:"event_id"`
|
|
EventType string `json:"event_type"`
|
|
RoomID string `json:"room_id"`
|
|
RoomVersion int64 `json:"room_version"`
|
|
OccurredAtMS int64 `json:"occurred_at_ms"`
|
|
Envelope []byte `json:"envelope"`
|
|
}
|
|
|
|
// RoomRocketLaunchDueMessage wakes room-service at the configured launch_at_ms.
|
|
type RoomRocketLaunchDueMessage struct {
|
|
MessageType string `json:"message_type"`
|
|
AppCode string `json:"app_code"`
|
|
RoomID string `json:"room_id"`
|
|
RocketID string `json:"rocket_id"`
|
|
Level int32 `json:"level"`
|
|
LaunchAtMS int64 `json:"launch_at_ms"`
|
|
ResetAtMS int64 `json:"reset_at_ms"`
|
|
CommandID string `json:"command_id"`
|
|
}
|
|
|
|
// EncodeRoomOutboxMessage serializes the protobuf envelope into an MQ JSON body.
|
|
func EncodeRoomOutboxMessage(envelope *roomeventsv1.EventEnvelope) ([]byte, error) {
|
|
if envelope == nil {
|
|
return nil, errors.New("room outbox envelope is required")
|
|
}
|
|
envelopeBytes, err := proto.Marshal(envelope)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return json.Marshal(RoomOutboxMessage{
|
|
MessageType: MessageTypeRoomOutboxEvent,
|
|
AppCode: envelope.GetAppCode(),
|
|
EventID: envelope.GetEventId(),
|
|
EventType: envelope.GetEventType(),
|
|
RoomID: envelope.GetRoomId(),
|
|
RoomVersion: envelope.GetRoomVersion(),
|
|
OccurredAtMS: envelope.GetOccurredAtMs(),
|
|
Envelope: envelopeBytes,
|
|
})
|
|
}
|
|
|
|
// DecodeRoomOutboxMessage validates the MQ body and restores the protobuf fact.
|
|
func DecodeRoomOutboxMessage(body []byte) (*roomeventsv1.EventEnvelope, RoomOutboxMessage, error) {
|
|
var message RoomOutboxMessage
|
|
if err := json.Unmarshal(body, &message); err != nil {
|
|
return nil, RoomOutboxMessage{}, err
|
|
}
|
|
if message.MessageType != MessageTypeRoomOutboxEvent {
|
|
return nil, RoomOutboxMessage{}, errors.New("unexpected room outbox message_type")
|
|
}
|
|
var envelope roomeventsv1.EventEnvelope
|
|
if err := proto.Unmarshal(message.Envelope, &envelope); err != nil {
|
|
return nil, RoomOutboxMessage{}, err
|
|
}
|
|
if envelope.GetEventId() == "" || envelope.GetEventType() == "" || envelope.GetRoomId() == "" {
|
|
return nil, RoomOutboxMessage{}, errors.New("room outbox envelope is incomplete")
|
|
}
|
|
return &envelope, message, nil
|
|
}
|
|
|
|
// EncodeRoomRocketLaunchDueMessage serializes a launch wakeup command.
|
|
func EncodeRoomRocketLaunchDueMessage(message RoomRocketLaunchDueMessage) ([]byte, error) {
|
|
message.MessageType = MessageTypeRoomRocketLaunchDue
|
|
if message.AppCode == "" || message.RoomID == "" || message.RocketID == "" || message.Level <= 0 || message.LaunchAtMS <= 0 {
|
|
return nil, errors.New("room rocket launch due message is incomplete")
|
|
}
|
|
return json.Marshal(message)
|
|
}
|
|
|
|
// DecodeRoomRocketLaunchDueMessage validates a launch wakeup body.
|
|
func DecodeRoomRocketLaunchDueMessage(body []byte) (RoomRocketLaunchDueMessage, error) {
|
|
var message RoomRocketLaunchDueMessage
|
|
if err := json.Unmarshal(body, &message); err != nil {
|
|
return RoomRocketLaunchDueMessage{}, err
|
|
}
|
|
if message.MessageType != MessageTypeRoomRocketLaunchDue {
|
|
return RoomRocketLaunchDueMessage{}, errors.New("unexpected room rocket message_type")
|
|
}
|
|
if message.AppCode == "" || message.RoomID == "" || message.RocketID == "" || message.Level <= 0 || message.LaunchAtMS <= 0 {
|
|
return RoomRocketLaunchDueMessage{}, errors.New("room rocket launch due message is incomplete")
|
|
}
|
|
return message, nil
|
|
}
|