.

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

lib/pleroma/notification.ex

@@ -2,6 +2,8 @@ defmodule Pleroma.Notification do
   use Ecto.Schema
   alias Pleroma.{User, Activity, Notification, Repo}
   import Ecto.Query
+  alias Pleroma.Web.Streamer
+  alias Pleroma.Web.MastodonAPI.NotificationView
 
   schema "notifications" do
     field :seen, :boolean, default: false
@@ -46,8 +48,14 @@ defmodule Pleroma.Notification do
 
   # TODO move to sql, too.
   def create_notification(%Activity{} = activity, %User{} = user) do
-    notification = %Notification{user_id: user.id, activity_id: activity.id}
+    notification = %Notification{user: user, activity: activity}
     {:ok, notification} = Repo.insert(notification)
+
+    json = NotificationView.render("notification.json", %{notification: notification})
+    |> Poison.encode!
+
+    Streamer.stream(user, %{type: "notification", payload: json})
+
     notification
   end
 end

lib/pleroma/web/mastodon_api/mastodon_api_controller.ex

@@ -3,10 +3,11 @@ defmodule Pleroma.Web.MastodonAPI.MastodonAPIController do
   alias Pleroma.{Repo, Activity, User, Notification}
   alias Pleroma.Web.OAuth.App
   alias Pleroma.Web
-  alias Pleroma.Web.MastodonAPI.{StatusView, AccountView}
+  alias Pleroma.Web.MastodonAPI.{StatusView, AccountView, NotificationView}
   alias Pleroma.Web.ActivityPub.ActivityPub
   alias Pleroma.Web.TwitterAPI.TwitterAPI
   alias Pleroma.Web.CommonAPI
+  alias Pleroma.Web.Streamer
   import Ecto.Query
   import Logger
 
@@ -167,23 +168,8 @@ defmodule Pleroma.Web.MastodonAPI.MastodonAPIController do
 
   def notifications(%{assigns: %{user: user}} = conn, params) do
     notifications = Notification.for_user(user, params)
