feat: ingest iOS push notifications via FCM Kafka topics

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

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

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 {

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"))
+			})
+		})
 	})
 })

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

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() {

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.

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)
+			}
 		}
 	}