libsoup/soup-client-input-stream.c

   1 /* -*- Mode: C; tab-width: 8; indent-tabs-mode: t; c-basic-offset: 8 -*- */
   2 /*
   3  * soup-client-input-stream.c
   4  *
   5  * Copyright 2010-2012 Red Hat, Inc.
   6  */
   7
   8 #ifdef HAVE_CONFIG_H
   9 #include <config.h>
  10 #endif
  11
  12 #include "soup-client-input-stream.h"
  13 #include "soup.h"
  14 #include "soup-marshal.h"
  15 #include "soup-message-private.h"
  16
  17 struct _SoupClientInputStreamPrivate {
  18         SoupMessage  *msg;
  19 };
  20
  21 enum {
  22         EOF,
  23         LAST_SIGNAL
  24 };
  25
  26 static guint signals[LAST_SIGNAL] = { 0 };
  27
  28 enum {
  29         PROP_0,
  30
  31         PROP_MESSAGE
  32 };
  33
  34 static GPollableInputStreamInterface *soup_client_input_stream_parent_pollable_interface;
  35 static void soup_client_input_stream_pollable_init (GPollableInputStreamInterface *pollable_interface, gpointer interface_data);
  36
  37 G_DEFINE_TYPE_WITH_CODE (SoupClientInputStream, soup_client_input_stream, SOUP_TYPE_FILTER_INPUT_STREAM,
  38                          G_IMPLEMENT_INTERFACE (G_TYPE_POLLABLE_INPUT_STREAM,
  39                                                 soup_client_input_stream_pollable_init))
  40
  41 static void
  42 soup_client_input_stream_init (SoupClientInputStream *stream)
  43 {
  44         stream->priv = G_TYPE_INSTANCE_GET_PRIVATE (stream,
  45                                                     SOUP_TYPE_CLIENT_INPUT_STREAM,
  46                                                     SoupClientInputStreamPrivate);
  47 }
  48
  49 static void
  50 soup_client_input_stream_finalize (GObject *object)
  51 {
  52         SoupClientInputStream *cistream = SOUP_CLIENT_INPUT_STREAM (object);
  53
  54         g_clear_object (&cistream->priv->msg);
  55
  56         G_OBJECT_CLASS (soup_client_input_stream_parent_class)->finalize (object);
  57 }
  58
  59 static void
  60 soup_client_input_stream_set_property (GObject *object, guint prop_id,
  61                                        const GValue *value, GParamSpec *pspec)
  62 {
  63         SoupClientInputStream *cistream = SOUP_CLIENT_INPUT_STREAM (object);
  64
  65         switch (prop_id) {
  66         case PROP_MESSAGE:
  67                 cistream->priv->msg = g_value_dup_object (value);
  68                 break;
  69         default:
  70                 G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
  71                 break;
  72         }
  73 }
  74
  75 static void
  76 soup_client_input_stream_get_property (GObject *object, guint prop_id,
  77                                        GValue *value, GParamSpec *pspec)
  78 {
  79         SoupClientInputStream *cistream = SOUP_CLIENT_INPUT_STREAM (object);
  80
  81         switch (prop_id) {
  82         case PROP_MESSAGE:
  83                 g_value_set_object (value, cistream->priv->msg);
  84                 break;
  85         default:
  86                 G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
  87                 break;
  88         }
  89 }
  90
  91 static gssize
  92 soup_client_input_stream_read_fn (GInputStream  *stream,
  93                                   void          *buffer,
  94                                   gsize          count,
  95                                   GCancellable  *cancellable,
  96                                   GError       **error)
  97 {
  98         gssize nread;
  99
 100         nread = G_INPUT_STREAM_CLASS (soup_client_input_stream_parent_class)->
 101                 read_fn (stream, buffer, count, cancellable, error);
 102
 103         if (nread == 0)
 104                 g_signal_emit (stream, signals[EOF], 0);
 105
 106         return nread;
 107 }
 108
 109 static gssize
 110 soup_client_input_stream_read_nonblocking (GPollableInputStream  *stream,
 111                                            void                  *buffer,
 112                                            gsize                  count,
 113                                            GError               **error)
 114 {
 115         gssize nread;
 116
 117         nread = soup_client_input_stream_parent_pollable_interface->
 118                 read_nonblocking (stream, buffer, count, error);
 119
 120         if (nread == 0)
 121                 g_signal_emit (stream, signals[EOF], 0);
 122
 123         return nread;
 124 }
 125
 126 static gboolean
 127 soup_client_input_stream_close_fn (GInputStream  *stream,
 128                                    GCancellable  *cancellable,
 129                                    GError       **error)
 130 {
 131         SoupClientInputStream *cistream = SOUP_CLIENT_INPUT_STREAM (stream);
 132
 133         return soup_message_io_run_until_finish (cistream->priv->msg,
 134                                                  cancellable, error);
 135 }
 136
 137 static gboolean
 138 idle_finish_close (gpointer user_data)
 139 {
 140         GTask *task = user_data;
 141
 142         g_task_return_boolean (task, TRUE);
 143         g_object_unref (task);
 144         return FALSE;
 145 }
 146
 147 static gboolean
 148 close_async_ready (SoupMessage *msg, gpointer user_data)
 149 {
 150         GTask *task = user_data;
 151         SoupClientInputStream *cistream = g_task_get_source_object (task);
 152         GError *error = NULL;
 153
 154         if (!soup_message_io_run_until_finish (cistream->priv->msg,
 155                                                g_task_get_cancellable (task),
 156                                                &error) &&
 157             g_error_matches (error, G_IO_ERROR, G_IO_ERROR_WOULD_BLOCK)) {
 158                 g_error_free (error);
 159                 return TRUE;
 160         }
 161
 162         if (error) {
 163                 g_task_return_error (task, error);
 164                 g_object_unref (task);
 165                 return FALSE;
 166         }
 167
 168         /* Due to a historical accident, SoupSessionAsync relies on us
 169          * waiting one extra cycle after run_until_finish() returns.
 170          * Ugh. FIXME later when it's easier to do.
 171          */
 172         soup_add_idle (g_main_context_get_thread_default (),
 173                        idle_finish_close, task);
 174         return FALSE;
 175 }
 176
 177 static void
 178 soup_client_input_stream_close_async (GInputStream        *stream,
 179                                       gint                 priority,
 180                                       GCancellable        *cancellable,
 181                                       GAsyncReadyCallback  callback,
 182                                       gpointer             user_data)
 183 {
 184         SoupClientInputStream *cistream = SOUP_CLIENT_INPUT_STREAM (stream);
 185         GTask *task;
 186         GSource *source;
 187
 188         task = g_task_new (stream, cancellable, callback, user_data);
 189         g_task_set_priority (task, priority);
 190
 191         source = soup_message_io_get_source (cistream->priv->msg,
 192                                              cancellable, NULL, NULL);
 193
 194         g_task_attach_source (task, source, (GSourceFunc) close_async_ready);
 195         g_source_unref (source);
 196 }
 197
 198 static gboolean
 199 soup_client_input_stream_close_finish (GInputStream  *stream,
 200                                        GAsyncResult  *result,
 201                                        GError       **error)
 202 {
 203         return g_task_propagate_boolean (G_TASK (result), error);
 204 }
 205
 206 static void
 207 soup_client_input_stream_class_init (SoupClientInputStreamClass *stream_class)
 208 {
 209         GObjectClass *object_class = G_OBJECT_CLASS (stream_class);
 210         GInputStreamClass *input_stream_class = G_INPUT_STREAM_CLASS (stream_class);
 211
 212         g_type_class_add_private (stream_class, sizeof (SoupClientInputStreamPrivate));
 213
 214         object_class->finalize = soup_client_input_stream_finalize;
 215         object_class->set_property = soup_client_input_stream_set_property;
 216         object_class->get_property = soup_client_input_stream_get_property;
 217
 218         input_stream_class->read_fn = soup_client_input_stream_read_fn;
 219         input_stream_class->close_fn = soup_client_input_stream_close_fn;
 220         input_stream_class->close_async = soup_client_input_stream_close_async;
 221         input_stream_class->close_finish = soup_client_input_stream_close_finish;
 222
 223         signals[EOF] =
 224                 g_signal_new ("eof",
 225                               G_OBJECT_CLASS_TYPE (object_class),
 226                               G_SIGNAL_RUN_LAST,
 227                               0,
 228                               NULL, NULL,
 229                               _soup_marshal_NONE__NONE,
 230                               G_TYPE_NONE, 0);
 231
 232         g_object_class_install_property (
 233                 object_class, PROP_MESSAGE,
 234                 g_param_spec_object ("message",
 235                                      "Message",
 236                                      "Message",
 237                                      SOUP_TYPE_MESSAGE,
 238                                      G_PARAM_READWRITE | G_PARAM_CONSTRUCT_ONLY));
 239 }
 240
 241 static void
 242 soup_client_input_stream_pollable_init (GPollableInputStreamInterface *pollable_interface,
 243                                         gpointer interface_data)
 244 {
 245         soup_client_input_stream_parent_pollable_interface =
 246                 g_type_interface_peek_parent (pollable_interface);
 247
 248         pollable_interface->read_nonblocking = soup_client_input_stream_read_nonblocking;
 249 }
 250
 251 GInputStream *
 252 soup_client_input_stream_new (GInputStream *base_stream,
 253                               SoupMessage  *msg)
 254 {
 255         return g_object_new (SOUP_TYPE_CLIENT_INPUT_STREAM,
 256                              "base-stream", base_stream,
 257                              "message", msg,
 258                              NULL);
 259 }