Merge branch 'streamer-refactoring' into 'develop'

Streamer refactoring

See merge request pleroma/pleroma!1653

9 files changed, 289 insertions, 86 deletions

diff --git a/test/activity/ir/topics_test.exs b/test/activity/ir/topics_test.exs
new file mode 100644
index 000000000..e75f83586
--- /dev/null
+++ b/test/activity/ir/topics_test.exs
@@ -0,0 +1,141 @@
+defmodule Pleroma.Activity.Ir.TopicsTest do
+  use Pleroma.DataCase
+
+  alias Pleroma.Activity
+  alias Pleroma.Activity.Ir.Topics
+  alias Pleroma.Object
+
+  require Pleroma.Constants
+
+  describe "poll answer" do
+    test "produce no topics" do
+      activity = %Activity{object: %Object{data: %{"type" => "Answer"}}}
+
+      assert [] == Topics.get_activity_topics(activity)
+    end
+  end
+
+  describe "non poll answer" do
+    test "always add user and list topics" do
+      activity = %Activity{object: %Object{data: %{"type" => "FooBar"}}}
+      topics = Topics.get_activity_topics(activity)
+
+      assert Enum.member?(topics, "user")
+      assert Enum.member?(topics, "list")
+    end
+  end
+
+  describe "public visibility" do
+    setup do
+      activity = %Activity{
+        object: %Object{data: %{"type" => "Note"}},
+        data: %{"to" => [Pleroma.Constants.as_public()]}
+      }
+
+      {:ok, activity: activity}
+    end
+
+    test "produces public topic", %{activity: activity} do
+      topics = Topics.get_activity_topics(activity)
+
+      assert Enum.member?(topics, "public")
+    end
+
+    test "local action produces public:local topic", %{activity: activity} do
+      activity = %{activity | local: true}
+      topics = Topics.get_activity_topics(activity)
+
+      assert Enum.member?(topics, "public:local")
+    end
+
+    test "non-local action does not produce public:local topic", %{activity: activity} do
+      activity = %{activity | local: false}
+      topics = Topics.get_activity_topics(activity)
+
+      refute Enum.member?(topics, "public:local")
+    end
+  end
+
+  describe "public visibility create events" do
+    setup do
+      activity = %Activity{
+        object: %Object{data: %{"type" => "Create", "attachment" => []}},
+        data: %{"to" => [Pleroma.Constants.as_public()]}
+      }
+
+      {:ok, activity: activity}
+    end
+
+    test "with no attachments doesn't produce public:media topics", %{activity: activity} do
+      topics = Topics.get_activity_topics(activity)
+
+      refute Enum.member?(topics, "public:media")
+      refute Enum.member?(topics, "public:local:media")
+    end
+
+    test "converts tags to hash tags", %{activity: %{object: %{data: data} = object} = activity} do
+      tagged_data = Map.put(data, "tag", ["foo", "bar"])
+      activity = %{activity | object: %{object | data: tagged_data}}
+
+      topics = Topics.get_activity_topics(activity)
+
+      assert Enum.member?(topics, "hashtag:foo")
+      assert Enum.member?(topics, "hashtag:bar")
+    end
+
+    test "only converts strinngs to hash tags", %{
+      activity: %{object: %{data: data} = object} = activity
+    } do
+      tagged_data = Map.put(data, "tag", [2])
+      activity = %{activity | object: %{object | data: tagged_data}}
+
+      topics = Topics.get_activity_topics(activity)
+
+      refute Enum.member?(topics, "hashtag:2")
+    end
+  end
+
+  describe "public visibility create events with attachments" do
+    setup do
+      activity = %Activity{
+        object: %Object{data: %{"type" => "Create", "attachment" => ["foo"]}},
+        data: %{"to" => [Pleroma.Constants.as_public()]}
+      }
+
+      {:ok, activity: activity}
+    end
+
+    test "produce public:media topics", %{activity: activity} do
+      topics = Topics.get_activity_topics(activity)
+
+      assert Enum.member?(topics, "public:media")
+    end
+
+    test "local produces public:local:media topics", %{activity: activity} do
+      topics = Topics.get_activity_topics(activity)
+
+      assert Enum.member?(topics, "public:local:media")
+    end
+
+    test "non-local doesn't produce public:local:media topics", %{activity: activity} do
+      activity = %{activity | local: false}
+
+      topics = Topics.get_activity_topics(activity)
+
+      refute Enum.member?(topics, "public:local:media")
+    end
+  end
+
+  describe "non-public visibility" do
+    test "produces direct topic" do
+      activity = %Activity{object: %Object{data: %{"type" => "Note"}}, data: %{"to" => []}}
+      topics = Topics.get_activity_topics(activity)
+
+      assert Enum.member?(topics, "direct")
+      refute Enum.member?(topics, "public")
+      refute Enum.member?(topics, "public:local")
+      refute Enum.member?(topics, "public:media")
+      refute Enum.member?(topics, "public:local:media")
+    end
+  end
+end
diff --git a/test/integration/mastodon_websocket_test.exs b/test/integration/mastodon_websocket_test.exs
index 63bf73412..c04262808 100644
--- a/test/integration/mastodon_websocket_test.exs
+++ b/test/integration/mastodon_websocket_test.exs
@@ -11,7 +11,6 @@ defmodule Pleroma.Integration.MastodonWebsocketTest do
   alias Pleroma.Integration.WebsocketClient
   alias Pleroma.Web.CommonAPI
   alias Pleroma.Web.OAuth
