6 files changed, 117 insertions, 2 deletions
diff --git a/lib/pleroma/application.ex b/lib/pleroma/application.ex
index 1f0a05568..5422cbc28 100644
--- a/lib/pleroma/application.ex
+++ b/lib/pleroma/application.ex
@@ -19,7 +19,8 @@ defmodule Pleroma.Application do
                          ttl_interval: 1000,
                          limit: 2500
                        ]]),
-      worker(Pleroma.Web.Federator, [])
+      worker(Pleroma.Web.Federator, []),
+      worker(Pleroma.Web.Streamer, [])
     ]
 
     # See http://elixir-lang.org/docs/stable/elixir/Supervisor.html
diff --git a/lib/pleroma/web/activity_pub/activity_pub.ex b/lib/pleroma/web/activity_pub/activity_pub.ex
index 1624c6545..35536a1e4 100644
--- a/lib/pleroma/web/activity_pub/activity_pub.ex
+++ b/lib/pleroma/web/activity_pub/activity_pub.ex
@@ -22,6 +22,9 @@ defmodule Pleroma.Web.ActivityPub.ActivityPub do
     with create_data <- make_create_data(%{to: to, actor: actor, published: published, context: context, object: object}, additional),
          {:ok, activity} <- insert(create_data, local),
          :ok <- maybe_federate(activity) do
+      if activity.data["type"] == "Create" and Enum.member?(activity.data["to"], "https://www.w3.org/ns/activitystreams#Public") do
+        Pleroma.Web.Streamer.stream("public", activity)
+      end
       {:ok, activity}
     end
   end
diff --git a/lib/pleroma/web/endpoint.ex b/lib/pleroma/web/endpoint.ex
index a1b4108cd..dc1ba2a05 100644
--- a/lib/pleroma/web/endpoint.ex
+++ b/lib/pleroma/web/endpoint.ex
@@ -2,6 +2,7 @@ defmodule Pleroma.Web.Endpoint do
   use Phoenix.Endpoint, otp_app: :pleroma
 
   socket "/socket", Pleroma.Web.UserSocket
+  socket "/api/v1", Pleroma.Web.MastodonAPI.MastodonSocket
 
   # Serve at "/" the static files from "priv/static" directory.
   #
diff --git a/lib/pleroma/web/mastodon_api/mastodon_socket.ex b/lib/pleroma/web/mastodon_api/mastodon_socket.ex
new file mode 100644
index 000000000..f9c8cec32
--- /dev/null
+++ b/lib/pleroma/web/mastodon_api/mastodon_socket.ex
@@ -0,0 +1,34 @@
+defmodule Pleroma.Web.MastodonAPI.MastodonSocket do
+  use Phoenix.Socket
+
+  transport :streaming, Phoenix.Transports.WebSocket.Raw,
+    timeout: :infinity # We never receive data.
+
+  def connect(params, socket) do
+    if params["stream"] == "public" do
+      socket = socket
+      |> assign(:topic, params["stream"])
+      Pleroma.Web.Streamer.add_socket(params["stream"], socket)
+      {:ok, socket}
+    else
+      :error
+    end
+  end
+
+  def id(socket), do: nil
+
+  def handle(:text, message, state) do
+    IO.inspect message
+    #| :ok
+    #| state
+    #| {:text, message}
+    #| {:text, message, state}
+    #| {:close, "Goodbye!"}
+    {:text, message}
+  end
+
+  def handle(:closed, reason, %{socket: socket}) do
+    topic = socket.assigns[:topic]
+    Pleroma.Web.Streamer.remove_socket(topic, socket)
+  end
+end
diff --git a/lib/pleroma/web/mastodon_api/views/status_view.ex b/lib/pleroma/web/mastodon_api/views/status_view.ex
index a0acdb0ac..d97b2acb4 100644
--- a/lib/pleroma/web/mastodon_api/views/status_view.ex
+++ b/lib/pleroma/web/mastodon_api/views/status_view.ex
@@ -46,7 +46,7 @@ defmodule Pleroma.Web.MastodonAPI.StatusView do
         website: nil
       },
       language: nil,
-      emoji: []
+      emojis: []
     }
   end
 
diff --git a/lib/pleroma/web/streamer.ex b/lib/pleroma/web/streamer.ex
new file mode 100644
index 000000000..3a7b91743
--- /dev/null
+++ b/lib/pleroma/web/streamer.ex
@@ -0,0 +1,76 @@
+defmodule Pleroma.Web.Streamer do
+  use GenServer
+  require Logger
+  import Plug.Conn
+
+  def start_link do
+    spawn(fn ->
+      Process.sleep(1000 * 30) # 30 seconds
+      GenServer.cast(__MODULE__, %{action: :ping})
+    end)
+    GenServer.start_link(__MODULE__, %{}, name: __MODULE__)
+  end
+
+  def add_socket(topic, socket) do
+    GenServer.cast(__MODULE__, %{action: :add, socket: socket, topic: topic})
+  end
+
+  def remove_socket(topic, socket) do
+    GenServer.cast(__MODULE__, %{action: :remove, socket: socket, topic: topic})
+  end
+
+  def stream(topic, item) do
+    GenServer.cast(__MODULE__, %{action: :stream, topic: topic, item: item})
+  end
+
+  def handle_cast(%{action: :ping}, topics) do
+    Map.values(topics)
+    |> List.flatten
+    |> Enum.each(fn (socket) ->
+      Logger.debug("Sending keepalive ping")
+      send socket.transport_pid, {:text, ""}
+    end)
+    spawn(fn ->
+      Process.sleep(1000 * 30) # 30 seconds
+      GenServer.cast(__MODULE__, %{action: :ping})
+    end)
+    {:noreply, topics}
+  end
+
+  def handle_cast(%{action: :stream, topic: topic, item: item}, topics) do
+    Logger.debug("Trying to push to #{topic}")
+    Logger.debug("Pushing item to #{topic}")
+    Enum.each(topics[topic] || [], fn (socket) ->
+      json = %{
+        event: "update",
+        payload: Pleroma.Web.MastodonAPI.StatusView.render("status.json", activity: item) |> Poison.encode!
+      } |> Poison.encode!
+
+      send socket.transport_pid, {:text, json}
+    end)
+    {:noreply, topics}
+  end
+
+  def handle_cast(%{action: :add, topic: topic, socket: socket}, sockets) do
+    sockets_for_topic = sockets[topic] || []
+    sockets_for_topic = Enum.uniq([socket | sockets_for_topic])
+    sockets = Map.put(sockets, topic, sockets_for_topic)
+    Logger.debug("Got new conn for #{topic}")
+    IO.inspect(sockets)
+    {:noreply, sockets}
+  end
+
+  def handle_cast(%{action: :remove, topic: topic, socket: socket}, sockets) do
+    sockets_for_topic = sockets[topic] || []
+    sockets_for_topic = List.delete(sockets_for_topic, socket)
+    sockets = Map.put(sockets, topic, sockets_for_topic)
+    Logger.debug("Removed conn for #{topic}")
+    IO.inspect(sockets)
+    {:noreply, sockets}
+  end
+
+  def handle_cast(m, state) do
+    IO.inspect("Unknown: #{inspect(m)}, #{inspect(state)}")
+    {:noreply, state}
+  end
+end