feat(introspection): add killFilter and parseFilters to client

lib/excache/command/introspection/client.ex

@@ -155,3 +155,103 @@ defmodule Excache.Command.Introspection.Client do
 
   defp typeFilter(_raw), do: :error
 
+  defp killFilter(args) do
+    case parseFilters(args, %{skipme: true}) do
+      {:ok, filters} -> {:ok, buildMatcher(filters)}
+      :error -> :error
+    end
+  end
+
+  defp parseFilters([], filters), do: {:ok, filters}
+
+  defp parseFilters([keyword, value | rest], filters) do
+    upcased = String.upcase(keyword)
+
+    if upcased in @killFilters do
+      case addFilter(upcased, value, filters) do
+        {:ok, updated} -> parseFilters(rest, updated)
+        :error -> :error
+      end
+    else
+      :error
+    end
+  end
+
+  defp parseFilters(_args, _filters), do: :error
+
+  defp addFilter("ID", value, filters) do
+    case parseId(value) do
+      :error -> :error
+      id -> {:ok, Map.put(filters, :id, id)}
+    end
+  end
+
+  defp addFilter("ADDR", value, filters), do: {:ok, Map.put(filters, :addr, value)}
+  defp addFilter("LADDR", value, filters), do: {:ok, Map.put(filters, :laddr, value)}
+
+  defp addFilter("TYPE", value, filters) do
+    lowered = String.downcase(value)
+    if lowered in @clientTypes, do: {:ok, Map.put(filters, :type, lowered)}, else: :error
+  end
+
+  defp addFilter("MAXAGE", value, filters) do
+    case Integer.parse(value) do
+      {seconds, ""} when seconds >= 0 -> {:ok, Map.put(filters, :maxage, seconds)}
+      _invalid -> :error
+    end
+  end
+
+  defp addFilter("SKIPME", value, filters) do
+    case String.downcase(value) do
+      "yes" -> {:ok, Map.put(filters, :skipme, true)}
+      "no" -> {:ok, Map.put(filters, :skipme, false)}
+      _other -> :error
+    end
+  end
+
+  defp buildMatcher(filters) do
+    caller = self()
+    moment = now()
+
+    fn client ->
+      matchesId?(filters, client) and matchesAddr?(filters, client) and
+        matchesLaddr?(filters, client) and matchesType?(filters, client) and
+        matchesAge?(filters, client, moment) and matchesSkipme?(filters, client, caller)
+    end
+  end
+
+  defp matchesId?(%{id: id}, client), do: client.id == id
+  defp matchesId?(_filters, _client), do: true
+
+  defp matchesAddr?(%{addr: addr}, client), do: client.addr == addr
+  defp matchesAddr?(_filters, _client), do: true
+
+  defp matchesLaddr?(%{laddr: laddr}, client), do: client.laddr == laddr
+  defp matchesLaddr?(_filters, _client), do: true
+
+  defp matchesType?(%{type: "normal"}, client), do: Clients.subscriptionCount(client.pid) == 0
+  defp matchesType?(%{type: "pubsub"}, client), do: Clients.subscriptionCount(client.pid) > 0
+  defp matchesType?(%{type: _other}, _client), do: false
+  defp matchesType?(_filters, _client), do: true
+
+  defp matchesAge?(%{maxage: maxage}, client, moment),
+    do: elapsed(moment, client.connectedAt) >= maxage
+
+  defp matchesAge?(_filters, _client, _moment), do: true
+
+  defp matchesSkipme?(%{skipme: true}, client, caller), do: client.pid != caller
+  defp matchesSkipme?(_filters, _client, _caller), do: true
+
+  defp parseId(raw) do
+    case Integer.parse(raw) do
+      {id, ""} when id > 0 -> id
+      _invalid -> :error
+    end
+  end
+
+  defp validName?(name), do: not String.contains?(name, [" ", "\n", "\r"])
+
+  defp elapsed(moment, since), do: div(moment - since, 1000)
+
+  defp now, do: System.system_time(:millisecond)
+end