Pool tags support

* Added support for tags in pool configurations
* Added support for selecting pools based on tags
* Updated sparrow ref (just now many FCM pools are supported in sparrow)
* Updated README
* Fixed issue with no matching pools in FCM

README.md

@@ -169,6 +169,7 @@ Each `FCM` pool may be configured by setting the following fields:
 * **pool_size** (*required*) - maximum number of used `HTTP/2` connections to google's service
 * **mode** (*either `:prod` or `:dev`*) - pool's mode. The `HTTP` client may select pool used to push a notification by specifying matching option in the request
 * **endpoint** (*optional*) - URL override for `FCM` service. Useful mainly in tests
+* **tags** (*optional*) - a list of tags. Used when choosing pool to match request tags when sending a notification. More details: https://github.com/esl/sparrow#tags
 
 You may entirely skip the `FCM` config entry to disable `FCM` support.
 
@@ -202,6 +203,7 @@ Each `APNS` pool may be configured by setting the following fields:
 * **mode** (*either `:prod` or `:dev`*) - pool's mode. The `HTTP` client may select pool used to push a notification by specifying matching option in the request
 * **endpoint** (*optional*) - URL override for `APNS` service. Useful mainly in tests
 * **use_2197** (*optional `true` or `false`*) - whether use alternative port for `APNS`: 2197
+* **tags** (*optional*) - a list of tags. Used when choosing pool to match request tags when sending a notification. More details: https://github.com/esl/sparrow#tags
 
 You may entirely skip the `APNS` config entry to disable `APNS` support.
 