-  alias Pleroma.Web.Streamer
 
   @path Pleroma.Web.Endpoint.url()
         |> URI.parse()
@@ -19,16 +18,6 @@ defmodule Pleroma.Integration.MastodonWebsocketTest do
         |> Map.put(:path, "/api/v1/streaming")
         |> URI.to_string()
 
-  setup do
-    GenServer.start(Streamer, %{}, name: Streamer)
-
-    on_exit(fn ->
-      if pid = Process.whereis(Streamer) do
-        Process.exit(pid, :kill)
-      end
-    end)
-  end
-
   def start_socket(qs \\ nil, headers \\ []) do
     path =
       case qs do
@@ -53,12 +42,14 @@ defmodule Pleroma.Integration.MastodonWebsocketTest do
     end)
   end
 
+  @tag needs_streamer: true
   test "allows public streams without authentication" do
     assert {:ok, _} = start_socket("?stream=public")
     assert {:ok, _} = start_socket("?stream=public:local")
     assert {:ok, _} = start_socket("?stream=hashtag&tag=lain")
   end
 
+  @tag needs_streamer: true
   test "receives well formatted events" do
     user = insert(:user)
     {:ok, _} = start_socket("?stream=public")
@@ -103,6 +94,7 @@ defmodule Pleroma.Integration.MastodonWebsocketTest do
       assert {:ok, _} = start_socket("?stream=user&access_token=#{state.token.token}")
     end
 
+    @tag needs_streamer: true
     test "accepts the 'user' stream", %{token: token} = _state do
       assert {:ok, _} = start_socket("?stream=user&access_token=#{token.token}")
 
@@ -111,6 +103,7 @@ defmodule Pleroma.Integration.MastodonWebsocketTest do
              end) =~ ":badarg"
     end
 
+    @tag needs_streamer: true
     test "accepts the 'user:notification' stream", %{token: token} = _state do
       assert {:ok, _} = start_socket("?stream=user:notification&access_token=#{token.token}")
 
@@ -119,6 +112,7 @@ defmodule Pleroma.Integration.MastodonWebsocketTest do
              end) =~ ":badarg"
     end
 
+    @tag needs_streamer: true
     test "accepts valid token on Sec-WebSocket-Protocol header", %{token: token} do
       assert {:ok, _} = start_socket("?stream=user", [{"Sec-WebSocket-Protocol", token.token}])
 
