gio/gsocketinputstream.c

   1 /* GIO - GLib Input, Output and Streaming Library
   2  *
   3  * Copyright © 2008 Christian Kellner, Samuel Cormier-Iijima
   4  *           © 2009 codethink
   5  *
   6  * This library is free software; you can redistribute it and/or
   7  * modify it under the terms of the GNU Lesser General Public
   8  * License as published by the Free Software Foundation; either
   9  * version 2 of the License, or (at your option) any later version.
  10  *
  11  * This library is distributed in the hope that it will be useful,
  12  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  13  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  14  * Lesser General Public License for more details.
  15  *
  16  * You should have received a copy of the GNU Lesser General
  17  * Public License along with this library; if not, write to the
  18  * Free Software Foundation, Inc., 59 Temple Place, Suite 330,
  19  * Boston, MA 02111-1307, USA.
  20  *
  21  * Authors: Christian Kellner <gicmo@gnome.org>
  22  *          Samuel Cormier-Iijima <sciyoshi@gmail.com>
  23  *          Ryan Lortie <desrt@desrt.ca>
  24  */
  25
  26 #include "config.h"
  27 #include "gsocketinputstream.h"
  28 #include "glibintl.h"
  29
  30 #include <gio/gsimpleasyncresult.h>
  31 #include <gio/gcancellable.h>
  32
  33 #define g_socket_input_stream_get_type _g_socket_input_stream_get_type
  34 G_DEFINE_TYPE (GSocketInputStream, g_socket_input_stream, G_TYPE_INPUT_STREAM);
  35
  36 enum
  37 {
  38   PROP_0,
  39   PROP_SOCKET
  40 };
  41
  42 struct _GSocketInputStreamPrivate
  43 {
  44   GSocket *socket;
  45
  46   /* pending operation metadata */
  47   GSimpleAsyncResult *result;
  48   GCancellable *cancellable;
  49   gboolean from_mainloop;
  50   gpointer buffer;
  51   gsize count;
  52 };
  53
  54 static void
  55 g_socket_input_stream_get_property (GObject    *object,
  56                                     guint       prop_id,
  57                                     GValue     *value,
  58                                     GParamSpec *pspec)
  59 {
  60   GSocketInputStream *stream = G_SOCKET_INPUT_STREAM (object);
  61
  62   switch (prop_id)
  63     {
  64       case PROP_SOCKET:
  65         g_value_set_object (value, stream->priv->socket);
  66         break;
  67
  68       default:
  69         G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
  70     }
  71 }
  72
  73 static void
  74 g_socket_input_stream_set_property (GObject      *object,
  75                                     guint         prop_id,
  76                                     const GValue *value,
  77                                     GParamSpec   *pspec)
  78 {
  79   GSocketInputStream *stream = G_SOCKET_INPUT_STREAM (object);
  80
  81   switch (prop_id)
  82     {
  83       case PROP_SOCKET:
  84         stream->priv->socket = g_value_dup_object (value);
  85         break;
  86
  87       default:
  88         G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
  89     }
  90 }
  91
  92 static void
  93 g_socket_input_stream_finalize (GObject *object)
  94 {
  95   GSocketInputStream *stream = G_SOCKET_INPUT_STREAM (object);
  96
  97   if (stream->priv->socket)
  98     g_object_unref (stream->priv->socket);
  99
 100   if (G_OBJECT_CLASS (g_socket_input_stream_parent_class)->finalize)
 101     (*G_OBJECT_CLASS (g_socket_input_stream_parent_class)->finalize) (object);
 102 }
 103
 104 static gssize
 105 g_socket_input_stream_read (GInputStream  *stream,
 106                             void          *buffer,
 107                             gsize          count,
 108                             GCancellable  *cancellable,
 109                             GError       **error)
 110 {
 111   GSocketInputStream *input_stream = G_SOCKET_INPUT_STREAM (stream);
 112
 113   if (!g_socket_condition_wait (input_stream->priv->socket,
 114                                 G_IO_IN, cancellable, error))
 115     return -1;
 116
 117   return g_socket_receive (input_stream->priv->socket, buffer, count, error);
 118 }
 119
 120 static gboolean
 121 g_socket_input_stream_read_ready (GSocket *socket,
 122                                   GIOCondition condition,
 123                                   GSocketInputStream *stream)
 124 {
 125   GSimpleAsyncResult *simple;
 126   GError *error = NULL;
 127
 128   simple = stream->priv->result;
 129   stream->priv->result = NULL;
 130
 131   if (!g_cancellable_set_error_if_cancelled (stream->priv->cancellable,
 132                                              &error))
 133     {
 134       gssize result;
 135
 136       result = g_socket_receive (stream->priv->socket,
 137                                  stream->priv->buffer,
 138                                  stream->priv->count,
 139                                  &error);
 140       if (result >= 0)
 141         g_simple_async_result_set_op_res_gssize (simple, result);
 142     }
 143
 144   if (error)
 145     {
 146       g_simple_async_result_set_from_error (simple, error);
 147       g_error_free (error);
 148     }
 149
 150   if (stream->priv->cancellable)
 151     g_object_unref (stream->priv->cancellable);
 152
 153   if (stream->priv->from_mainloop)
 154     g_simple_async_result_complete (simple);
 155   else
 156     g_simple_async_result_complete_in_idle (simple);
 157
 158   g_object_unref (simple);
 159
 160   return FALSE;
 161 }
 162
 163 static void
 164 g_socket_input_stream_read_async (GInputStream        *stream,
 165                                   void                *buffer,
 166                                   gsize                count,
 167                                   gint                 io_priority,
 168                                   GCancellable        *cancellable,
 169                                   GAsyncReadyCallback  callback,
 170                                   gpointer             user_data)
 171 {
 172   GSocketInputStream *input_stream = G_SOCKET_INPUT_STREAM (stream);
 173
 174   g_assert (input_stream->priv->result == NULL);
 175
 176   input_stream->priv->result =
 177     g_simple_async_result_new (G_OBJECT (stream), callback, user_data,
 178                                g_socket_input_stream_read_async);
 179   if (cancellable)
 180     g_object_ref (cancellable);
 181   input_stream->priv->cancellable = cancellable;
 182   input_stream->priv->buffer = buffer;
 183   input_stream->priv->count = count;
 184
 185   if (!g_socket_condition_check (input_stream->priv->socket, G_IO_IN))
 186     {
 187       GSource *source;
 188
 189       input_stream->priv->from_mainloop = TRUE;
 190       source = g_socket_create_source (input_stream->priv->socket,
 191                                        G_IO_IN | G_IO_HUP | G_IO_ERR,
 192                                        cancellable);
 193       g_source_set_callback (source,
 194                              (GSourceFunc) g_socket_input_stream_read_ready,
 195                              g_object_ref (input_stream), g_object_unref);
 196       g_source_attach (source, NULL);
 197       g_source_unref (source);
 198     }
 199   else
 200     {
 201       input_stream->priv->from_mainloop = FALSE;
 202       g_socket_input_stream_read_ready (input_stream->priv->socket, G_IO_IN, input_stream);
 203     }
 204 }
 205
 206 static gssize
 207 g_socket_input_stream_read_finish (GInputStream  *stream,
 208                                    GAsyncResult  *result,
 209                                    GError       **error)
 210 {
 211   GSimpleAsyncResult *simple;
 212   gssize count;
 213
 214   g_return_val_if_fail (G_IS_SOCKET_INPUT_STREAM (stream), -1);
 215
 216   simple = G_SIMPLE_ASYNC_RESULT (result);
 217
 218   g_warn_if_fail (g_simple_async_result_get_source_tag (simple) == g_socket_input_stream_read_async);
 219
 220   count = g_simple_async_result_get_op_res_gssize (simple);
 221
 222   return count;
 223 }
 224
 225 static void
 226 g_socket_input_stream_class_init (GSocketInputStreamClass *klass)
 227 {
 228   GObjectClass *gobject_class = G_OBJECT_CLASS (klass);
 229   GInputStreamClass *ginputstream_class = G_INPUT_STREAM_CLASS (klass);
 230
 231   g_type_class_add_private (klass, sizeof (GSocketInputStreamPrivate));
 232
 233   gobject_class->finalize = g_socket_input_stream_finalize;
 234   gobject_class->get_property = g_socket_input_stream_get_property;
 235   gobject_class->set_property = g_socket_input_stream_set_property;
 236
 237   ginputstream_class->read_fn = g_socket_input_stream_read;
 238   ginputstream_class->read_async = g_socket_input_stream_read_async;
 239   ginputstream_class->read_finish = g_socket_input_stream_read_finish;
 240
 241   g_object_class_install_property (gobject_class, PROP_SOCKET,
 242                                    g_param_spec_object ("socket",
 243                                                         P_("socket"),
 244                                                         P_("The socket that this stream wraps"),
 245                                                         G_TYPE_SOCKET, G_PARAM_CONSTRUCT_ONLY |
 246                                                         G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 247 }
 248
 249 static void
 250 g_socket_input_stream_init (GSocketInputStream *stream)
 251 {
 252   stream->priv = G_TYPE_INSTANCE_GET_PRIVATE (stream, G_TYPE_SOCKET_INPUT_STREAM, GSocketInputStreamPrivate);
 253 }
 254
 255 GSocketInputStream *
 256 _g_socket_input_stream_new (GSocket *socket)
 257 {
 258   return G_SOCKET_INPUT_STREAM (g_object_new (G_TYPE_SOCKET_INPUT_STREAM, "socket", socket, NULL));
 259 }