gio/gpollableutils.c

   1 /* GIO - GLib Input, Output and Streaming Library
   2  *
   3  * Copyright (C) 2010 Red Hat, Inc.
   4  *
   5  * This library is free software; you can redistribute it and/or
   6  * modify it under the terms of the GNU Lesser General Public
   7  * License as published by the Free Software Foundation; either
   8  * version 2 of the License, or (at your option) any later version.
   9  *
  10  * This library is distributed in the hope that it will be useful,
  11  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13  * Lesser General Public License for more details.
  14  *
  15  * You should have received a copy of the GNU Lesser General
  16  * Public License along with this library; if not, write to the
  17  * Free Software Foundation, Inc., 59 Temple Place, Suite 330,
  18  * Boston, MA 02111-1307, USA.
  19  */
  20
  21 #include "config.h"
  22
  23 #include <errno.h>
  24
  25 #include "gpollableinputstream.h"
  26 #include "gasynchelper.h"
  27 #include "glibintl.h"
  28
  29 /**
  30  * SECTION:gpollableutils
  31  * @short_description: Utilities for pollable streams
  32  * @include: gio/gio.h
  33  *
  34  * Utility functions for #GPollableInputStream and
  35  * #GPollableOutputStream implementations.
  36  */
  37
  38 typedef struct {
  39   GSource       source;
  40
  41   GObject      *stream;
  42 } GPollableSource;
  43
  44 static gboolean
  45 pollable_source_prepare (GSource *source,
  46                          gint    *timeout)
  47 {
  48   *timeout = -1;
  49   return FALSE;
  50 }
  51
  52 static gboolean
  53 pollable_source_check (GSource *source)
  54 {
  55   return FALSE;
  56 }
  57
  58 static gboolean
  59 pollable_source_dispatch (GSource     *source,
  60                           GSourceFunc  callback,
  61                           gpointer     user_data)
  62 {
  63   GPollableSourceFunc func = (GPollableSourceFunc)callback;
  64   GPollableSource *pollable_source = (GPollableSource *)source;
  65
  66   return (*func) (pollable_source->stream, user_data);
  67 }
  68
  69 static void
  70 pollable_source_finalize (GSource *source)
  71 {
  72   GPollableSource *pollable_source = (GPollableSource *)source;
  73
  74   g_object_unref (pollable_source->stream);
  75 }
  76
  77 static gboolean
  78 pollable_source_closure_callback (GObject  *stream,
  79                                   gpointer  data)
  80 {
  81   GClosure *closure = data;
  82
  83   GValue param = G_VALUE_INIT;
  84   GValue result_value = G_VALUE_INIT;
  85   gboolean result;
  86
  87   g_value_init (&result_value, G_TYPE_BOOLEAN);
  88
  89   g_value_init (&param, G_TYPE_OBJECT);
  90   g_value_set_object (&param, stream);
  91
  92   g_closure_invoke (closure, &result_value, 1, &param, NULL);
  93
  94   result = g_value_get_boolean (&result_value);
  95   g_value_unset (&result_value);
  96   g_value_unset (&param);
  97
  98   return result;
  99 }
 100
 101 static GSourceFuncs pollable_source_funcs =
 102 {
 103   pollable_source_prepare,
 104   pollable_source_check,
 105   pollable_source_dispatch,
 106   pollable_source_finalize,
 107   (GSourceFunc)pollable_source_closure_callback,
 108   (GSourceDummyMarshal)g_cclosure_marshal_generic,
 109 };
 110
 111 /**
 112  * g_pollable_source_new:
 113  * @pollable_stream: the stream associated with the new source
 114  *
 115  * Utility method for #GPollableInputStream and #GPollableOutputStream
 116  * implementations. Creates a new #GSource that expects a callback of
 117  * type #GPollableSourceFunc. The new source does not actually do
 118  * anything on its own; use g_source_add_child_source() to add other
 119  * sources to it to cause it to trigger.
 120  *
 121  * Return value: (transfer full): the new #GSource.
 122  *
 123  * Since: 2.28
 124  */
 125 GSource *
 126 g_pollable_source_new (GObject *pollable_stream)
 127 {
 128   GSource *source;
 129   GPollableSource *pollable_source;
 130
 131   g_return_val_if_fail (G_IS_POLLABLE_INPUT_STREAM (pollable_stream) ||
 132                         G_IS_POLLABLE_OUTPUT_STREAM (pollable_stream), NULL);
 133
 134   source = g_source_new (&pollable_source_funcs, sizeof (GPollableSource));
 135   g_source_set_name (source, "GPollableSource");
 136   pollable_source = (GPollableSource *)source;
 137   pollable_source->stream = g_object_ref (pollable_stream);
 138
 139   return source;
 140 }
 141
 142 /**
 143  * g_pollable_source_new_full:
 144  * @pollable_stream: (type GObject): the stream associated with the
 145  *   new source
 146  * @child_source: (allow-none): optional child source to attach
 147  * @cancellable: (allow-none): optional #GCancellable to attach
 148  *
 149  * Utility method for #GPollableInputStream and #GPollableOutputStream
 150  * implementations. Creates a new #GSource, as with
 151  * g_pollable_source_new(), but also attaching @child_source (with a
 152  * dummy callback), and @cancellable, if they are non-%NULL.
 153  *
 154  * Return value: (transfer full): the new #GSource.
 155  *
 156  * Since: 2.34
 157  */
 158 GSource *
 159 g_pollable_source_new_full (gpointer      pollable_stream,
 160                             GSource      *child_source,
 161                             GCancellable *cancellable)
 162 {
 163   GSource *source;
 164
 165   g_return_val_if_fail (G_IS_POLLABLE_INPUT_STREAM (pollable_stream) ||
 166                         G_IS_POLLABLE_OUTPUT_STREAM (pollable_stream), NULL);
 167
 168   source = g_pollable_source_new (pollable_stream);
 169   if (child_source)
 170     {
 171       g_source_set_dummy_callback (child_source);
 172       g_source_add_child_source (source, child_source);
 173     }
 174   if (cancellable)
 175     {
 176       GSource *cancellable_source = g_cancellable_source_new (cancellable);
 177
 178       g_source_set_dummy_callback (cancellable_source);
 179       g_source_add_child_source (source, cancellable_source);
 180       g_source_unref (cancellable_source);
 181     }
 182
 183   return source;
 184 }
 185
 186 /**
 187  * g_pollable_stream_read:
 188  * @stream: a #GInputStream
 189  * @buffer: a buffer to read data into
 190  * @count: the number of bytes to read
 191  * @blocking: whether to do blocking I/O
 192  * @cancellable: (allow-none): optional #GCancellable object, %NULL to ignore.
 193  * @error: location to store the error occurring, or %NULL to ignore
 194  *
 195  * Tries to read from @stream, as with g_input_stream_read() (if
 196  * @blocking is %TRUE) or g_pollable_input_stream_read_nonblocking()
 197  * (if @blocking is %FALSE). This can be used to more easily share
 198  * code between blocking and non-blocking implementations of a method.
 199  *
 200  * If @blocking is %FALSE, then @stream must be a
 201  * #GPollableInputStream for which g_pollable_input_stream_can_poll()
 202  * returns %TRUE, or else the behavior is undefined. If @blocking is
 203  * %TRUE, then @stream does not need to be a #GPollableInputStream.
 204  *
 205  * Returns: the number of bytes read, or -1 on error.
 206  *
 207  * Since: 2.34
 208  */
 209 gssize
 210 g_pollable_stream_read (GInputStream   *stream,
 211                         void           *buffer,
 212                         gsize           count,
 213                         gboolean        blocking,
 214                         GCancellable   *cancellable,
 215                         GError        **error)
 216 {
 217   if (blocking)
 218     {
 219       return g_input_stream_read (stream,
 220                                   buffer, count,
 221                                   cancellable, error);
 222     }
 223   else
 224     {
 225       return g_pollable_input_stream_read_nonblocking (G_POLLABLE_INPUT_STREAM (stream),
 226                                                        buffer, count,
 227                                                        cancellable, error);
 228     }
 229 }
 230
 231 /**
 232  * g_pollable_stream_write:
 233  * @stream: a #GOutputStream.
 234  * @buffer: (array length=count) (element-type guint8): the buffer
 235  *   containing the data to write.
 236  * @count: the number of bytes to write
 237  * @blocking: whether to do blocking I/O
 238  * @cancellable: (allow-none): optional #GCancellable object, %NULL to ignore.
 239  * @error: location to store the error occurring, or %NULL to ignore
 240  *
 241  * Tries to write to @stream, as with g_output_stream_write() (if
 242  * @blocking is %TRUE) or g_pollable_output_stream_write_nonblocking()
 243  * (if @blocking is %FALSE). This can be used to more easily share
 244  * code between blocking and non-blocking implementations of a method.
 245  *
 246  * If @blocking is %FALSE, then @stream must be a
 247  * #GPollableOutputStream for which
 248  * g_pollable_output_stream_can_poll() returns %TRUE or else the
 249  * behavior is undefined. If @blocking is %TRUE, then @stream does not
 250  * need to be a #GPollableOutputStream.
 251  *
 252  * Returns: the number of bytes written, or -1 on error.
 253  *
 254  * Since: 2.34
 255  */
 256 gssize
 257 g_pollable_stream_write (GOutputStream   *stream,
 258                          const void      *buffer,
 259                          gsize            count,
 260                          gboolean         blocking,
 261                          GCancellable    *cancellable,
 262                          GError         **error)
 263 {
 264   if (blocking)
 265     {
 266       return g_output_stream_write (stream,
 267                                     buffer, count,
 268                                     cancellable, error);
 269     }
 270   else
 271     {
 272       return g_pollable_output_stream_write_nonblocking (G_POLLABLE_OUTPUT_STREAM (stream),
 273                                                          buffer, count,
 274                                                          cancellable, error);
 275     }
 276 }
 277
 278 /**
 279  * g_pollable_stream_write_all:
 280  * @stream: a #GOutputStream.
 281  * @buffer: (array length=count) (element-type guint8): the buffer
 282  *   containing the data to write.
 283  * @count: the number of bytes to write
 284  * @blocking: whether to do blocking I/O
 285  * @bytes_written: (out): location to store the number of bytes that was
 286  *   written to the stream
 287  * @cancellable: (allow-none): optional #GCancellable object, %NULL to ignore.
 288  * @error: location to store the error occurring, or %NULL to ignore
 289  *
 290  * Tries to write @count bytes to @stream, as with
 291  * g_output_stream_write_all(), but using g_pollable_stream_write()
 292  * rather than g_output_stream_write().
 293  *
 294  * On a successful write of @count bytes, %TRUE is returned, and
 295  * @bytes_written is set to @count.
 296  *
 297  * If there is an error during the operation (including
 298  * %G_IO_ERROR_WOULD_BLOCK in the non-blocking case), %FALSE is
 299  * returned and @error is set to indicate the error status,
 300  * @bytes_written is updated to contain the number of bytes written
 301  * into the stream before the error occurred.
 302  *
 303  * As with g_pollable_stream_write(), if @blocking is %FALSE, then
 304  * @stream must be a #GPollableOutputStream for which
 305  * g_pollable_output_stream_can_poll() returns %TRUE or else the
 306  * behavior is undefined. If @blocking is %TRUE, then @stream does not
 307  * need to be a #GPollableOutputStream.
 308  *
 309  * Return value: %TRUE on success, %FALSE if there was an error
 310  *
 311  * Since: 2.34
 312  */
 313 gboolean
 314 g_pollable_stream_write_all (GOutputStream  *stream,
 315                              const void     *buffer,
 316                              gsize           count,
 317                              gboolean        blocking,
 318                              gsize          *bytes_written,
 319                              GCancellable   *cancellable,
 320                              GError        **error)
 321 {
 322   gsize _bytes_written;
 323   gssize res;
 324
 325   _bytes_written = 0;
 326   while (_bytes_written < count)
 327     {
 328       res = g_pollable_stream_write (stream,
 329                                      (char *)buffer + _bytes_written,
 330                                      count - _bytes_written,
 331                                      blocking,
 332                                      cancellable, error);
 333       if (res == -1)
 334         {
 335           if (bytes_written)
 336             *bytes_written = _bytes_written;
 337           return FALSE;
 338         }
 339
 340       if (res == 0)
 341         g_warning ("Write returned zero without error");
 342
 343       _bytes_written += res;
 344     }
 345
 346   if (bytes_written)
 347     *bytes_written = _bytes_written;
 348
 349   return TRUE;
 350 }