diff --git a/test/notification_test.exs b/test/notification_test.exs
index 3be9db09b..3d2f9a8fc 100644
--- a/test/notification_test.exs
+++ b/test/notification_test.exs
@@ -69,16 +69,7 @@ defmodule Pleroma.NotificationTest do
   end
 
   describe "create_notification" do
-    setup do
-      GenServer.start(Streamer, %{}, name: Streamer)
-
-      on_exit(fn ->
-        if pid = Process.whereis(Streamer) do
-          Process.exit(pid, :kill)
-        end
-      end)
-    end
-
+    @tag needs_streamer: true
     test "it creates a notification for user and send to the 'user' and the 'user:notification' stream" do
       user = insert(:user)
       task = Task.async(fn -> assert_receive {:text, _}, 4_000 end)
diff --git a/test/support/conn_case.ex b/test/support/conn_case.ex
index ec5892ff5..b39c70677 100644
--- a/test/support/conn_case.ex
+++ b/test/support/conn_case.ex
@@ -40,6 +40,10 @@ defmodule Pleroma.Web.ConnCase do
       Ecto.Adapters.SQL.Sandbox.mode(Pleroma.Repo, {:shared, self()})
     end
 
+    if tags[:needs_streamer] do
+      start_supervised(Pleroma.Web.Streamer.supervisor())
+    end
+
     {:ok, conn: Phoenix.ConnTest.build_conn()}
   end
 end
diff --git a/test/support/data_case.ex b/test/support/data_case.ex
index f3d98e7e3..17fa15214 100644
--- a/test/support/data_case.ex
+++ b/test/support/data_case.ex
@@ -39,6 +39,10 @@ defmodule Pleroma.DataCase do
       Ecto.Adapters.SQL.Sandbox.mode(Pleroma.Repo, {:shared, self()})
     end
 
+    if tags[:needs_streamer] do
+      start_supervised(Pleroma.Web.Streamer.supervisor())
+    end
+
     :ok
   end
 
diff --git a/test/web/activity_pub/activity_pub_test.exs b/test/web/activity_pub/activity_pub_test.exs
index d0118fefa..4100108a5 100644
--- a/test/web/activity_pub/activity_pub_test.exs
+++ b/test/web/activity_pub/activity_pub_test.exs
@@ -38,9 +38,7 @@ defmodule Pleroma.Web.ActivityPub.ActivityPubTest do
         stream: fn _, _ -> nil end do
         ActivityPub.stream_out_participations(conversation.participations)
 
-        Enum.each(participations, fn participation ->
-          assert called(Pleroma.Web.Streamer.stream("participation", participation))
-        end)
+        assert called(Pleroma.Web.Streamer.stream("participation", participations))
       end
     end
   end