@@ -250,6 +252,7 @@ The full list of options contains the following:
 * **time_to_live** (*optional*) - Maximum lifespan of an FCM notification. For more details, please, refer to [the official FCM documentation](https://firebase.google.com/docs/cloud-messaging/concept-options#ttl).
 * **mutable_content** (*optional*, `true` / `false` (default)) - Only applicable to APNS. Sets "mutable-content=1" in APNS payload.
 * **topic** (*optional*, `APNS` specific) - if APNS certificate configured in `MongoosePush` allows for multiple applications, this field selects the application. Please refer to `APNS` documentation for more datails
+* **tags** (*optional*) - a list of tags used to choose a pool with matching tags. To see how tags work read: https://github.com/esl/sparrow#tags
 * **data** (*optional*) - custom JSON structure sent to the target device. For `APNS`, all keys form this stucture are merged into highest level APS message (the one that holds 'aps' key), while for `FCM` the whole `data` json stucture is sent as FCM's `data payload` along with `notification`.
 * **alert** (*optional*) - JSON stucture that if provided will send non-silent notification with the following fields:
   * **body** (*required*) - text body of notification

config/test.exs

@@ -21,12 +21,21 @@ config :maru, MongoosePush.Router,
 
 config :mongoose_push,
   fcm: [
-    default: [
+    pool1: [
       appfile: "priv/fcm/token.json",
       endpoint: "localhost",
       pool_size: 5,
       mode: :prod,
-      port: 4000
+      port: 4000,
+      tags: [:I, :am, :your, :father]
+    ],
+    pool2: [
+      appfile: "priv/fcm/token.json",
+      endpoint: "localhost",
+      pool_size: 3,
+      mode: :dev,
+      port: 4000,
+      tags: [:these, :are, :not]
     ]
   ]
 

lib/mongoose_push.ex

@@ -63,21 +63,32 @@ defmodule MongoosePush do
   def push(device_id, %{:service => service} = request) do
     mode = Map.get(request, :mode, :prod)
     module = MongoosePush.Application.services()[service]
-    pool = module.choose_pool(mode)
-    # Just make sure both data and alert keys exist for convenience (but may be nil)
-    request =
-      request
-      |> Map.put(:alert, request[:alert])
-      |> Map.put(:data, request[:data])
-
-    notification = module.prepare_notification(device_id, request, pool)
-    opts = [timeout: 60_000]
-    {time, push_result} = :timer.tc(module, :push, [notification, device_id, pool, opts])
-
-    push_result
-    |> Metrics.update(:spiral, [:push, service, mode])
-    |> Metrics.update(:timer, [:push, service, mode], time)
-    |> maybe_log
+    tags = Map.get(request, :tags, [])
+    pool = module.choose_pool(mode, tags)
+
+    if pool == nil do
+      Logger.error(~s"No pool matching mode=#{mode} and tags=#{inspect(tags)}")
+
+      {:error, :no_matching_pool}
+      |> Metrics.update(:spiral, [:push, service, mode])
+      |> Metrics.update(:timer, [:push, service, mode])
+      |> maybe_log
+    else
+      # Just make sure both data and alert keys exist for convenience (but may be nil)
+      request =
+        request
+        |> Map.put(:alert, request[:alert])
+        |> Map.put(:data, request[:data])
+
+      notification = module.prepare_notification(device_id, request, pool)
+      opts = [timeout: 60_000]
+      {time, push_result} = :timer.tc(module, :push, [notification, device_id, pool, opts])
+
+      push_result
+      |> Metrics.update(:spiral, [:push, service, mode])
+      |> Metrics.update(:timer, [:push, service, mode], time)
+      |> maybe_log
+    end
   end
 
   defp maybe_log(:ok), do: :ok

lib/mongoose_push/service/apns.ex

@@ -59,9 +59,9 @@ defmodule MongoosePush.Service.APNS do
     {MongoosePush.Service.APNS.Supervisor, pool_configs}
   end
 
-  @spec choose_pool(MongoosePush.mode()) :: Application.pool_name() | nil
-  def choose_pool(mode) do
-    Sparrow.PoolsWarden.choose_pool({:apns, mode})
+  @spec choose_pool(MongoosePush.mode(), [any]) :: Application.pool_name() | nil
+  def choose_pool(mode, tags \\ []) do
+    Sparrow.PoolsWarden.choose_pool({:apns, mode}, tags)
   end
 
   defp maybe(notification, :add_mutable_content, true),

lib/mongoose_push/service/apns/supervisor.ex

@@ -83,8 +83,9 @@ defmodule MongoosePush.Service.APNS.Supervisor do
     port = {:port, port_config}
 
     name = {:pool_name, pool_name}
+    tags = {:tags, pool_config[:tags]}
 
-    [auth_type, cert, key, pool_size, endpoint, port, name]
+    [auth_type, cert, key, pool_size, endpoint, port, name, tags]
     |> Enum.filter(fn {_key, value} -> !is_nil(value) end)
   end
 
@@ -99,6 +100,7 @@ defmodule MongoosePush.Service.APNS.Supervisor do
     port = {:port, port_config}
 
     name = {:pool_name, pool_name}
+    tags = {:tags, pool_config[:tags]}
 
     key = pool_config[:auth][:key_id]
     team = pool_config[:auth][:team_id]
@@ -119,7 +121,7 @@ defmodule MongoosePush.Service.APNS.Supervisor do
     pool_token = {:token_id, token_id}
 
     single_config =
-      [auth_type, pool_token, pool_size, endpoint, port, name]
+      [auth_type, pool_token, pool_size, endpoint, port, name, tags]
       |> Enum.filter(fn {_key, value} -> !is_nil(value) end)
 
     token = [pool_token, key_id, team_id, p8_file]

lib/mongoose_push/service/fcm.ex

@@ -64,9 +64,9 @@ defmodule MongoosePush.Service.FCM do
     {PoolSupervisor, pools_configs}
   end
 
-  @spec choose_pool(MongoosePush.mode()) :: Application.pool_name() | nil
-  def choose_pool(mode) do
-    Sparrow.PoolsWarden.choose_pool(:fcm, [mode])
+  @spec choose_pool(MongoosePush.mode(), [any]) :: Application.pool_name() | nil
+  def choose_pool(mode, tags \\ []) do
+    Sparrow.PoolsWarden.choose_pool(:fcm, [mode | tags])
   end
 
   defp maybe(notification, _function, nil), do: notification

lib/mongoose_push/service/fcm/pool/supervisor.ex

@@ -39,9 +39,10 @@ defmodule MongoosePush.Service.FCM.Pool.Supervisor do
     endpoint = {:endpoint, pool_config[:endpoint] || @default_endpoint}
     port = {:port, pool_config[:port] || @default_port}
     pool_size = {:worker_num, pool_config[:pool_size]}
+    raw_tags = if is_nil(pool_config[:tags]), do: [], else: pool_config[:tags]
 
     # mode has to be either `prod` or `dev`, for now we pass it in form of a tag
-    tags = {:tags, [pool_config[:mode]]}
+    tags = {:tags, [pool_config[:mode] | raw_tags]}
 
     [token_path, endpoint, port, pool_size, tags]
     |> Enum.filter(fn {_key, value} -> !is_nil(value) end)

mix.exs

@@ -27,7 +27,7 @@ defmodule MongoosePush.Mixfile do
   defp deps do
     [
       {:chatterbox, github: "joedevivo/chatterbox", ref: "ff0c2e0", override: true},
-      {:sparrow, github: "esl/sparrow", ref: "358a816b913362daa99f126ceb35b172aa511044"},
+      {:sparrow, github: "esl/sparrow", ref: "ed31463f03e83d227d5b6af5c1379a8ab6a551b2"},
       {:maru, github: "rslota/maru", ref: "54fc038", override: true},
       {:cowboy, "~> 2.3", override: true},
       {:jason, "~> 1.0"},

mix.lock

@@ -21,7 +21,7 @@
   "exometer_report_graphite": {:git, "https://github.com/esl/exometer_report_graphite.git", "264dd7bcbadbd7febcd43917302251286c88b681", []},
   "folsom": {:hex, :folsom, "0.8.5", "94a027b56fe84feed264f9b33cb4c6ac9a801fad84b87dbda0836ce83c3b8d69", [:rebar3], [{:bear, "0.8.5", [hex: :bear, repo: "hexpm", optional: false]}], "hexpm"},
   "goldrush": {:hex, :goldrush, "0.1.9", "f06e5d5f1277da5c413e84d5a2924174182fb108dabb39d5ec548b27424cd106", [:rebar3], [], "hexpm"},
-  "goth": {:hex, :goth, "0.8.2", "edd1359f4e612266188a6837fcb562626403458398c6ba36d6f8c88a14075366", [:mix], [{:httpoison, "~> 0.11 or ~> 1.0", [hex: :httpoison, repo: "hexpm", optional: false]}, {:json_web_token, "~> 0.2.10", [hex: :json_web_token, repo: "hexpm", optional: false]}, {:poison, "~> 2.1 or ~> 3.0", [hex: :poison, repo: "hexpm", optional: false]}], "hexpm"},
+  "goth": {:hex, :goth, "1.1.0", "85977656822e54217bc0472666f1ce15dc3921495ef5f4f0774ef15503bae207", [:mix], [{:httpoison, "~> 0.11 or ~> 1.0", [hex: :httpoison, repo: "hexpm", optional: false]}, {:jason, "~> 1.1", [hex: :jason, repo: "hexpm", optional: false]}, {:joken, "~> 2.0", [hex: :joken, repo: "hexpm", optional: false]}], "hexpm"},
   "hackney": {:hex, :hackney, "1.11.0", "4951ee019df102492dabba66a09e305f61919a8a183a7860236c0fde586134b6", [:rebar3], [{:certifi, "2.0.0", [hex: :certifi, repo: "hexpm", optional: false]}, {:idna, "5.1.0", [hex: :idna, repo: "hexpm", optional: false]}, {:metrics, "1.0.1", [hex: :metrics, repo: "hexpm", optional: false]}, {:mimerl, "1.0.2", [hex: :mimerl, repo: "hexpm", optional: false]}, {:ssl_verify_fun, "1.1.1", [hex: :ssl_verify_fun, repo: "hexpm", optional: false]}], "hexpm"},
   "hpack": {:git, "https://github.com/joedevivo/hpack.git", "6b58b6231e9b6ab83096715120578976f72f4f7c", [tag: "0.2.3"]},
   "httpoison": {:hex, :httpoison, "0.13.0", "bfaf44d9f133a6599886720f3937a7699466d23bb0cd7a88b6ba011f53c6f562", [:mix], [{:hackney, "~> 1.8", [hex: :hackney, repo: "hexpm", optional: false]}], "hexpm"},
@@ -51,7 +51,7 @@
   "poolboy": {:hex, :poolboy, "1.5.1", "6b46163901cfd0a1b43d692657ed9d7e599853b3b21b95ae5ae0a777cf9b6ca8", [:rebar], [], "hexpm"},
   "quixir": {:hex, :quixir, "0.9.3", "f01c37386b9e1d0526f01a8734a6d7884af294a0ec360f05c24c7171d74632bd", [:mix], [{:pollution, "~> 0.9.2", [hex: :pollution, repo: "hexpm", optional: false]}], "hexpm"},
   "ranch": {:hex, :ranch, "1.4.0", "10272f95da79340fa7e8774ba7930b901713d272905d0012b06ca6d994f8826b", [:rebar3], [], "hexpm"},
-  "sparrow": {:git, "https://github.com/esl/sparrow.git", "358a816b913362daa99f126ceb35b172aa511044", [ref: "358a816b913362daa99f126ceb35b172aa511044"]},
+  "sparrow": {:git, "https://github.com/esl/sparrow.git", "ed31463f03e83d227d5b6af5c1379a8ab6a551b2", [ref: "ed31463f03e83d227d5b6af5c1379a8ab6a551b2"]},
   "ssl_verify_fun": {:hex, :ssl_verify_fun, "1.1.1", "28a4d65b7f59893bc2c7de786dec1e1555bd742d336043fe644ae956c3497fbe", [:make, :rebar], [], "hexpm"},
   "unicode_util_compat": {:hex, :unicode_util_compat, "0.3.1", "a1f612a7b512638634a603c8f401892afbf99b8ce93a45041f8aaca99cadb85e", [:rebar3], [], "hexpm"},
   "uuid": {:hex, :uuid, "1.1.8", "e22fc04499de0de3ed1116b770c7737779f226ceefa0badb3592e64d5cfb4eb9", [:mix], [], "hexpm"},

test/mongoose_push_test.exs

@@ -389,6 +389,165 @@ defmodule MongoosePushTest do
     TestHelper.reload_app()
   end
 
+  describe "tagged pools" do
+    setup do
+      apns_config = [
+        dev1: [
+          auth: %{
+            type: :token_based,
+            key_id: "fake_key",
+            team_id: "fake_team",
+            p8_file_path: "priv/apns/token.p8"
+          },
+          endpoint: "localhost",
+          mode: :dev,
+          use_2197: true,
+          pool_size: 3,
+          default_topic: "dev_topic1",
+          tags: [:tag1, :tag2]
+        ],
+        dev2: [
+          auth: %{
+            type: :token_based,
+            key_id: "fake_key",
+            team_id: "fake_team",
+            p8_file_path: "priv/apns/token.p8"
+          },
+          endpoint: "localhost",
+          mode: :dev,
+          use_2197: true,
+          pool_size: 3,
+          default_topic: "prod_topic1",
+          tags: [:tag2, :tag3]
+        ],
+        prod1: [
+          auth: %{
+            type: :token_based,
+            key_id: "fake_key",
+            team_id: "fake_team",
+            p8_file_path: "priv/apns/token.p8"
+          },
+          endpoint: "localhost",
+          mode: :prod,
+          use_2197: true,
+          pool_size: 3,
+          default_topic: "dev_topic1",
+          tags: [:tag1, :tag2]
+        ],
+        prod2: [
+          auth: %{
+            type: :token_based,
+            key_id: "fake_key",
+            team_id: "fake_team",
+            p8_file_path: "priv/apns/token.p8"
+          },
+          endpoint: "localhost",
+          mode: :prod,
+          use_2197: true,
+          pool_size: 3,
+          default_topic: "prod_topic1",
+          tags: [:tag2, :tag3]
+        ]
+      ]
+
+      fcm_config = [
+        pool1: [
+          appfile: "priv/fcm/token.json",
+          endpoint: "localhost",
+          pool_size: 5,
+          mode: :prod,
+          port: 4000,
+          tags: [:tag1, :tag2, :tag3]
+        ],
+        pool2: [
+          appfile: "priv/fcm/token.json",
+          endpoint: "localhost",
+          pool_size: 4,
+          mode: :dev,
+          port: 4000,
+          tags: [:tag2, :tag3, :tag4]
+        ]
+      ]
+
+      Application.stop(:mongoose_push)
+      Application.stop(:sparrow)
+      Application.put_env(:mongoose_push, :apns, apns_config)
+      # Application.put_env(:mongoose_push, :fcm, fcm_config)
+      Application.ensure_all_started(:mongoose_push)
+      :ok
+    end
+
+    test "are tagged and chosen correctly" do
+      assert :dev1 == MongoosePush.Service.APNS.choose_pool(:dev, [:tag1])
+      assert :dev1 == MongoosePush.Service.APNS.choose_pool(:dev, [:tag1, :tag2])
+      assert :dev2 == MongoosePush.Service.APNS.choose_pool(:dev, [:tag3])
+      assert :dev2 == MongoosePush.Service.APNS.choose_pool(:dev, [:tag2, :tag3])
+      assert nil == MongoosePush.Service.APNS.choose_pool(:dev, [:tag1, :tag2, :tag3])
+
+      assert :prod1 == MongoosePush.Service.APNS.choose_pool(:prod, [:tag1])
+      assert :prod1 == MongoosePush.Service.APNS.choose_pool(:prod, [:tag1, :tag2])
+      assert :prod2 == MongoosePush.Service.APNS.choose_pool(:prod, [:tag3])
+      assert :prod2 == MongoosePush.Service.APNS.choose_pool(:prod, [:tag2, :tag3])
+      assert nil == MongoosePush.Service.APNS.choose_pool(:prod, [:tag1, :tag2, :tag3])
+
+      # assert :pool1 == MongoosePush.Service.FCM.choose_pool(:prod)
+      # assert :pool1 == MongoosePush.Service.FCM.choose_pool(:prod, [:tag1])
+      # assert :pool1 == MongoosePush.Service.FCM.choose_pool(:prod, [:tag1, :tag2, :tag3])
+      # assert :pool2 == MongoosePush.Service.FCM.choose_pool(:dev)
+      # assert :pool1 == MongoosePush.Service.FCM.choose_pool(:dev, [:tag2])
+      # assert :pool1 == MongoosePush.Service.FCM.choose_pool(:dev, [:tag2, :tag3, :tag4])
+      # assert nil == MongoosePush.Service.FCM.choose_pool(:prod, [:tag2, :tag3, :tag4])
+      # assert nil == MongoosePush.Service.FCM.choose_pool(:dev, [:tag1, :tag2, :tag3])
+      TestHelper.reload_app()
+    end
+
+    test "are integrated with APNS" do
+      notification = %{
+        :service => :apns,
+        :alert => %{
+          :title => "title",
+          :body => "body"
+        },
+        :mode => :prod,
+        :tags => [:tag2, :tag3],
+        :data => %{
+          "acme1" => "apns1",
+          "acme2" => "apns2",
+          "acme3" => "apns3"
+        }
+      }
+
+      assert :ok == push(@test_token, notification)
+
+      invalid_notification =
+        notification
+        |> Map.replace!(:tags, [:tag1, :tag2, :tag3])
+
+      assert {:error, :no_matching_pool} == push(@test_token, invalid_notification)
+      TestHelper.reload_app()
+    end
+
+    test "are integrated with FCM" do
+      notification = %{
+        :service => :fcm,
+        :data => %{
+          "acme1" => "fcm1",
+          "acme2" => "fcm2",
+          "acme3" => "fcm3"
+        }
+      }
+
+      assert :ok == push(@test_token, notification)
+
+      invalid_notification =
+        notification
+        |> Map.put(:tags, [:tag1, :tag2, :tag3, :tag4])
+
+      assert {:error, :no_matching_pool} == push(@test_token, invalid_notification)
+      TestHelper.reload_app()
+    end
+  end
+
   defp reset(:apns) do
     {:ok, conn} = get_connection(:apns)
     headers = headers("POST", "/reset")