-    result = Enum.map(notifications, fn (%{id: id, activity: activity, inserted_at: created_at}) ->
-      actor = User.get_cached_by_ap_id(activity.data["actor"])
-      created_at = NaiveDateTime.to_iso8601(created_at)
-      |> String.replace(~r/(\.\d+)?$/, ".000Z", global: false)
-      case activity.data["type"] do
-        "Create" ->
-          %{id: id, type: "mention", created_at: created_at, account: AccountView.render("account.json", %{user: actor}), status: StatusView.render("status.json", %{activity: activity})}
-        "Like" ->
-          liked_activity = Activity.get_create_activity_by_object_ap_id(activity.data["object"])
-          %{id: id, type: "favourite", created_at: created_at, account: AccountView.render("account.json", %{user: actor}), status: StatusView.render("status.json", %{activity: liked_activity})}
-        "Announce" ->
-          announced_activity = Activity.get_create_activity_by_object_ap_id(activity.data["object"])
-          %{id: id, type: "reblog", created_at: created_at, account: AccountView.render("account.json", %{user: actor}), status: StatusView.render("status.json", %{activity: announced_activity})}
-        "Follow" ->
-          %{id: id, type: "follow", created_at: created_at, account: AccountView.render("account.json", %{user: actor})}
-        _ -> nil
-      end
+    result = Enum.map(notifications, fn (notification) ->
+      NotificationView.render("notification.json", %{notification: notification})
     end)
     |> Enum.filter(&(&1))
 
@@ -280,6 +266,12 @@ defmodule Pleroma.Web.MastodonAPI.MastodonAPIController do
     end
   end
 
+  def stream_user(%{assigns: %{user: user}} = conn, _params) do
+    Streamer.add_conn(user, conn)
+    receive do
+    end
+  end
+
   def relationship_noop(%{assigns: %{user: user}} = conn, %{"id" => id}) do
     Logger.debug("Unimplemented, returning unmodified relationship")
     with %User{} = target <- Repo.get(User, id) do

lib/pleroma/web/mastodon_api/views/notification_view.ex

@@ -0,0 +1,27 @@
+defmodule Pleroma.Web.MastodonAPI.NotificationView do
+  use Pleroma.Web, :view
+  alias Pleroma.{User, Activity}
+  alias Pleroma.Web.MastodonAPI.{AccountView, StatusView}
+
+  def render("notification.json", %{notification: notification}) do
+    id = notification.id
+    activity = notification.activity
+    created_at = notification.inserted_at
+    actor = User.get_cached_by_ap_id(activity.data["actor"])
+    created_at = NaiveDateTime.to_iso8601(created_at)
+    |> String.replace(~r/(\.\d+)?$/, ".000Z", global: false)
+    case activity.data["type"] do
+      "Create" ->
+        %{id: id, type: "mention", created_at: created_at, account: AccountView.render("account.json", %{user: actor}), status: StatusView.render("status.json", %{activity: activity})}
+      "Like" ->
+        liked_activity = Activity.get_create_activity_by_object_ap_id(activity.data["object"])
+        %{id: id, type: "favourite", created_at: created_at, account: AccountView.render("account.json", %{user: actor}), status: StatusView.render("status.json", %{activity: liked_activity})}
+      "Announce" ->
+        announced_activity = Activity.get_create_activity_by_object_ap_id(activity.data["object"])
+        %{id: id, type: "reblog", created_at: created_at, account: AccountView.render("account.json", %{user: actor}), status: StatusView.render("status.json", %{activity: announced_activity})}
+      "Follow" ->
+        %{id: id, type: "follow", created_at: created_at, account: AccountView.render("account.json", %{user: actor})}
+      _ -> nil
+    end
+  end
+end

lib/pleroma/web/router.ex

@@ -42,6 +42,8 @@ defmodule Pleroma.Web.Router do
   scope "/api/v1", Pleroma.Web.MastodonAPI do
     pipe_through :authenticated_api
 
+    get "/streaming/user", MastodonAPIController, :stream_user
+
     get "/accounts/verify_credentials", MastodonAPIController, :verify_credentials
     get "/accounts/relationships", MastodonAPIController, :relationships
     post "/accounts/:id/follow", MastodonAPIController, :follow

lib/pleroma/web/streamer/streamer.ex

@@ -0,0 +1,40 @@
+defmodule Pleroma.Web.Streamer do
+  use GenServer
+  require Logger
+  import Plug.Conn
+
+  def start_link do
+    GenServer.start_link(__MODULE__, %{}, name: __MODULE__)
+  end
+
+  def add_conn(user, conn) do
+    GenServer.cast(__MODULE__, %{action: :add, user: user, conn: conn})
+  end
+
+  def stream(user, item) do
+    GenServer.cast(__MODULE__, %{action: :stream, user: user, item: item})
+  end
+
+  def handle_cast(%{action: :stream, user: user, item: item}, users) do
+    if conn = users[user.id] do
+      Logger.debug("Pushing item to #{user.id}, #{user.nickname}")
+      chunk(conn, "event: #{item.type}\ndata: #{item.payload}\n\n")
+    end
+    {:noreply, users}
+  end
+
+  def handle_cast(%{action: :add, user: user, conn: conn}, users) do
+    conn = conn
+    |> put_resp_header("content-type", "text/event-stream")
+    |> send_chunked(200)
+
+    users = Map.put(users, user.id, conn)
+    Logger.debug("Got new conn for user #{user.id}, #{user.nickname}")
+    {:noreply, users}
+  end
+
+  def handle_cast(m, state) do
+    IO.inspect("Unknown: #{inspect(m)}, #{inspect(state)}")
+    {:noreply, state}
+  end
+end

lib/pleroma/web/twitter_api/twitter_api.ex

@@ -5,6 +5,8 @@ defmodule Pleroma.Web.TwitterAPI.TwitterAPI do
   alias Pleroma.Web.TwitterAPI.UserView
   alias Pleroma.Web.{OStatus, CommonAPI}
   alias Pleroma.Formatter
+  alias Pleroma.Web.Streamer
+  alias Pleroma.Web.MastodonAPI.StatusView
 
   import Pleroma.Web.TwitterAPI.Utils
 
@@ -40,6 +42,10 @@ defmodule Pleroma.Web.TwitterAPI.TwitterAPI do
          tags <- Formatter.parse_tags(status),
          object <- make_note_data(user.ap_id, to, context, content_html, attachments, inReplyTo, tags) do
       res = ActivityPub.create(to, user, context, object)
+      {:ok, activity} = res
+      json = StatusView.render("status.json", activity: activity)
+      |> Poison.encode!
+      Streamer.stream(user, %{type: "update", payload: json})
       User.update_note_count(user)
       res
     end