diff --git a/test/web/streamer/ping_test.exs b/test/web/streamer/ping_test.exs
new file mode 100644
index 000000000..3d52c00e4
--- /dev/null
+++ b/test/web/streamer/ping_test.exs
@@ -0,0 +1,36 @@
+# Pleroma: A lightweight social networking server
+# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
+# SPDX-License-Identifier: AGPL-3.0-only
+
+defmodule Pleroma.Web.PingTest do
+  use Pleroma.DataCase
+
+  import Pleroma.Factory
+  alias Pleroma.Web.Streamer
+
+  setup do
+    start_supervised({Streamer.supervisor(), [ping_interval: 30]})
+
+    :ok
+  end
+
+  describe "sockets" do
+    setup do
+      user = insert(:user)
+      {:ok, %{user: user}}
+    end
+
+    test "it sends pings", %{user: user} do
+      task =
+        Task.async(fn ->
+          assert_receive {:text, received_event}, 40
+          assert_receive {:text, received_event}, 40
+          assert_receive {:text, received_event}, 40
+        end)
+
+      Streamer.add_socket("public", %{transport_pid: task.pid, assigns: %{user: user}})
+
+      Task.await(task)
+    end
+  end
+end
diff --git a/test/web/streamer/state_test.exs b/test/web/streamer/state_test.exs
new file mode 100644
index 000000000..d1aeac541
--- /dev/null
+++ b/test/web/streamer/state_test.exs
@@ -0,0 +1,54 @@
+# Pleroma: A lightweight social networking server
+# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
+# SPDX-License-Identifier: AGPL-3.0-only
+
+defmodule Pleroma.Web.StateTest do
+  use Pleroma.DataCase
+
+  import Pleroma.Factory
+  alias Pleroma.Web.Streamer
+  alias Pleroma.Web.Streamer.StreamerSocket
+
+  @moduletag needs_streamer: true
+
+  describe "sockets" do
+    setup do
+      user = insert(:user)
+      user2 = insert(:user)
+      {:ok, %{user: user, user2: user2}}
+    end
+
+    test "it can add a socket", %{user: user} do
+      Streamer.add_socket("public", %{transport_pid: 1, assigns: %{user: user}})
+
+      assert(%{"public" => [%StreamerSocket{transport_pid: 1}]} = Streamer.get_sockets())
+    end
+
+    test "it can add multiple sockets per user", %{user: user} do
+      Streamer.add_socket("public", %{transport_pid: 1, assigns: %{user: user}})
+      Streamer.add_socket("public", %{transport_pid: 2, assigns: %{user: user}})
+
+      assert(
+        %{
+          "public" => [
+            %StreamerSocket{transport_pid: 2},
+            %StreamerSocket{transport_pid: 1}
+          ]
+        } = Streamer.get_sockets()
+      )
+    end
+
+    test "it will not add a duplicate socket", %{user: user} do
+      Streamer.add_socket("activity", %{transport_pid: 1, assigns: %{user: user}})
+      Streamer.add_socket("activity", %{transport_pid: 1, assigns: %{user: user}})
+
+      assert(
+        %{
+          "activity" => [
+            %StreamerSocket{transport_pid: 1}
+          ]
+        } = Streamer.get_sockets()
+      )
+    end
+  end
+end
diff --git a/test/web/streamer_test.exs b/test/web/streamer/streamer_test.exs
index 96fa7645f..88847e20f 100644
--- a/test/web/streamer_test.exs
+++ b/test/web/streamer/streamer_test.exs
@@ -5,24 +5,20 @@
 defmodule Pleroma.Web.StreamerTest do
   use Pleroma.DataCase
 
+  import Pleroma.Factory
+
   alias Pleroma.List
   alias Pleroma.User
   alias Pleroma.Web.CommonAPI
   alias Pleroma.Web.Streamer
-  import Pleroma.Factory
+  alias Pleroma.Web.Streamer.StreamerSocket
+  alias Pleroma.Web.Streamer.Worker
 
+  @moduletag needs_streamer: true
   clear_config_all([:instance, :skip_thread_containment])
 
   describe "user streams" do
     setup do
-      GenServer.start(Streamer, %{}, name: Streamer)
-
-      on_exit(fn ->
-        if pid = Process.whereis(Streamer) do
-          Process.exit(pid, :kill)
-        end
-      end)
-
       user = insert(:user)
       notify = insert(:notification, user: user, activity: build(:note_activity))
       {:ok, %{user: user, notify: notify}}
@@ -125,11 +121,9 @@ defmodule Pleroma.Web.StreamerTest do
         assert_receive {:text, _}, 4_000
       end)
 
-    fake_socket = %{
+    fake_socket = %StreamerSocket{
       transport_pid: task.pid,
-      assigns: %{
-        user: user
-      }
+      user: user
     }
 
     {:ok, activity} = CommonAPI.post(other_user, %{"status" => "Test"})
@@ -138,7 +132,7 @@ defmodule Pleroma.Web.StreamerTest do
       "public" => [fake_socket]
     }
 
