Allow consuming from wildcard-topics

lib/kaffe/config/consumer.ex

@@ -28,7 +28,21 @@ defmodule Kaffe.Config.Consumer do
 
   def subscriber_name(idx), do: config_get(idx, :subscriber_name, consumer_group(idx)) |> String.to_atom()
 
-  def topics(idx), do: config_get!(idx, :topics)
+  def topics(idx) do
+    {wildcard_topics, specified_topics} = Enum.split_with(config_get!(idx, :topics), &(String.starts_with?(&1, "*")))
+    specified_topics ++ specify_wildcard_topics(wildcard_topics, idx)
+  end
+
+  defp specify_wildcard_topics([], _), do: []
+
+  defp specify_wildcard_topics(topics, idx) do
+    metadata = :brod.get_metadata(endpoints(idx))
+    topic_names = metadata |> elem(1) |> Map.get(:topics) |> Enum.map(& &1[:name])
+
+    Enum.flat_map(topics, fn("*" <> topic_name_tail) ->
+      Enum.filter(topic_names, &(String.ends_with?(&1, topic_name_tail)))
+    end) |> IO.inspect(label: "specify_wildcard_topics")
+  end
 
   def message_handler(idx), do: config_get!(idx, :message_handler)