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 #include <gio/gioerror.h>
  33
  34
  35 #define g_socket_input_stream_get_type _g_socket_input_stream_get_type
  36 G_DEFINE_TYPE (GSocketInputStream, g_socket_input_stream, G_TYPE_INPUT_STREAM);
  37
  38 enum
  39 {
  40   PROP_0,
  41   PROP_SOCKET
  42 };
  43
  44 struct _GSocketInputStreamPrivate
  45 {
  46   GSocket *socket;
  47
  48   /* pending operation metadata */
  49   GSimpleAsyncResult *result;
  50   GCancellable *cancellable;
  51   gpointer buffer;
  52   gsize count;
  53 };
  54
  55 static void
  56 g_socket_input_stream_get_property (GObject    *object,
  57                                     guint       prop_id,
  58                                     GValue     *value,
  59                                     GParamSpec *pspec)
  60 {
  61   GSocketInputStream *stream = G_SOCKET_INPUT_STREAM (object);
  62
  63   switch (prop_id)
  64     {
  65       case PROP_SOCKET:
  66         g_value_set_object (value, stream->priv->socket);
  67         break;
  68
  69       default:
  70         G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
  71     }
  72 }
  73
  74 static void
  75 g_socket_input_stream_set_property (GObject      *object,
  76                                     guint         prop_id,
  77                                     const GValue *value,
  78                                     GParamSpec   *pspec)
  79 {
  80   GSocketInputStream *stream = G_SOCKET_INPUT_STREAM (object);
  81
  82   switch (prop_id)
  83     {
  84       case PROP_SOCKET:
  85         stream->priv->socket = g_value_dup_object (value);
  86         break;
  87
  88       default:
  89         G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
  90     }
  91 }
  92
  93 static void
  94 g_socket_input_stream_finalize (GObject *object)
  95 {
  96   GSocketInputStream *stream = G_SOCKET_INPUT_STREAM (object);
  97
  98   if (stream->priv->socket)
  99     g_object_unref (stream->priv->socket);
 100
 101   if (G_OBJECT_CLASS (g_socket_input_stream_parent_class)->finalize)
 102     (*G_OBJECT_CLASS (g_socket_input_stream_parent_class)->finalize) (object);
 103 }
 104
 105 static gssize
 106 g_socket_input_stream_read (GInputStream  *stream,
 107                             void          *buffer,
 108                             gsize          count,
 109                             GCancellable  *cancellable,
 110                             GError       **error)
 111 {
 112   GSocketInputStream *input_stream = G_SOCKET_INPUT_STREAM (stream);
 113
 114   return g_socket_receive_with_blocking (input_stream->priv->socket,
 115                                          buffer, count, TRUE,
 116                                          cancellable, error);
 117 }
 118
 119 static gboolean
 120 g_socket_input_stream_read_ready (GSocket *socket,
 121                                   GIOCondition condition,
 122                                   GSocketInputStream *stream)
 123 {
 124   GSimpleAsyncResult *simple;
 125   GError *error = NULL;
 126   gssize result;
 127
 128   result = g_socket_receive_with_blocking (stream->priv->socket,
 129                                            stream->priv->buffer,
 130                                            stream->priv->count,
 131                                            FALSE,
 132                                            stream->priv->cancellable,
 133                                            &error);
 134
 135   if (g_error_matches (error, G_IO_ERROR, G_IO_ERROR_WOULD_BLOCK))
 136     return TRUE;
 137
 138   simple = stream->priv->result;
 139   stream->priv->result = NULL;
 140
 141   if (result >= 0)
 142     g_simple_async_result_set_op_res_gssize (simple, result);
 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   g_simple_async_result_complete (simple);
 154   g_object_unref (simple);
 155
 156   return FALSE;
 157 }
 158
 159 static void
 160 g_socket_input_stream_read_async (GInputStream        *stream,
 161                                   void                *buffer,
 162                                   gsize                count,
 163                                   gint                 io_priority,
 164                                   GCancellable        *cancellable,
 165                                   GAsyncReadyCallback  callback,
 166                                   gpointer             user_data)
 167 {
 168   GSocketInputStream *input_stream = G_SOCKET_INPUT_STREAM (stream);
 169   GSource *source;
 170
 171   g_assert (input_stream->priv->result == NULL);
 172
 173   input_stream->priv->result =
 174     g_simple_async_result_new (G_OBJECT (stream), callback, user_data,
 175                                g_socket_input_stream_read_async);
 176   if (cancellable)
 177     g_object_ref (cancellable);
 178   input_stream->priv->cancellable = cancellable;
 179   input_stream->priv->buffer = buffer;
 180   input_stream->priv->count = count;
 181
 182   source = g_socket_create_source (input_stream->priv->socket,
 183                                    G_IO_IN | G_IO_HUP | G_IO_ERR,
 184                                    cancellable);
 185   g_source_set_callback (source,
 186                          (GSourceFunc) g_socket_input_stream_read_ready,
 187                          g_object_ref (input_stream), g_object_unref);
 188   g_source_attach (source, g_main_context_get_thread_default ());
 189   g_source_unref (source);
 190 }
 191
 192 static gssize
 193 g_socket_input_stream_read_finish (GInputStream  *stream,
 194                                    GAsyncResult  *result,
 195                                    GError       **error)
 196 {
 197   GSimpleAsyncResult *simple;
 198   gssize count;
 199
 200   g_return_val_if_fail (G_IS_SOCKET_INPUT_STREAM (stream), -1);
 201
 202   simple = G_SIMPLE_ASYNC_RESULT (result);
 203
 204   g_warn_if_fail (g_simple_async_result_get_source_tag (simple) == g_socket_input_stream_read_async);
 205
 206   count = g_simple_async_result_get_op_res_gssize (simple);
 207
 208   return count;
 209 }
 210
 211 static void
 212 g_socket_input_stream_class_init (GSocketInputStreamClass *klass)
 213 {
 214   GObjectClass *gobject_class = G_OBJECT_CLASS (klass);
 215   GInputStreamClass *ginputstream_class = G_INPUT_STREAM_CLASS (klass);
 216
 217   g_type_class_add_private (klass, sizeof (GSocketInputStreamPrivate));
 218
 219   gobject_class->finalize = g_socket_input_stream_finalize;
 220   gobject_class->get_property = g_socket_input_stream_get_property;
 221   gobject_class->set_property = g_socket_input_stream_set_property;
 222
 223   ginputstream_class->read_fn = g_socket_input_stream_read;
 224   ginputstream_class->read_async = g_socket_input_stream_read_async;
 225   ginputstream_class->read_finish = g_socket_input_stream_read_finish;
 226
 227   g_object_class_install_property (gobject_class, PROP_SOCKET,
 228                                    g_param_spec_object ("socket",
 229                                                         P_("socket"),
 230                                                         P_("The socket that this stream wraps"),
 231                                                         G_TYPE_SOCKET, G_PARAM_CONSTRUCT_ONLY |
 232                                                         G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 233 }
 234
 235 static void
 236 g_socket_input_stream_init (GSocketInputStream *stream)
 237 {
 238   stream->priv = G_TYPE_INSTANCE_GET_PRIVATE (stream, G_TYPE_SOCKET_INPUT_STREAM, GSocketInputStreamPrivate);
 239 }
 240
 241 GSocketInputStream *
 242 _g_socket_input_stream_new (GSocket *socket)
 243 {
 244   return G_SOCKET_INPUT_STREAM (g_object_new (G_TYPE_SOCKET_INPUT_STREAM, "socket", socket, NULL));
 245 }