1) Priority support

2) External config files
3) Upgraded java-apns to latest available build

Author: phaneesh

CHANGELOG.md

@@ -18,3 +18,7 @@
 * Adaptive Loadbalancing on akka cluster
 * Scala 2.11.2 update
 * Java 8 update
+
+## 2.2 (December 30, 2014)
+* Priority support
+* External configuration files (In config directory)

README.md

@@ -18,6 +18,7 @@ Support for MongoDB datasource is provided out-of-the-box.
 | Version 1.0.1  | May 2014      |   Upgrade to Spray 1.3 & Scala 2.11.0 |
 | Version 2.0    | December 2014 |   Move to akka.io cluster & distributed pubsub |
 | Version 2.1    | December 2014 |   Adaptive Loadbalancing on akka cluster |
+| Version 2.2    | December 2014 |   Priority support for push messages |
 
 ## Changelog
 Changelog can be viewed in [CHANGELOG.md](https://github.com/Flipkart/flipcast/blob/master/CHANGELOG.md) file
@@ -41,9 +42,9 @@ Changelog can be viewed in [CHANGELOG.md](https://github.com/Flipkart/flipcast/b
 ## Library Dependencies
 --------------------
 * [spray](http://spray.io) 1.3.2
-* [Scala](http://www.scala-lang.org) 2.11.0
+* [Scala](http://www.scala-lang.org) 2.11.2
 * [akka.io](http://akka.io) 2.3.6
-* [java-apns](https://github.com/notnoop/java-apns) - 0.2.3
+* [java-apns](https://github.com/notnoop/java-apns) - 1.0.0.Beta4
 * [casbah](http://mongodb.github.io/casbah) - 2.8.0-RC0
 
 ## Infrastructure Dependencies

build.sbt

@@ -7,7 +7,7 @@ organization  := "com.flipcast"
 
 name          := "flipcast"
 
-version       := "2.1"
+version       := "2.2"
 
 scalaVersion  := "2.11.2"
 

src/main/resources/application.conf config/application.conf

@@ -137,27 +137,35 @@ flipcast.config {
       source = "push_configuration"
     }
   }
-  queue {
+  worker {
     configs = [ "gcm", "apns", "mpns" ]
     gcm {
-      workerName = "gcmRequestConsumer"
-      sidelineWorkerName = "flipcastSidelineConsumer"
-      workerInstances = 8
+      priorityTags = ["default"]
+      default {
+        sidelineWorkerName = "flipcastSidelineConsumer"
+        workerInstances = 8
+      }
     }
     apns {
-      workerName = "apnsRequestConsumer"
-      sidelineWorkerName = "flipcastSidelineConsumer"
-      workerInstances = 4
+      priorityTags = ["default"]
+      default {
+        sidelineWorkerName = "flipcastSidelineConsumer"
+        workerInstances = 4
+      }
     }
     mpns {
-      workerName = "mpnsRequestConsumer"
-      sidelineWorkerName = "flipcastSidelineConsumer"
-      workerInstances = 4
+      priorityTags = ["default"]
+      default {
+        sidelineWorkerName = "flipcastSidelineConsumer"
+        workerInstances = 4
+      }
     }
     bulk {
-      workerName = "bulkRequestConsumer"
-      sidelineWorkerName = "flipcastSidelineConsumer"
-      workerInstances = 2
+      priorityTags = ["default"]
+      default {
+        sidelineWorkerName = "flipcastSidelineConsumer"
+        workerInstances = 2
+      }
     }
   }
 }

src/main/resources/logback.xml config/logback.xml

@@ -60,6 +60,10 @@ JAVA_OPTS="${JAVA_OPTS} -Dcom.sun.management.jmxremote.port=29005"
 JAVA_OPTS="${JAVA_OPTS} -Dcom.sun.management.jmxremote.authenticate=false"
 JAVA_OPTS="${JAVA_OPTS} -Dcom.sun.management.jmxremote.ssl=false"
 
+#Provide application configuration & log configuration file path
+JAVA_OPTS="${JAVA_OPTS} -Dapp.config=config/application.conf"
+JAVA_OPTS="${JAVA_OPTS} -Dlogback.configurationFile=config/logback.xml"
+
 #Java Executable
 JAVA=`which java`
 

flipcast-start

@@ -56,5 +56,9 @@ set JAVA_OPTS="%JAVA_OPTS% -Dcom.sun.management.jmxremote.port=29005"
 set JAVA_OPTS="%JAVA_OPTS% -Dcom.sun.management.jmxremote.authenticate=false"
 set JAVA_OPTS="%JAVA_OPTS% -Dcom.sun.management.jmxremote.ssl=false"
 
+REM Provide application configuration & log configuration file path
+set JAVA_OPTS="%JAVA_OPTS% -Dapp.config=config\\application.conf"
+set JAVA_OPTS="%JAVA_OPTS% -Dlogback.configurationFile=config\\logback.xml"
+
 REM Execute main application JAR
 java -jar "%JAVA_OPTS%" "%APP_JAR%"

flipcast-start.cmd

@@ -1,5 +1,6 @@
 package com.flipcast
 
+import java.io.File
 import java.net.InetAddress
 import java.util.concurrent.TimeUnit
 
@@ -53,7 +54,7 @@ object Flipcast extends App {
   /**
    * Load application configuration
    */
-  lazy val config = ConfigFactory.load()
+  lazy val config = ConfigFactory.parseFile(new File(System.getProperty("app.config"))).resolve()
 
   /**
    * Actor system for flipcast service
@@ -136,18 +137,25 @@ object Flipcast extends App {
    * Start all the message consumers
    */
   def startMessageConsumers(isLocal: Boolean) {
-    serviceRegistry.register[FlipcastGcmRequestConsumer]("gcmRequestConsumer",
-      instances = QueueConfigurationManager.config("gcm").workerInstances,
-      dispatcher = "akka.actor.gcm-dispatcher", isLocal = isLocal)
-    serviceRegistry.register[FlipcastApnsRequestConsumer]("apnsRequestConsumer",
-      instances = QueueConfigurationManager.config("apns").workerInstances,
-      dispatcher = "akka.actor.apns-dispatcher", isLocal = isLocal)
-    serviceRegistry.register[FlipcastMpnsRequestConsumer]("mpnsRequestConsumer",
-      instances = QueueConfigurationManager.config("mpns").workerInstances,
-      dispatcher = "akka.actor.mpns-dispatcher", isLocal = isLocal)
-    serviceRegistry.register[BulkMessageConsumer]("bulkMessageConsumer",
-      instances = QueueConfigurationManager.config("bulk").workerInstances,
-      isLocal = isLocal)
+    WorkerConfigurationManager.config("gcm").priorityConfigs.foreach{ case (w, c) => {
+      serviceRegistry.register[FlipcastGcmRequestConsumer](c.workerName,
+        instances = c.workerInstances,
+        dispatcher = "akka.actor.gcm-dispatcher", isLocal = isLocal)
+    }}
+    WorkerConfigurationManager.config("apns").priorityConfigs.foreach{ case (w, c) => {
+      serviceRegistry.register[FlipcastApnsRequestConsumer](c.workerName,
+        instances = c.workerInstances,
+        dispatcher = "akka.actor.apns-dispatcher", isLocal = isLocal)
+    }}
+    WorkerConfigurationManager.config("mpns").priorityConfigs.foreach{ case (w, c) => {
+      serviceRegistry.register[FlipcastMpnsRequestConsumer](c.workerName,
+        instances = c.workerInstances,
+        dispatcher = "akka.actor.mpns-dispatcher", isLocal = isLocal)
+    }}
+    WorkerConfigurationManager.config("bulk").priorityConfigs.foreach{ case (w, c) => {
+      serviceRegistry.register[BulkMessageConsumer](c.workerName,
+        instances = c.workerInstances, isLocal = isLocal)
+    }}
   }
 
   def startMetrics() {

src/main/scala/com/flipcast/Flipcast.scala

@@ -77,7 +77,7 @@ trait BulkMessageRequestProtocol extends DefaultJsonProtocol with SprayJsonSuppo
       }
       val message = json.asJsObject.fields.contains("message") match {
         case true => json.asJsObject.fields("message").convertTo[PushMessage]
-        case _ => PushMessage("{}", Option(0), Option(true))
+        case _ => PushMessage("{}", Option(0), Option(true), Option("default"))
       }
       BulkMessageRequest(configName, query, message, start, end)
     }

src/main/scala/com/flipcast/protocol/BulkMessageRequestProtocol.scala

@@ -7,7 +7,7 @@ import akka.actor.{Actor, ActorSystem}
 import akka.event.slf4j.Logger
 import akka.util.Timeout
 import com.flipcast.Flipcast
-import com.flipcast.push.config.QueueConfigurationManager
+import com.flipcast.push.config.WorkerConfigurationManager
 import com.flipcast.push.model.SidelinedMessage
 import com.flipcast.push.model.requests.{FlipcastPushRequest, FlipcastRequest}
 import com.flipcast.push.protocol.FlipcastPushProtocol
@@ -33,19 +33,19 @@ abstract class  FlipcastRequestConsumer[T <: FlipcastRequest: ClassTag] extends
 
   def consume(message: T) : Boolean
 
-  def config = QueueConfigurationManager.config(configType())
+  def config = WorkerConfigurationManager.config(configType())
 
   def init()
 
   lazy val log = Logger(configType())
 
   override def preStart() {
     init()
-    log.info("Starting message consumer on: " +config.workerName +" Worker: " +self.path)
+    log.info("Starting message consumer on: " +config.configName +" Worker: " +self.path)
   }
 
   override def postStop(): Unit = {
-    log.info("Stopping message consumer on: " +config.workerName +" Worker: " +self.path)
+    log.info("Stopping message consumer on: " +config.configName +" Worker: " +self.path)
   }
 
   def receive = {
@@ -67,14 +67,16 @@ abstract class  FlipcastRequestConsumer[T <: FlipcastRequest: ClassTag] extends
   private def sideline(message: T) {
     message match {
       case x: FlipcastPushRequest =>
-        Flipcast.serviceRegistry.actor(config.sidelineWorkerName) !
+        Flipcast.serviceRegistry.actorLookup(config.priorityConfigs(x.priority.getOrElse("default")).sidelineWorkerName) !
           SidelinedMessage(UUID.randomUUID().toString,
             x.configName, configType(), x.toJson.compactPrint, new Date())
     }
   }
 
   def resend(message: T) {
-    Flipcast.serviceRegistry.actor(config.workerName) ! message
+    message match {
+      case x: FlipcastPushRequest =>
+        Flipcast.serviceRegistry.actorLookup(config.priorityConfigs(x.priority.getOrElse("default")).workerName) ! message
+    }
   }
-
 }

src/main/scala/com/flipcast/push/common/FlipcastRequestConsumer.scala

@@ -0,0 +1,11 @@
+package com.flipcast.push.config
+
+
+case class PriorityConfig(workerInstances: Int = 1, workerName: String, sidelineWorkerName: String)
+
+/**
+ * Config to hold worker configuration
+ *
+ * @author Phaneesh Nagaraja
+ */
+case class WorkerConfig(configName: String, priorityConfigs: Map[String, PriorityConfig])

src/main/scala/com/flipcast/push/config/QueueConfig.scala

@@ -0,0 +1,43 @@
+package com.flipcast.push.config
+
+import java.util.concurrent.ConcurrentHashMap
+
+import com.flipcast.Flipcast
+import collection.JavaConverters._
+
+/**
+ * Provides queue configuration for different push service providers
+ *
+ * @author Phaneesh Nagaraja
+ */
+object WorkerConfigurationManager {
+
+  val workerConfigurationCache = new ConcurrentHashMap[String, WorkerConfig]()
+
+  def configs() : List[String] = {
+    Flipcast.config.getStringList("flipcast.config.queue.configs").asScala.toList
+  }
+
+  def config(configType: String) : WorkerConfig = {
+    if(!workerConfigurationCache.containsKey(configType)) {
+      val c = Flipcast.config.getConfig("flipcast.config.worker." + configType)
+      val priorityTags = c.getStringList("priorityTags").asScala
+      workerConfigurationCache.put(configType, WorkerConfig(
+        configType,
+        priorityTags.map( pt => {
+          pt -> PriorityConfig(c.getInt(pt +".workerInstances"), String.format("%sMessageConsumer-%s", configType, pt),
+            c.getString( pt +".sidelineWorkerName"))
+        }).toMap))
+    }
+    workerConfigurationCache.get(configType)
+  }
+
+  def worker(configType: String, priorityTag: String) = {
+    config(configType).priorityConfigs(priorityTag).workerName
+  }
+
+  def bulkConfig() = {
+    config("bulk")
+  }
+
+}

src/main/scala/com/flipcast/push/config/QueueConfigurationManager.scala

@@ -109,7 +109,7 @@ class FlipcastGcmRequestConsumer extends FlipcastRequestConsumer[FlipcastPushReq
             failedIds.isEmpty match {
               case true => None
               case false =>
-                resend(FlipcastPushRequest(request.configName, failedIds.toList, request.data, request.ttl, request.delayWhileIdle))
+                resend(FlipcastPushRequest(request.configName, failedIds.toList, request.data, request.ttl, request.delayWhileIdle, request.priority))
             }
             //Record history for all successful devices
             request.registration_ids.diff(failedIds.toList).par.foreach( r => {

src/main/scala/com/flipcast/push/config/WorkerConfig.scala

@@ -5,4 +5,4 @@ package com.flipcast.push.model
  *
  * @author Phaneesh Nagaraja
  */
-case class PushMessage(message: String, ttl: Option[Int], delayWhileIdle: Option[Boolean])
+case class PushMessage(message: String, ttl: Option[Int], delayWhileIdle: Option[Boolean], priority: Option[String])

src/main/scala/com/flipcast/push/config/WorkerConfigurationManager.scala

@@ -14,4 +14,4 @@ case class FlipcastPushRequest(configName: String,
                                registration_ids: List[String],
                                data: String,
                                ttl: Option[Int],
-                               delayWhileIdle: Option[Boolean]) extends FlipcastRequest
+                               delayWhileIdle: Option[Boolean], priority: Option[String]) extends FlipcastRequest

src/main/scala/com/flipcast/push/gcm/service/FlipcastGcmRequestConsumer.scala

@@ -82,7 +82,7 @@ class FlipcastMpnsRequestConsumer extends FlipcastRequestConsumer[FlipcastPushRe
       Flipcast.serviceRegistry.actor("deviceHouseKeepingManager") ! DeviceHousekeepingRequest(request.configName, r)
     })
     if(failedIds.size > 0) {
-      resend(FlipcastPushRequest(request.configName, failedIds.toList, request.data, request.ttl, request.delayWhileIdle))
+      resend(FlipcastPushRequest(request.configName, failedIds.toList, request.data, request.ttl, request.delayWhileIdle, request.priority))
     }
     true
   }

src/main/scala/com/flipcast/push/model/PushMessage.scala

@@ -68,12 +68,21 @@ trait FlipcastPushProtocol extends DefaultJsonProtocol with SprayJsonSupport {
           }
         case _ => false
       }
+      val priority = json.asJsObject.fields.contains("priority") match {
+        case true =>
+          json.asJsObject.fields("priority") match {
+            case a: JsString => Option(a.value)
+            case _ => None
+          }
+        case _ => None
+      }
+
       val data = json.asJsObject.fields.contains("data") match {
         case true => json.asJsObject.fields("data").compactPrint
         case _ => "{}"
       }
       FlipcastPushRequest(configName, registration_ids.filter( _.trim.length > 0), data, Option(ttl),
-        Option(delayWhileIdle))
+        Option(delayWhileIdle), priority)
     }
   }