make sync more robust, and send messages with connection status

1 files changed, 39 insertions, 18 deletions

diff --git a/lib/polyjuice/client/sync.ex b/lib/polyjuice/client/sync.ex
index b97ea02..ac81568 100644
--- a/lib/polyjuice/client/sync.ex
+++ b/lib/polyjuice/client/sync.ex
@@ -39,7 +39,8 @@ defmodule Polyjuice.Client.Sync do
     :since,
     query_params: "",
     backoff: nil,
-    set_filter: nil
+    set_filter: nil,
+    initial_done: false,
   ]
 
   @sync_path "_matrix/client/r0/sync"
@@ -121,11 +122,12 @@ defmodule Polyjuice.Client.Sync do
          ) do
       {:ok, conn_ref} ->
         Logger.info("Connected to sync")
+        send(state.listener, {:connected})
 
         if state.set_filter do
           set_filter(%{state | conn_ref: conn_ref, backoff: nil})
         else
-          do_sync(%{state | conn_ref: conn_ref, backoff: nil, set_filter: nil})
+          do_sync(%{state | conn_ref: conn_ref, backoff: nil})
         end
 
       # FIXME: what errors do we need to handle differently?
@@ -163,15 +165,24 @@ defmodule Polyjuice.Client.Sync do
         case status_code do
           200 ->
             {:ok, body} = :hackney.body(client_ref)
-            filter_id = body |> Poison.decode!() |> Map.get("filter_id")
-            Logger.debug("got filter id #{filter_id}")
-            Polyjuice.Client.Storage.set_filter_id(state.storage, state.set_filter, filter_id)
-
-            do_sync(%{
-              state
-              | query_params: "#{state.query_params}&filter=#{e.(filter_id)}",
-                set_filter: nil
-            })
+            with {:ok, %{} = json_body} <- Poison.decode(body),
+                 filter_id = Map.get(json_body, "filter_id") do
+              Logger.debug("got filter id #{filter_id}")
+              Polyjuice.Client.Storage.set_filter_id(
+                state.storage, state.set_filter, filter_id
+              )
+
+              do_sync(%{
+                state
+                | query_params: "#{state.query_params}&filter=#{e.(filter_id)}",
+                  set_filter: nil
+              })
+            else
+              _ ->
+                backoff = calc_backoff(state.backoff)
+                Logger.error("Server sent us garbage; retrying in #{backoff} seconds")
+                set_filter(%{state | backoff: backoff})
+            end
 
           _ ->
             {:ok, body} = :hackney.body(client_ref)
@@ -188,6 +199,7 @@ defmodule Polyjuice.Client.Sync do
         backoff = calc_backoff(state.backoff)
         Logger.error("Set filter error: closed; retrying in #{backoff} seconds.")
         connect(%{state | backoff: backoff, conn_ref: nil})
+        send(state.listener, {:disconnected})
 
       # FIXME: what other error codes do we need to handle?
       {:error, err} ->
@@ -215,14 +227,22 @@ defmodule Polyjuice.Client.Sync do
       {:ok, status_code, _resp_headers, client_ref} ->
         case status_code do
           200 ->
-            if state.backoff, do: Logger.info("Sync resumed")
             {:ok, body} = :hackney.body(client_ref)
-            Logger.debug(body)
-            json_body = Poison.decode!(body)
-            process_body(json_body, state)
-            %{"next_batch" => next_batch} = json_body
-            Polyjuice.Client.Storage.set_sync_token(state.storage, next_batch)
-            do_sync(%{state | since: next_batch, backoff: nil})
+            with {:ok, json_body} <- Poison.decode(body),
+                 %{"next_batch" => next_batch} <- json_body do
+              if state.backoff, do: Logger.info("Sync resumed")
+              process_body(json_body, state)
+              Polyjuice.Client.Storage.set_sync_token(state.storage, next_batch)
+              if not state.initial_done do
+                send(state.listener, {:initial_sync_completed})
+              end
+              do_sync(%{state | since: next_batch, backoff: nil, initial_done: true})
+            else
+              _ ->
+                backoff = calc_backoff(state.backoff)
+                Logger.error("Server sent us garbage; retrying in #{backoff} seconds")
+                do_sync(%{state | backoff: backoff})
+            end
 
           # FIXME: other status codes/error messages
           _ ->
@@ -240,6 +260,7 @@ defmodule Polyjuice.Client.Sync do
         backoff = calc_backoff(state.backoff)
         Logger.error("Sync error: closed; retrying in #{backoff} seconds.")
         connect(%{state | backoff: backoff, conn_ref: nil})
+        send(state.listener, {:disconnected})
 
       # FIXME: what other error codes do we need to handle?
       {:error, err} ->