andwati/telecom-platform
1
0
Fork 0
mirror of https://github.com/andwati/telecom-platform.git synced 2026-10-20 03:44:25 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: Add RabbitMQ message queue integration for profile event publishing in carrier-connector

Browse Source 
- Add github.com/rabbitmq/amqp091-go v1.11.0 dependency to go.mod
- Add messageQueue parameter to OrderProfileHandlerWithRepo and DeleteProfileHandlerWithRepo
- Add PublishProfileEvent calls for profile.downloaded and profile.deleted events with goroutines
- Add message queue initialization in main.go with RABBITMQ_URL environment variable
- Add profile-events queue declaration with error handling and fallback
- Add messageQueue parameter to setupRoutes
This commit is contained in:
nutcas3
2026-04-25 06:50:13 +03:00
parent 0ea496b627
commit 6b2a38b51b
5 changed files with 52 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/carrier-connector/go.mod
+1
View File
@@ -6,6 +6,7 @@ require (
github.com/gin-gonic/gin v1.12.0
github.com/joho/godotenv v1.5.1
github.com/prometheus/client_golang v1.20.5
github.com/rabbitmq/amqp091-go v1.11.0
github.com/rs/zerolog v1.35.0
gorm.io/driver/postgres v1.6.0
gorm.io/gorm v1.31.1
apps/carrier-connector/go.sum
+4
View File
@@ -85,6 +85,8 @@ github.com/quic-go/qpack v0.6.0 h1:g7W+BMYynC1LbYLSqRt8PBg5Tgwxn214ZZR34VIOjz8=
github.com/quic-go/qpack v0.6.0/go.mod h1:lUpLKChi8njB4ty2bFLX2x4gzDqXwUpaO1DP9qMDZII=
github.com/quic-go/quic-go v0.59.0 h1:OLJkp1Mlm/aS7dpKgTc6cnpynnD2Xg7C1pwL6vy/SAw=
github.com/quic-go/quic-go v0.59.0/go.mod h1:upnsH4Ju1YkqpLXC305eW3yDZ4NfnNbmQRCMWS58IKU=
github.com/rabbitmq/amqp091-go v1.11.0 h1:HxIctVm9Gid/Vtn706necmZ7Wj6pgGI2eqplRbEY8O8=
github.com/rabbitmq/amqp091-go v1.11.0/go.mod h1:Hy4jKW5kQART1u+JkDTF9YYOQUHXqMuhrgxOEeS7G4o=
github.com/rs/zerolog v1.35.0 h1:VD0ykx7HMiMJytqINBsKcbLS+BJ4WYjz+05us+LRTdI=
github.com/rs/zerolog v1.35.0/go.mod h1:EjML9kdfa/RMA7h/6z6pYmq1ykOuA8/mjWaEvGI+jcw=
github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
@@ -105,6 +107,8 @@ github.com/ugorji/go/codec v1.3.1 h1:waO7eEiFDwidsBN6agj1vJQ4AG7lh2yqXyOXqhgQuyY
github.com/ugorji/go/codec v1.3.1/go.mod h1:pRBVtBSKl77K30Bv8R2P+cLSGaTtex6fsA2Wjqmfxj4=
go.mongodb.org/mongo-driver/v2 v2.5.0 h1:yXUhImUjjAInNcpTcAlPHiT7bIXhshCTL3jVBkF3xaE=
go.mongodb.org/mongo-driver/v2 v2.5.0/go.mod h1:yOI9kBsufol30iFsl1slpdq1I0eHPzybRWdyYUs8K/0=
go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE=
go.uber.org/mock v0.6.0 h1:hyF9dfmbgIX5EfOdasqLsWD6xqpNZlXblLB/Dbnwv3Y=
go.uber.org/mock v0.6.0/go.mod h1:KiVJ4BqZJaMj4svdfmHM0AUx4NJYO8ZNpPnZn1Z+BBU=
golang.org/x/arch v0.22.0 h1:c/Zle32i5ttqRXjdLyyHZESLD/bB90DCU1g9l/0YBDI=
apps/carrier-connector/internal/handler/profile_handlers.go
+20 -2
View File
@@ -10,12 +10,13 @@ import (
"github.com/gin-gonic/gin"
"github.com/nutcas3/telecom-platform/apps/carrier-connector/internal/es2"
"github.com/nutcas3/telecom-platform/apps/carrier-connector/internal/mq"
"github.com/nutcas3/telecom-platform/apps/carrier-connector/internal/repository"
"github.com/nutcas3/telecom-platform/apps/carrier-connector/internal/webhook"
)
// OrderProfileHandlerWithRepo orders a profile via ES2+ and persists it in the repo.
func OrderProfileHandlerWithRepo(client *es2.ES2Client, repo repository.ProfileRepository, webhookClient *webhook.WebhookClient) gin.HandlerFunc {
func OrderProfileHandlerWithRepo(client *es2.ES2Client, repo repository.ProfileRepository, webhookClient *webhook.WebhookClient, messageQueue *mq.MessageQueue) gin.HandlerFunc {
return func(c *gin.Context) {
var order ProfileOrder
if err := c.ShouldBindJSON(&order); err != nil {
@@ -78,6 +79,15 @@ func OrderProfileHandlerWithRepo(client *es2.ES2Client, repo repository.ProfileR
})
}
// Publish message queue event for profile download
if messageQueue != nil {
go messageQueue.PublishProfileEvent("profile.downloaded", order.ICCID, map[string]any{
"imsi": order.IMSI,
"profileType": order.ProfileType,
"status": downloadResp.ExecutionStatus,
})
}
c.JSON(http.StatusOK, gin.H{
"success": true,
"profile": &ProfileResponse{
@@ -170,7 +180,7 @@ func ListProfilesHandlerWithRepo(repo repository.ProfileRepository) gin.HandlerF
}
// DeleteProfileHandlerWithRepo deletes a profile from the SM-DP+ and marks it deleted in the repo.
func DeleteProfileHandlerWithRepo(client *es2.ES2Client, repo repository.ProfileRepository, webhookClient *webhook.WebhookClient) gin.HandlerFunc {
func DeleteProfileHandlerWithRepo(client *es2.ES2Client, repo repository.ProfileRepository, webhookClient *webhook.WebhookClient, messageQueue *mq.MessageQueue) gin.HandlerFunc {
return func(c *gin.Context) {
profileID := c.Param("profileId")
if profileID == "" {
@@ -197,6 +207,14 @@ func DeleteProfileHandlerWithRepo(client *es2.ES2Client, repo repository.Profile
})
}
// Publish message queue event for profile deletion
if messageQueue != nil {
go messageQueue.PublishProfileEvent("profile.deleted", profileID, map[string]any{
"executionStatus": deleteResp.ExecutionStatus,
"statusMessage": deleteResp.StatusMessage,
})
}
c.JSON(http.StatusOK, gin.H{
"success": true,
"message": "Profile deletion requested",
apps/carrier-connector/main.go
+22 -1
View File
@@ -10,6 +10,7 @@ import (
"github.com/nutcas3/telecom-platform/apps/carrier-connector/internal/config"
"github.com/nutcas3/telecom-platform/apps/carrier-connector/internal/es2"
handler "github.com/nutcas3/telecom-platform/apps/carrier-connector/internal/handler"
"github.com/nutcas3/telecom-platform/apps/carrier-connector/internal/mq"
"github.com/nutcas3/telecom-platform/apps/carrier-connector/internal/repository"
"github.com/nutcas3/telecom-platform/apps/carrier-connector/internal/webhook"
"github.com/rs/zerolog"
@@ -62,7 +63,27 @@ func main() {
webhookAPIKey := handler.GetEnv("API_SERVER_WEBHOOK_API_KEY", "")
webhookClient := webhook.NewWebhookClient(webhookURL, webhookAPIKey)
setupRoutes(router, client, profileRepo, webhookClient)
// Initialize message queue for async events (optional)
var messageQueue *mq.MessageQueue
rabbitmqURL := handler.GetEnv("RABBITMQ_URL", "")
if rabbitmqURL != "" {
mq, err := mq.NewMessageQueue(rabbitmqURL)
if err != nil {
handler.Logger.Warn().Err(err).Msg("Failed to connect to RabbitMQ, continuing without message queue")
} else {
// Declare profile events queue
if err := mq.DeclareQueue("profile-events"); err != nil {
handler.Logger.Warn().Err(err).Msg("Failed to declare profile-events queue")
mq.Close()
} else {
messageQueue = mq
handler.Logger.Info().Msg("Message queue initialized")
defer messageQueue.Close()
}
}
}
setupRoutes(router, client, profileRepo, webhookClient, messageQueue)
handler.Logger.Info().Str("port", port).Msg("Carrier Connector API server starting")
if err := router.Run(":" + port); err != nil {
apps/carrier-connector/routes.go
+5 -4
View File
@@ -9,12 +9,13 @@ import (
"github.com/nutcas3/telecom-platform/apps/carrier-connector/internal/es2"
handler "github.com/nutcas3/telecom-platform/apps/carrier-connector/internal/handler"
"github.com/nutcas3/telecom-platform/apps/carrier-connector/internal/mq"
"github.com/nutcas3/telecom-platform/apps/carrier-connector/internal/repository"
"github.com/nutcas3/telecom-platform/apps/carrier-connector/internal/webhook"
)
// setupRoutes registers all HTTP routes, wiring the ES2+ client, profile repo, and webhook client.
func setupRoutes(router *gin.Engine, client *es2.ES2Client, repo repository.ProfileRepository, webhookClient *webhook.WebhookClient) {
// setupRoutes registers all HTTP routes, wiring the ES2+ client, profile repo, webhook client, and message queue.
func setupRoutes(router *gin.Engine, client *es2.ES2Client, repo repository.ProfileRepository, webhookClient *webhook.WebhookClient, messageQueue *mq.MessageQueue) {
api := router.Group("/api/v1")
api.GET("/health", healthHandler)
api.GET("/health/ready", readinessHandler(repo))
@@ -23,10 +24,10 @@ func setupRoutes(router *gin.Engine, client *es2.ES2Client, repo repository.Prof
esim := api.Group("/esim")
{
esim.POST("/profiles", handler.OrderProfileHandlerWithRepo(client, repo, webhookClient))
esim.POST("/profiles", handler.OrderProfileHandlerWithRepo(client, repo, webhookClient, messageQueue))
esim.GET("/profiles", handler.ListProfilesHandlerWithRepo(repo))
esim.GET("/profiles/:profileId", handler.GetProfileHandlerWithRepo(client, repo))
esim.DELETE("/profiles/:profileId", handler.DeleteProfileHandlerWithRepo(client, repo, webhookClient))
esim.DELETE("/profiles/:profileId", handler.DeleteProfileHandlerWithRepo(client, repo, webhookClient, messageQueue))
}
carrier := api.Group("/carrier")