-    Streamer.push_to_socket(topics, "public", activity)
+    Worker.push_to_socket(topics, "public", activity)
 
     Task.await(task)
 
@@ -155,11 +149,9 @@ defmodule Pleroma.Web.StreamerTest do
         assert received_event == expected_event
       end)
 
-    fake_socket = %{
+    fake_socket = %StreamerSocket{
       transport_pid: task.pid,
-      assigns: %{
-        user: user
-      }
+      user: user
     }
 
     {:ok, activity} = CommonAPI.delete(activity.id, other_user)
@@ -168,7 +160,7 @@ defmodule Pleroma.Web.StreamerTest do
       "public" => [fake_socket]
     }
 
-    Streamer.push_to_socket(topics, "public", activity)
+    Worker.push_to_socket(topics, "public", activity)
 
     Task.await(task)
   end
@@ -189,9 +181,9 @@ defmodule Pleroma.Web.StreamerTest do
         )
 
       task = Task.async(fn -> refute_receive {:text, _}, 1_000 end)
-      fake_socket = %{transport_pid: task.pid, assigns: %{user: user}}
+      fake_socket = %StreamerSocket{transport_pid: task.pid, user: user}
       topics = %{"public" => [fake_socket]}
-      Streamer.push_to_socket(topics, "public", activity)
+      Worker.push_to_socket(topics, "public", activity)
 
       Task.await(task)
     end
@@ -211,9 +203,9 @@ defmodule Pleroma.Web.StreamerTest do
         )
 
       task = Task.async(fn -> assert_receive {:text, _}, 1_000 end)
-      fake_socket = %{transport_pid: task.pid, assigns: %{user: user}}
+      fake_socket = %StreamerSocket{transport_pid: task.pid, user: user}
       topics = %{"public" => [fake_socket]}
-      Streamer.push_to_socket(topics, "public", activity)
+      Worker.push_to_socket(topics, "public", activity)
 
       Task.await(task)
     end
@@ -233,9 +225,9 @@ defmodule Pleroma.Web.StreamerTest do
         )
 
       task = Task.async(fn -> assert_receive {:text, _}, 1_000 end)
-      fake_socket = %{transport_pid: task.pid, assigns: %{user: user}}
+      fake_socket = %StreamerSocket{transport_pid: task.pid, user: user}
       topics = %{"public" => [fake_socket]}
-      Streamer.push_to_socket(topics, "public", activity)
+      Worker.push_to_socket(topics, "public", activity)
 
       Task.await(task)
     end
@@ -251,11 +243,9 @@ defmodule Pleroma.Web.StreamerTest do
         refute_receive {:text, _}, 1_000
       end)
 
-    fake_socket = %{
+    fake_socket = %StreamerSocket{
       transport_pid: task.pid,
-      assigns: %{
-        user: user
-      }
+      user: user
     }
 
     {:ok, activity} = CommonAPI.post(blocked_user, %{"status" => "Test"})
@@ -264,7 +254,7 @@ defmodule Pleroma.Web.StreamerTest do
       "public" => [fake_socket]
     }
 
-    Streamer.push_to_socket(topics, "public", activity)
+    Worker.push_to_socket(topics, "public", activity)
 
     Task.await(task)
   end
@@ -284,11 +274,9 @@ defmodule Pleroma.Web.StreamerTest do
         refute_receive {:text, _}, 1_000
       end)
 
-    fake_socket = %{
+    fake_socket = %StreamerSocket{
       transport_pid: task.pid,
-      assigns: %{
-        user: user_a
-      }
+      user: user_a
     }
 
     {:ok, activity} =
@@ -301,7 +289,7 @@ defmodule Pleroma.Web.StreamerTest do
       "list:#{list.id}" => [fake_socket]
     }
 
-    Streamer.handle_cast(%{action: :stream, topic: "list", item: activity}, topics)
+    Worker.handle_call({:stream, "list", activity}, self(), topics)
 
     Task.await(task)
   end
