From d78859ea07e846512ac80d8fe1b8f9ed240959c8 Mon Sep 17 00:00:00 2001 From: Lucas Monteiro Date: Tue, 28 Apr 2026 13:54:32 -0300 Subject: [PATCH] feat: ingest iOS push notifications via FCM Kafka topics Add ingestion for the new push-_ios-(single|massive) topics so the GCM pusher can consume iOS messages alongside the existing Android ones. - Extend topic regex in config and common.go to match the ios platform - Add Platform field to KafkaMessage and populate it from the topic - Subscribe pusher gcm to ios single/massive topics for every gcm app - Cover new parsing and consumer behavior with unit tests --- config/default.yaml | 2 +- config/test.yaml | 2 +- extensions/common.go | 2 +- extensions/common_test.go | 44 +++++++++++++++++++++++++++++++ extensions/kafka_consumer.go | 8 +++--- extensions/kafka_consumer_test.go | 38 ++++++++++++++++++++++++++ interfaces/queue.go | 7 ++--- pusher/gcm.go | 16 ++++++----- 8 files changed, 103 insertions(+), 16 deletions(-) diff --git a/config/default.yaml b/config/default.yaml index 097db795..3dec5dfe 100644 --- a/config/default.yaml +++ b/config/default.yaml @@ -34,7 +34,7 @@ gcm: concurrentWorkers: 10 queue: topics: - - "^push-[^-_]+_(apns|gcm)[_-](single|massive)" + - "^push-[^-_]+_(apns|gcm|ios)[_-](single|massive)" brokers: "localhost:9941" group: testGroup sessionTimeout: 6000 diff --git a/config/test.yaml b/config/test.yaml index 1e505208..c87d5a92 100644 --- a/config/test.yaml +++ b/config/test.yaml @@ -28,7 +28,7 @@ gcm: concurrentWorkers: 10 queue: topics: - - "^push-[^-_]+_(apns|gcm)[_-](single|massive)" + - "^push-[^-_]+_(apns|gcm|ios)[_-](single|massive)" brokers: "localhost:9941" group: testGroup sessionTimeout: 6000 diff --git a/extensions/common.go b/extensions/common.go index 111cfb41..cf6ab43c 100644 --- a/extensions/common.go +++ b/extensions/common.go @@ -31,7 +31,7 @@ import ( "github.com/topfreegames/pusher/interfaces" ) -var topicRegex = regexp.MustCompile("^push-([\\w]+(?:[_-][\\w]+)*)[-_](gcm|apns)") +var topicRegex = regexp.MustCompile("^push-([\\w]+(?:[_-][\\w]+)*)[-_](gcm|apns|ios)") // ParsedTopic contains game and platform extracted from topic name type ParsedTopic struct { diff --git a/extensions/common_test.go b/extensions/common_test.go index 4c3ab381..7ffacefa 100644 --- a/extensions/common_test.go +++ b/extensions/common_test.go @@ -68,5 +68,49 @@ var _ = Describe("Common", func() { Expect(err.Error()).To(Equal("json: unsupported type: chan int")) }) }) + + Describe("GetGameAndPlatformFromTopic", func() { + It("should parse gcm single topic", func() { + parsed := GetGameAndPlatformFromTopic("push-mygame_gcm-single") + Expect(parsed.Game).To(Equal("mygame")) + Expect(parsed.Platform).To(Equal("gcm")) + }) + + It("should parse gcm massive topic", func() { + parsed := GetGameAndPlatformFromTopic("push-mygame_gcm-massive") + Expect(parsed.Game).To(Equal("mygame")) + Expect(parsed.Platform).To(Equal("gcm")) + }) + + It("should parse apns single topic", func() { + parsed := GetGameAndPlatformFromTopic("push-mygame_apns-single") + Expect(parsed.Game).To(Equal("mygame")) + Expect(parsed.Platform).To(Equal("apns")) + }) + + It("should parse apns massive topic", func() { + parsed := GetGameAndPlatformFromTopic("push-mygame_apns-massive") + Expect(parsed.Game).To(Equal("mygame")) + Expect(parsed.Platform).To(Equal("apns")) + }) + + It("should parse ios single topic", func() { + parsed := GetGameAndPlatformFromTopic("push-mygame_ios-single") + Expect(parsed.Game).To(Equal("mygame")) + Expect(parsed.Platform).To(Equal("ios")) + }) + + It("should parse ios massive topic", func() { + parsed := GetGameAndPlatformFromTopic("push-mygame_ios-massive") + Expect(parsed.Game).To(Equal("mygame")) + Expect(parsed.Platform).To(Equal("ios")) + }) + + It("should parse ios topic with compound game name", func() { + parsed := GetGameAndPlatformFromTopic("push-com_my_game_ios-single") + Expect(parsed.Game).To(Equal("com_my_game")) + Expect(parsed.Platform).To(Equal("ios")) + }) + }) }) }) diff --git a/extensions/kafka_consumer.go b/extensions/kafka_consumer.go index 2014f4ca..91b7b410 100644 --- a/extensions/kafka_consumer.go +++ b/extensions/kafka_consumer.go @@ -273,10 +273,12 @@ func (q *KafkaConsumer) receiveMessage(topicPartition kafka.TopicPartition, valu q.pendingMessagesWG.Add(1) } + parsed := GetGameAndPlatformFromTopic(*topicPartition.Topic) message := interfaces.KafkaMessage{ - Game: GetGameAndPlatformFromTopic(*topicPartition.Topic).Game, - Topic: *topicPartition.Topic, - Value: value, + Game: parsed.Game, + Platform: parsed.Platform, + Topic: *topicPartition.Topic, + Value: value, } q.msgChan <- message diff --git a/extensions/kafka_consumer_test.go b/extensions/kafka_consumer_test.go index 2e1747f6..cbe58e50 100644 --- a/extensions/kafka_consumer_test.go +++ b/extensions/kafka_consumer_test.go @@ -116,6 +116,44 @@ var _ = Describe("Kafka Extension", func() { Value: val, })) }) + + It("should populate Game and Platform on received message for ios topic", func() { + topic := "push-mygame_ios-single" + startConsuming() + defer consumer.StopConsuming() + part := kafka.TopicPartition{ + Topic: &topic, + Partition: 1, + } + val := []byte("test") + event := &kafka.Message{TopicPartition: part, Value: val} + + publishEvent(event) + var received interfaces.KafkaMessage + Eventually(consumer.msgChan, 5).Should(Receive(&received)) + Expect(received.Topic).To(Equal(topic)) + Expect(received.Game).To(Equal("mygame")) + Expect(received.Platform).To(Equal("ios")) + Expect(received.Value).To(Equal(val)) + }) + + It("should populate Game and Platform on received message for gcm topic", func() { + topic := "push-mygame_gcm-massive" + startConsuming() + defer consumer.StopConsuming() + part := kafka.TopicPartition{ + Topic: &topic, + Partition: 1, + } + val := []byte("test") + event := &kafka.Message{TopicPartition: part, Value: val} + + publishEvent(event) + var received interfaces.KafkaMessage + Eventually(consumer.msgChan, 5).Should(Receive(&received)) + Expect(received.Game).To(Equal("mygame")) + Expect(received.Platform).To(Equal("gcm")) + }) }) Describe("Configuration Defaults", func() { diff --git a/interfaces/queue.go b/interfaces/queue.go index 4695d1ea..7e8243f6 100644 --- a/interfaces/queue.go +++ b/interfaces/queue.go @@ -29,9 +29,10 @@ import ( // KafkaMessage sent through the Channel. type KafkaMessage struct { - Game string - Topic string - Value []byte + Game string + Platform string + Topic string + Value []byte } // Queue interface for making new queues pluggable easily. diff --git a/pusher/gcm.go b/pusher/gcm.go index 288cb691..b0639035 100644 --- a/pusher/gcm.go +++ b/pusher/gcm.go @@ -82,13 +82,15 @@ func NewGCMPusher( } g.Queue = q for _, a := range g.Config.GetGcmAppsArray() { - singleTopic := fmt.Sprintf("push-%s_gcm-single", a) - if !slices.Contains(q.Topics, singleTopic) { - q.Topics = append(q.Topics, singleTopic) - } - massiveTopic := fmt.Sprintf("push-%s_gcm-massive", a) - if !slices.Contains(q.Topics, massiveTopic) { - q.Topics = append(q.Topics, massiveTopic) + for _, platform := range []string{"gcm", "ios"} { + singleTopic := fmt.Sprintf("push-%s_%s-single", a, platform) + if !slices.Contains(q.Topics, singleTopic) { + q.Topics = append(q.Topics, singleTopic) + } + massiveTopic := fmt.Sprintf("push-%s_%s-massive", a, platform) + if !slices.Contains(q.Topics, massiveTopic) { + q.Topics = append(q.Topics, massiveTopic) + } } }