さくらのクラウドの「シンプルMQ」で始めるイベント駆動アプリケーション

git clone https://github.com/thara/sakura-simplemq-sample.git
cd sakura-simplemq-sample

export SAKURACLOUD_ACCESS_TOKEN="your-access-token"
export SAKURACLOUD_ACCESS_TOKEN_SECRET="your-access-token-secret"

cd terraform
terraform init
terraform apply
cd ..

export SIMPLEMQ_API_KEY="your-simplemq-api-key"

go run cmd/consumer-worker/main.go -queue=playground-mq

go run cmd/producer/main.go -queue=playground-mq -message="Hello from CLI!"

# サーバーを起動
go run cmd/producer-api-server/main.go -addr=:8080 -queue=playground-mq

# 別ターミナルからHTTP POSTで送信
curl -X POST http://localhost:8080 -d "Hello from API Server!"

edition = "2023";

option go_package = "github.com/thara/sakura-simplemq-sample/samplepb";

message Notification {
  string message = 1;
}

func encodeProtoMessage(msg proto.Message) (string, error) {
    // 1. Protocol Buffersでバイナリ化
    b, err := proto.Marshal(msg)
    if err != nil {
        return "", fmt.Errorf("failed to marshal proto message: %w", err)
    }

    // 2. Deflate圧縮
    compressed, err := compress(b)
    if err != nil {
        return "", fmt.Errorf("failed to compress proto message: %w", err)
    }

    // 3. Base64エンコード
    return base64.StdEncoding.EncodeToString(compressed), nil
}

func decodeProtoMessage(src string, msg proto.Message) error {
    // 1. Base64デコード
    data, err := base64.StdEncoding.DecodeString(src)
    if err != nil {
        return fmt.Errorf("failed to decode base64 string: %w", err)
    }

    // 2. Deflate解凍
    decompressed, err := decompress(data)
    if err != nil {
        return fmt.Errorf("failed to decompress proto message: %w", err)
    }

    // 3. Protocol Buffersでデシリアライズ
    if err := proto.Unmarshal(decompressed, msg); err != nil {
        return fmt.Errorf("failed to unmarshal proto message: %w", err)
    }
    return nil
}

notification := &samplepb.Notification{
    Message: proto.String(message),
}
if err := internal.SendNotification(ctx, messageOp, notification); err != nil {
    return fmt.Errorf("failed to send notification: %v", err)
}

handler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
    body, err := io.ReadAll(r.Body)
    if err != nil {
        http.Error(w, "failed to read request body", http.StatusBadRequest)
        return
    }

    notification := &samplepb.Notification{
        Message: proto.String(string(body)),
    }
    if err := internal.SendNotification(r.Context(), messageOp, notification); err != nil {
        http.Error(w, fmt.Sprintf("failed to send notification: %w", err), http.StatusInternalServerError)
        return
    }
})

func SendNotification(ctx context.Context, messageOp simplemq.MessageAPI, notification *samplepb.Notification) error {
    content, err := encodeProtoMessage(notification)
    if err != nil {
        return fmt.Errorf("failed to encode message: %v", err)
    }

    resSend, err := messageOp.Send(ctx, content)
    if err != nil {
        return fmt.Errorf("failed to send message: %v", err)
    }

    slog.Info("Message Sent", slog.String("ID", string(resSend.ID)))
    return nil
}

ticker := time.NewTicker(1 * time.Second)
defer ticker.Stop()

for {
    select {
    case <-ctx.Done():
        return ctx.Err()
    case <-ticker.C:
        if err := receiveMessages(ctx, client, queueName); err != nil {
            slog.ErrorContext(ctx, "failed to receive messages", slog.Any("error", err))
        }
    }
}

func ReceiveNotifications(ctx context.Context, messageOp simplemq.MessageAPI) ([]*samplepb.Notification, error) {
    // 1. シンプルMQからメッセージを受信
    messages, err := messageOp.Receive(ctx)
    if err != nil {
        return nil, fmt.Errorf("failed to receive messages: %v", err)
    }
    slog.Info("messages received", slog.Int("count", len(messages)))

    var notifications []*samplepb.Notification

    // 2. 各メッセージを処理
    for _, msg := range messages {
        if err := func() error {
            // 2-1. デコード
            var notification samplepb.Notification
            if err := decodeProtoMessage(string(msg.Content), &notification); err != nil {
                return fmt.Errorf("failed to decode message ID %s: %v", msg.ID, err)
            }
            slog.Info("message received", slog.String("id", string(msg.ID)), slog.String("content", notification.GetMessage()))

            notifications = append(notifications, &notification)

            // 2-2. メッセージを削除してacknowledge
            if err := messageOp.Delete(ctx, string(msg.ID)); err != nil {
                return fmt.Errorf("failed to delete message ID %s: %v", msg.ID, err)
            }
            slog.Info("message deleted", slog.String("id", string(msg.ID)))

            return nil
        }(); err != nil {
            slog.Error("failed to process message", slog.String("id", string(msg.ID)), slog.Any("error", err))
        }
    }
    return notifications, nil
}

# ターミナル1
go run cmd/consumer-worker/main.go -queue=playground-mq

# ターミナル2
go run cmd/consumer-worker/main.go -queue=playground-mq

# ターミナル3（さらに追加）
go run cmd/consumer-worker/main.go -queue=playground-mq