gio/gsocketoutputstream.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 "goutputstream.h"
  28 #include "gsocketoutputstream.h"
  29 #include "gsocket.h"
  30
  31 #include "gsimpleasyncresult.h"
  32 #include "gcancellable.h"
  33 #include "gpollableinputstream.h"
  34 #include "gpollableoutputstream.h"
  35 #include "gioerror.h"
  36 #include "glibintl.h"
  37
  38
  39 static void g_socket_output_stream_pollable_iface_init (GPollableOutputStreamInterface *iface);
  40
  41 #define g_socket_output_stream_get_type _g_socket_output_stream_get_type
  42 G_DEFINE_TYPE_WITH_CODE (GSocketOutputStream, g_socket_output_stream, G_TYPE_OUTPUT_STREAM,
  43                          G_IMPLEMENT_INTERFACE (G_TYPE_POLLABLE_OUTPUT_STREAM, g_socket_output_stream_pollable_iface_init)
  44                          )
  45
  46 enum
  47 {
  48   PROP_0,
  49   PROP_SOCKET
  50 };
  51
  52 struct _GSocketOutputStreamPrivate
  53 {
  54   GSocket *socket;
  55
  56   /* pending operation metadata */
  57   GSimpleAsyncResult *result;
  58   GCancellable *cancellable;
  59   gconstpointer buffer;
  60   gsize count;
  61 };
  62
  63 static void
  64 g_socket_output_stream_get_property (GObject    *object,
  65                                      guint       prop_id,
  66                                      GValue     *value,
  67                                      GParamSpec *pspec)
  68 {
  69   GSocketOutputStream *stream = G_SOCKET_OUTPUT_STREAM (object);
  70
  71   switch (prop_id)
  72     {
  73       case PROP_SOCKET:
  74         g_value_set_object (value, stream->priv->socket);
  75         break;
  76
  77       default:
  78         G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
  79     }
  80 }
  81
  82 static void
  83 g_socket_output_stream_set_property (GObject      *object,
  84                                      guint         prop_id,
  85                                      const GValue *value,
  86                                      GParamSpec   *pspec)
  87 {
  88   GSocketOutputStream *stream = G_SOCKET_OUTPUT_STREAM (object);
  89
  90   switch (prop_id)
  91     {
  92       case PROP_SOCKET:
  93         stream->priv->socket = g_value_dup_object (value);
  94         break;
  95
  96       default:
  97         G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
  98     }
  99 }
 100
 101 static void
 102 g_socket_output_stream_finalize (GObject *object)
 103 {
 104   GSocketOutputStream *stream = G_SOCKET_OUTPUT_STREAM (object);
 105
 106   if (stream->priv->socket)
 107     g_object_unref (stream->priv->socket);
 108
 109   if (G_OBJECT_CLASS (g_socket_output_stream_parent_class)->finalize)
 110     (*G_OBJECT_CLASS (g_socket_output_stream_parent_class)->finalize) (object);
 111 }
 112
 113 static gssize
 114 g_socket_output_stream_write (GOutputStream  *stream,
 115                               const void     *buffer,
 116                               gsize           count,
 117                               GCancellable   *cancellable,
 118                               GError        **error)
 119 {
 120   GSocketOutputStream *onput_stream = G_SOCKET_OUTPUT_STREAM (stream);
 121
 122   return g_socket_send_with_blocking (onput_stream->priv->socket,
 123                                       buffer, count, TRUE,
 124                                       cancellable, error);
 125 }
 126
 127 static gboolean
 128 g_socket_output_stream_write_ready (GSocket *socket,
 129                                     GIOCondition condition,
 130                                     GSocketOutputStream *stream)
 131 {
 132   GSimpleAsyncResult *simple;
 133   GError *error = NULL;
 134   gssize result;
 135
 136   result = g_socket_send_with_blocking (stream->priv->socket,
 137                                         stream->priv->buffer,
 138                                         stream->priv->count,
 139                                         FALSE,
 140                                         stream->priv->cancellable,
 141                                         &error);
 142
 143   if (g_error_matches (error, G_IO_ERROR, G_IO_ERROR_WOULD_BLOCK))
 144     return TRUE;
 145
 146   simple = stream->priv->result;
 147   stream->priv->result = NULL;
 148
 149   if (result >= 0)
 150     g_simple_async_result_set_op_res_gssize (simple, result);
 151
 152   if (error)
 153     g_simple_async_result_take_error (simple, error);
 154
 155   if (stream->priv->cancellable)
 156     g_object_unref (stream->priv->cancellable);
 157
 158   g_simple_async_result_complete (simple);
 159   g_object_unref (simple);
 160
 161   return FALSE;
 162 }
 163
 164 static void
 165 g_socket_output_stream_write_async (GOutputStream        *stream,
 166                                     const void           *buffer,
 167                                     gsize                 count,
 168                                     gint                  io_priority,
 169                                     GCancellable         *cancellable,
 170                                     GAsyncReadyCallback   callback,
 171                                     gpointer              user_data)
 172 {
 173   GSocketOutputStream *output_stream = G_SOCKET_OUTPUT_STREAM (stream);
 174   GSource *source;
 175
 176   g_assert (output_stream->priv->result == NULL);
 177
 178   output_stream->priv->result =
 179     g_simple_async_result_new (G_OBJECT (stream), callback, user_data,
 180                                g_socket_output_stream_write_async);
 181   if (cancellable)
 182     g_object_ref (cancellable);
 183   output_stream->priv->cancellable = cancellable;
 184   output_stream->priv->buffer = buffer;
 185   output_stream->priv->count = count;
 186
 187   source = g_socket_create_source (output_stream->priv->socket,
 188                                    G_IO_OUT | G_IO_HUP | G_IO_ERR,
 189                                    cancellable);
 190   g_source_set_callback (source,
 191                          (GSourceFunc) g_socket_output_stream_write_ready,
 192                          g_object_ref (output_stream), g_object_unref);
 193   g_source_attach (source, g_main_context_get_thread_default ());
 194   g_source_unref (source);
 195 }
 196
 197 static gssize
 198 g_socket_output_stream_write_finish (GOutputStream  *stream,
 199                                      GAsyncResult   *result,
 200                                      GError        **error)
 201 {
 202   GSimpleAsyncResult *simple;
 203   gssize count;
 204
 205   g_return_val_if_fail (G_IS_SOCKET_OUTPUT_STREAM (stream), -1);
 206
 207   simple = G_SIMPLE_ASYNC_RESULT (result);
 208
 209   g_warn_if_fail (g_simple_async_result_get_source_tag (simple) == g_socket_output_stream_write_async);
 210
 211   count = g_simple_async_result_get_op_res_gssize (simple);
 212
 213   return count;
 214 }
 215
 216 static gboolean
 217 g_socket_output_stream_pollable_is_writable (GPollableOutputStream *pollable)
 218 {
 219   GSocketOutputStream *output_stream = G_SOCKET_OUTPUT_STREAM (pollable);
 220
 221   return g_socket_condition_check (output_stream->priv->socket, G_IO_OUT);
 222 }
 223
 224 static GSource *
 225 g_socket_output_stream_pollable_create_source (GPollableOutputStream *pollable,
 226                                                GCancellable          *cancellable)
 227 {
 228   GSocketOutputStream *output_stream = G_SOCKET_OUTPUT_STREAM (pollable);
 229   GSource *socket_source, *pollable_source;
 230
 231   pollable_source = g_pollable_source_new (G_OBJECT (output_stream));
 232   socket_source = g_socket_create_source (output_stream->priv->socket,
 233                                           G_IO_OUT, cancellable);
 234   g_source_set_dummy_callback (socket_source);
 235   g_source_add_child_source (pollable_source, socket_source);
 236   g_source_unref (socket_source);
 237
 238   return pollable_source;
 239 }
 240
 241 static gssize
 242 g_socket_output_stream_pollable_write_nonblocking (GPollableOutputStream  *pollable,
 243                                                    const void             *buffer,
 244                                                    gsize                   size,
 245                                                    GError                **error)
 246 {
 247   GSocketOutputStream *output_stream = G_SOCKET_OUTPUT_STREAM (pollable);
 248
 249   return g_socket_send_with_blocking (output_stream->priv->socket,
 250                                       buffer, size, FALSE,
 251                                       NULL, error);
 252 }
 253
 254 static void
 255 g_socket_output_stream_class_init (GSocketOutputStreamClass *klass)
 256 {
 257   GObjectClass *gobject_class = G_OBJECT_CLASS (klass);
 258   GOutputStreamClass *goutputstream_class = G_OUTPUT_STREAM_CLASS (klass);
 259
 260   g_type_class_add_private (klass, sizeof (GSocketOutputStreamPrivate));
 261
 262   gobject_class->finalize = g_socket_output_stream_finalize;
 263   gobject_class->get_property = g_socket_output_stream_get_property;
 264   gobject_class->set_property = g_socket_output_stream_set_property;
 265
 266   goutputstream_class->write_fn = g_socket_output_stream_write;
 267   goutputstream_class->write_async = g_socket_output_stream_write_async;
 268   goutputstream_class->write_finish = g_socket_output_stream_write_finish;
 269
 270   g_object_class_install_property (gobject_class, PROP_SOCKET,
 271                                    g_param_spec_object ("socket",
 272                                                         P_("socket"),
 273                                                         P_("The socket that this stream wraps"),
 274                                                         G_TYPE_SOCKET, G_PARAM_CONSTRUCT_ONLY |
 275                                                         G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 276 }
 277
 278 static void
 279 g_socket_output_stream_pollable_iface_init (GPollableOutputStreamInterface *iface)
 280 {
 281   iface->is_writable = g_socket_output_stream_pollable_is_writable;
 282   iface->create_source = g_socket_output_stream_pollable_create_source;
 283   iface->write_nonblocking = g_socket_output_stream_pollable_write_nonblocking;
 284 }
 285
 286 static void
 287 g_socket_output_stream_init (GSocketOutputStream *stream)
 288 {
 289   stream->priv = G_TYPE_INSTANCE_GET_PRIVATE (stream, G_TYPE_SOCKET_OUTPUT_STREAM, GSocketOutputStreamPrivate);
 290 }
 291
 292 GSocketOutputStream *
 293 _g_socket_output_stream_new (GSocket *socket)
 294 {
 295   return G_SOCKET_OUTPUT_STREAM (g_object_new (G_TYPE_SOCKET_OUTPUT_STREAM, "socket", socket, NULL));
 296 }