@@ -318,11 +306,9 @@ defmodule Pleroma.Web.StreamerTest do
         refute_receive {:text, _}, 1_000
       end)
 
-    fake_socket = %{
+    fake_socket = %StreamerSocket{
       transport_pid: task.pid,
-      assigns: %{
-        user: user_a
-      }
+      user: user_a
     }
 
     {:ok, activity} =
@@ -335,12 +321,12 @@ defmodule Pleroma.Web.StreamerTest do
       "list:#{list.id}" => [fake_socket]
     }
 
-    Streamer.handle_cast(%{action: :stream, topic: "list", item: activity}, topics)
+    Worker.handle_call({:stream, "list", activity}, self(), topics)
 
     Task.await(task)
   end
 
-  test "it send wanted private posts to list" do
+  test "it sends wanted private posts to list" do
     user_a = insert(:user)
     user_b = insert(:user)
 
@@ -354,11 +340,9 @@ defmodule Pleroma.Web.StreamerTest do
         assert_receive {:text, _}, 1_000
       end)
 
-    fake_socket = %{
+    fake_socket = %StreamerSocket{
       transport_pid: task.pid,
-      assigns: %{
-        user: user_a
-      }
+      user: user_a
     }
 
     {:ok, activity} =
@@ -367,11 +351,12 @@ defmodule Pleroma.Web.StreamerTest do
         "visibility" => "private"
       })
 
-    topics = %{
-      "list:#{list.id}" => [fake_socket]
-    }
+    Streamer.add_socket(
+      "list:#{list.id}",
+      fake_socket
+    )
 
-    Streamer.handle_cast(%{action: :stream, topic: "list", item: activity}, topics)
+    Worker.handle_call({:stream, "list", activity}, self(), %{})
 
     Task.await(task)
   end
@@ -387,11 +372,9 @@ defmodule Pleroma.Web.StreamerTest do
         refute_receive {:text, _}, 1_000
       end)
 
-    fake_socket = %{
+    fake_socket = %StreamerSocket{
       transport_pid: task.pid,
-      assigns: %{
-        user: user1
-      }
+      user: user1
     }
 
     {:ok, create_activity} = CommonAPI.post(user3, %{"status" => "I'm kawen"})
@@ -401,7 +384,7 @@ defmodule Pleroma.Web.StreamerTest do
       "public" => [fake_socket]
     }
 
-    Streamer.push_to_socket(topics, "public", announce_activity)
+    Worker.push_to_socket(topics, "public", announce_activity)
 
     Task.await(task)
   end
@@ -417,6 +400,8 @@ defmodule Pleroma.Web.StreamerTest do
 
     task = Task.async(fn -> refute_receive {:text, _}, 4_000 end)
 
+    Process.sleep(4000)
+
     Streamer.add_socket(
       "user",
       %{transport_pid: task.pid, assigns: %{user: user2}}
@@ -428,14 +413,6 @@ defmodule Pleroma.Web.StreamerTest do
 
   describe "direct streams" do
     setup do
-      GenServer.start(Streamer, %{}, name: Streamer)
-
-      on_exit(fn ->
-        if pid = Process.whereis(Streamer) do
-          Process.exit(pid, :kill)
-        end
-      end)
-
       :ok
     end
 
@@ -480,6 +457,8 @@ defmodule Pleroma.Web.StreamerTest do
           refute_receive {:text, _}, 4_000
         end)
 
+      Process.sleep(1000)
+
       Streamer.add_socket(
         "direct",
         %{transport_pid: task.pid, assigns: %{user: user}}
@@ -521,6 +500,8 @@ defmodule Pleroma.Web.StreamerTest do
           assert last_status["id"] == to_string(create_activity.id)
         end)
 
+      Process.sleep(1000)
+
       Streamer.add_socket(
         "direct",
         %{transport_pid: task.pid, assigns: %{user: user}}