Refresh subscriptions.
This commit is contained in:
parent
fca7390c69
commit
2e753e8cd7
|
@ -1,7 +1,7 @@
|
||||||
defmodule Pleroma.Web.Federator do
|
defmodule Pleroma.Web.Federator do
|
||||||
use GenServer
|
use GenServer
|
||||||
alias Pleroma.User
|
alias Pleroma.User
|
||||||
alias Pleroma.Web.WebFinger
|
alias Pleroma.Web.{WebFinger, Websub}
|
||||||
require Logger
|
require Logger
|
||||||
|
|
||||||
@websub Application.get_env(:pleroma, :websub)
|
@websub Application.get_env(:pleroma, :websub)
|
||||||
|
@ -9,14 +9,27 @@ defmodule Pleroma.Web.Federator do
|
||||||
@max_jobs 10
|
@max_jobs 10
|
||||||
|
|
||||||
def start_link do
|
def start_link do
|
||||||
|
spawn(fn ->
|
||||||
|
Process.sleep(1000 * 60 * 1) # 10 minutes
|
||||||
|
enqueue(:refresh_subscriptions, nil)
|
||||||
|
end)
|
||||||
GenServer.start_link(__MODULE__, {:sets.new(), :queue.new()}, name: __MODULE__)
|
GenServer.start_link(__MODULE__, {:sets.new(), :queue.new()}, name: __MODULE__)
|
||||||
end
|
end
|
||||||
|
|
||||||
|
def handle(:refresh_subscriptions, _) do
|
||||||
|
Logger.debug("Federator running refresh subscriptions")
|
||||||
|
Websub.refresh_subscriptions()
|
||||||
|
spawn(fn ->
|
||||||
|
Process.sleep(1000 * 60 * 60) # 60 minutes
|
||||||
|
enqueue(:refresh_subscriptions, nil)
|
||||||
|
end)
|
||||||
|
end
|
||||||
|
|
||||||
def handle(:publish, activity) do
|
def handle(:publish, activity) do
|
||||||
Logger.debug(fn -> "Running publish for #{activity.data["id"]}" end)
|
Logger.debug(fn -> "Running publish for #{activity.data["id"]}" end)
|
||||||
with actor when not is_nil(actor) <- User.get_cached_by_ap_id(activity.data["actor"]) do
|
with actor when not is_nil(actor) <- User.get_cached_by_ap_id(activity.data["actor"]) do
|
||||||
Logger.debug(fn -> "Sending #{activity.data["id"]} out via websub" end)
|
Logger.debug(fn -> "Sending #{activity.data["id"]} out via websub" end)
|
||||||
Pleroma.Web.Websub.publish(Pleroma.Web.OStatus.feed_path(actor), actor, activity)
|
Websub.publish(Pleroma.Web.OStatus.feed_path(actor), actor, activity)
|
||||||
|
|
||||||
{:ok, actor} = WebFinger.ensure_keys_present(actor)
|
{:ok, actor} = WebFinger.ensure_keys_present(actor)
|
||||||
Logger.debug(fn -> "Sending #{activity.data["id"]} out via salmon" end)
|
Logger.debug(fn -> "Sending #{activity.data["id"]} out via salmon" end)
|
||||||
|
|
|
@ -204,4 +204,19 @@ def request_subscription(websub, poster \\ &@httpoison.post/3, timeout \\ 10_000
|
||||||
{:error, websub}
|
{:error, websub}
|
||||||
end
|
end
|
||||||
end
|
end
|
||||||
|
|
||||||
|
def refresh_subscriptions(delta \\ 60 * 60 * 24) do
|
||||||
|
Logger.debug("Refreshing subscriptions")
|
||||||
|
|
||||||
|
cut_off = NaiveDateTime.add(NaiveDateTime.utc_now, delta)
|
||||||
|
|
||||||
|
query = from sub in WebsubClientSubscription,
|
||||||
|
where: sub.valid_until < ^cut_off and sub.state == "active"
|
||||||
|
|
||||||
|
subs = Repo.all(query)
|
||||||
|
|
||||||
|
Enum.map(subs, fn (sub) ->
|
||||||
|
request_subscription(sub)
|
||||||
|
end)
|
||||||
|
end
|
||||||
end
|
end
|
||||||
|
|
|
@ -7,7 +7,7 @@ def verify(sub) do
|
||||||
defmodule Pleroma.Web.WebsubTest do
|
defmodule Pleroma.Web.WebsubTest do
|
||||||
use Pleroma.DataCase
|
use Pleroma.DataCase
|
||||||
alias Pleroma.Web.Websub
|
alias Pleroma.Web.Websub
|
||||||
alias Pleroma.Web.Websub.WebsubServerSubscription
|
alias Pleroma.Web.Websub.{WebsubServerSubscription, WebsubClientSubscription}
|
||||||
import Pleroma.Factory
|
import Pleroma.Factory
|
||||||
alias Pleroma.Web.Router.Helpers
|
alias Pleroma.Web.Router.Helpers
|
||||||
|
|
||||||
|
@ -174,4 +174,18 @@ test "sign a text" do
|
||||||
|
|
||||||
signed = Websub.sign("secret", [["て"], ['す']])
|
signed = Websub.sign("secret", [["て"], ['す']])
|
||||||
end
|
end
|
||||||
|
|
||||||
|
describe "renewing subscriptions" do
|
||||||
|
test "it renews subscriptions that have less than a day of time left" do
|
||||||
|
day = 60 * 60 * 24
|
||||||
|
now = NaiveDateTime.utc_now
|
||||||
|
still_good = insert(:websub_client_subscription, %{valid_until: NaiveDateTime.add(now, 2 * day), topic: "http://example.org/still_good", state: "active"})
|
||||||
|
needs_refresh = insert(:websub_client_subscription, %{valid_until: NaiveDateTime.add(now, day - 100), topic: "http://example.org/needs_refresh", state: "active"})
|
||||||
|
|
||||||
|
refresh = Websub.refresh_subscriptions()
|
||||||
|
|
||||||
|
assert still_good == Repo.get(WebsubClientSubscription, still_good.id)
|
||||||
|
refute needs_refresh == Repo.get(WebsubClientSubscription, needs_refresh.id)
|
||||||
|
end
|
||||||
|
end
|
||||||
end
|
end
|
||||||
|
|
Loading…
Reference in a new issue