proxy-call: Add first pass at continuous call mode

3 files changed, 144 insertions, 0 deletions

diff --git a/rest/rest-proxy-call-private.h b/rest/rest-proxy-call-private.h
index 50a2a0b..584a77b 100644
--- a/rest/rest-proxy-call-private.h
+++ b/rest/rest-proxy-call-private.h
@@ -30,6 +30,7 @@
 G_BEGIN_DECLS
 
 typedef struct _RestProxyCallAsyncClosure RestProxyCallAsyncClosure;
+typedef struct _RestProxyCallContinuousClosure RestProxyCallContinuousClosure;
 
 struct _RestProxyCallPrivate {
   gchar *method;
diff --git a/rest/rest-proxy-call.c b/rest/rest-proxy-call.c
index 007fd25..c90c75b 100644
--- a/rest/rest-proxy-call.c
+++ b/rest/rest-proxy-call.c
@@ -41,6 +41,15 @@ struct _RestProxyCallAsyncClosure {
   SoupMessage *message;
 };
 
+
+struct _RestProxyCallContinuousClosure {
+  RestProxyCall *call;
+  RestProxyCallContinuousCallback callback;
+  GObject *weak_object;
+  gpointer userdata;
+  SoupMessage *message;
+};
+
 enum
 {
   PROP_0 = 0,
@@ -627,6 +636,49 @@ _call_message_completed_cb (SoupSession *session,
   g_slice_free (RestProxyCallAsyncClosure, closure);
 }
 
+
+static void
+_continuous_call_message_completed_cb (SoupSession *session,
+                                       SoupMessage *message,
+                                       gpointer     userdata)
+{
+  RestProxyCallContinuousClosure *closure;
+  RestProxyCall *call;
+  RestProxyCallPrivate *priv;
+  GError *error = NULL;
+
+  closure = (RestProxyCallContinuousClosure *)userdata;
+  call = closure->call;
+  priv = GET_PRIVATE (call);
+
+  priv->status_code = message->status_code;
+  priv->status_message = g_strdup (message->reason_phrase);
+
+  _handle_error_from_message (message, &error);
+
+  closure->callback (closure->call,
+                     NULL,
+                     0,
+                     error,
+                     closure->weak_object,
+                     closure->userdata);
+
+  g_clear_error (&error);
+
+  /* Success. We don't need the weak reference any more */
+  if (closure->weak_object)
+  {
+    g_object_weak_unref (closure->weak_object,
+        (GWeakNotify)_call_async_weak_notify_cb,
+        closure);
+  }
+
+  priv->cur_call_closure = NULL;
+  g_object_unref (closure->call);
+  g_slice_free (RestProxyCallContinuousClosure, closure);
+}
+
+
 static void
 _call_async_weak_notify_cb (gpointer *data,
                             GObject  *dead_object)
@@ -818,6 +870,83 @@ error:
   return FALSE;
 }
 
+static void
+_continuous_call_message_got_chunk_cb (SoupMessage                    *msg,
+                                       SoupBuffer                     *chunk,
+                                       RestProxyCallContinuousClosure *closure)
+{
+  closure->callback (closure->call,
+                     chunk->data,
+                     chunk->length,
+                     NULL,
+                     closure->weak_object,
+                     closure->userdata);
+}
+
+gboolean
+rest_proxy_call_continuous (RestProxyCall                    *call,
+                            RestProxyCallContinuousCallback   callback,
+                            GObject                          *weak_object,
+                            gpointer                          userdata,
+                            GError                          **error)
+{
+  RestProxyCallPrivate *priv;
+  RestProxyCallClass *call_class;
+  SoupMessage *message;
+  RestProxyCallContinuousClosure *closure;
+
+  g_return_val_if_fail (REST_IS_PROXY_CALL (call), FALSE);
+  priv = GET_PRIVATE (call);
+  g_assert (priv->proxy);
+  call_class = REST_PROXY_CALL_GET_CLASS (call);
+
+  if (priv->cur_call_closure)
+  {
+    /* FIXME: Use GError here */
+    g_critical (G_STRLOC ": Call already in progress.");
+    return FALSE;
+  }
+
+  message = prepare_message (call, error);
+  if (message == NULL)
+    goto error;
+
+  closure = g_slice_new0 (RestProxyCallContinuousClosure);
+  closure->call = g_object_ref (call);
+  closure->callback = callback;
+  closure->weak_object = weak_object;
+  closure->message = message;
+  closure->userdata = userdata;
+
+  priv->cur_call_closure = closure;
+
+  /* Weakly reference this object. We remove our callback if it goes away. */
+  if (closure->weak_object)
+  {
+    g_object_weak_ref (closure->weak_object,
+        (GWeakNotify)_call_async_weak_notify_cb,
+        closure);
+  }
+
+  g_signal_connect (message,
+                    "got-chunk",
+                    (GCallback)_continuous_call_message_got_chunk_cb,
+                    closure);
+
+  _rest_proxy_queue_message (priv->proxy,
+                             message,
+                             _continuous_call_message_completed_cb,
+                             closure);
+  g_free (priv->url);
+  priv->url = NULL;
+  return TRUE;
+
+error:
+  g_free (priv->url);
+  priv->url = NULL;
+  return FALSE;
+}
+
 /**
  * rest_proxy_call_cancel:
  * @call: The #RestProxyCall
diff --git a/rest/rest-proxy-call.h b/rest/rest-proxy-call.h
index c9648b2..6f65914 100644
--- a/rest/rest-proxy-call.h
+++ b/rest/rest-proxy-call.h
@@ -155,6 +155,20 @@ gboolean rest_proxy_call_async (RestProxyCall                *call,
                                 gpointer                      userdata,
                                 GError                      **error);
 
+
+typedef void (*RestProxyCallContinuousCallback) (RestProxyCall *call,
+                                                 const gchar   *buf,
+                                                 gsize          len,
+                                                 const GError  *error,
+                                                 GObject       *weak_object,
+                                                 gpointer       userdata);
+
+gboolean rest_proxy_call_continuous (RestProxyCall                    *call,
+                                     RestProxyCallContinuousCallback   cb,
+                                     GObject                          *weak_object,
+                                     gpointer                          userdata,
+                                     GError                          **error);
+
 gboolean rest_proxy_call_cancel (RestProxyCall *call);
 
 gboolean rest_proxy_call_sync (RestProxyCall *call, GError **error_out);