gio/gsocketoutputstream.c

   1 /* GIO - GLib Input, Output and Streaming Library
   2  *
   3  * Copyright (C) 2006-2007 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  * Author: Alexander Larsson <alexl@redhat.com>
  21  */
  22
  23 #include <config.h>
  24
  25 #include <sys/types.h>
  26 #include <sys/stat.h>
  27 #include <unistd.h>
  28 #include <errno.h>
  29 #include <stdio.h>
  30 #include <fcntl.h>
  31 #include <poll.h>
  32
  33 #include <glib.h>
  34 #include <glib/gstdio.h>
  35 #include "gioerror.h"
  36 #include "gsocketoutputstream.h"
  37 #include "gcancellable.h"
  38 #include "gsimpleasyncresult.h"
  39 #include "gasynchelper.h"
  40
  41 #include "glibintl.h"
  42
  43 G_DEFINE_TYPE (GSocketOutputStream, g_socket_output_stream, G_TYPE_OUTPUT_STREAM);
  44
  45
  46 struct _GSocketOutputStreamPrivate {
  47   int fd;
  48   gboolean close_fd_at_close;
  49 };
  50
  51 static gssize   g_socket_output_stream_write        (GOutputStream              *stream,
  52                                                      const void                 *buffer,
  53                                                      gsize                       count,
  54                                                      GCancellable               *cancellable,
  55                                                      GError                    **error);
  56 static gboolean g_socket_output_stream_close        (GOutputStream              *stream,
  57                                                      GCancellable               *cancellable,
  58                                                      GError                    **error);
  59 static void     g_socket_output_stream_write_async  (GOutputStream              *stream,
  60                                                      const void                 *buffer,
  61                                                      gsize                       count,
  62                                                      int                         io_priority,
  63                                                      GCancellable               *cancellable,
  64                                                      GAsyncReadyCallback         callback,
  65                                                      gpointer                    data);
  66 static gssize   g_socket_output_stream_write_finish (GOutputStream              *stream,
  67                                                      GAsyncResult               *result,
  68                                                      GError                    **error);
  69 static void     g_socket_output_stream_close_async  (GOutputStream              *stream,
  70                                                      int                         io_priority,
  71                                                      GCancellable               *cancellable,
  72                                                      GAsyncReadyCallback         callback,
  73                                                      gpointer                    data);
  74 static gboolean g_socket_output_stream_close_finish (GOutputStream              *stream,
  75                                                      GAsyncResult               *result,
  76                                                      GError                    **error);
  77
  78
  79 static void
  80 g_socket_output_stream_finalize (GObject *object)
  81 {
  82   GSocketOutputStream *stream;
  83
  84   stream = G_SOCKET_OUTPUT_STREAM (object);
  85
  86   if (G_OBJECT_CLASS (g_socket_output_stream_parent_class)->finalize)
  87     (*G_OBJECT_CLASS (g_socket_output_stream_parent_class)->finalize) (object);
  88 }
  89
  90 static void
  91 g_socket_output_stream_class_init (GSocketOutputStreamClass *klass)
  92 {
  93   GObjectClass *gobject_class = G_OBJECT_CLASS (klass);
  94   GOutputStreamClass *stream_class = G_OUTPUT_STREAM_CLASS (klass);
  95
  96   g_type_class_add_private (klass, sizeof (GSocketOutputStreamPrivate));
  97
  98   gobject_class->finalize = g_socket_output_stream_finalize;
  99
 100   stream_class->write = g_socket_output_stream_write;
 101   stream_class->close = g_socket_output_stream_close;
 102   stream_class->write_async = g_socket_output_stream_write_async;
 103   stream_class->write_finish = g_socket_output_stream_write_finish;
 104   stream_class->close_async = g_socket_output_stream_close_async;
 105   stream_class->close_finish = g_socket_output_stream_close_finish;
 106 }
 107
 108 static void
 109 g_socket_output_stream_init (GSocketOutputStream *socket)
 110 {
 111   socket->priv = G_TYPE_INSTANCE_GET_PRIVATE (socket,
 112                                               G_TYPE_SOCKET_OUTPUT_STREAM,
 113                                               GSocketOutputStreamPrivate);
 114 }
 115
 116
 117 /**
 118  * g_socket_output_stream_new:
 119  * @fd: file descriptor.
 120  * @close_fd_at_close: boolean value.
 121  *
 122  * Returns: #GOutputStream. If @close_fd_at_close is %TRUE, then
 123  * @fd will be closed when the #GOutputStream is closed.
 124  **/
 125 GOutputStream *
 126 g_socket_output_stream_new (int fd,
 127                            gboolean close_fd_at_close)
 128 {
 129   GSocketOutputStream *stream;
 130
 131   g_return_val_if_fail (fd != -1, NULL);
 132
 133   stream = g_object_new (G_TYPE_SOCKET_OUTPUT_STREAM, NULL);
 134
 135   stream->priv->fd = fd;
 136   stream->priv->close_fd_at_close = close_fd_at_close;
 137
 138   return G_OUTPUT_STREAM (stream);
 139 }
 140
 141 static gssize
 142 g_socket_output_stream_write (GOutputStream *stream,
 143                               const void    *buffer,
 144                               gsize          count,
 145                               GCancellable  *cancellable,
 146                               GError       **error)
 147 {
 148   GSocketOutputStream *socket_stream;
 149   gssize res;
 150   struct pollfd poll_fds[2];
 151   int poll_ret;
 152   int cancel_fd;
 153
 154   socket_stream = G_SOCKET_OUTPUT_STREAM (stream);
 155
 156   cancel_fd = g_cancellable_get_fd (cancellable);
 157   if (cancel_fd != -1)
 158     {
 159       do
 160         {
 161           poll_fds[0].events = POLLOUT;
 162           poll_fds[0].fd = socket_stream->priv->fd;
 163           poll_fds[1].events = POLLIN;
 164           poll_fds[1].fd = cancel_fd;
 165           poll_ret = poll (poll_fds, 2, -1);
 166         }
 167       while (poll_ret == -1 && errno == EINTR);
 168
 169       if (poll_ret == -1)
 170         {
 171           g_set_error (error, G_IO_ERROR,
 172                        g_io_error_from_errno (errno),
 173                        _("Error writing to socket: %s"),
 174                        g_strerror (errno));
 175           return -1;
 176         }
 177     }
 178
 179   while (1)
 180     {
 181       if (g_cancellable_set_error_if_cancelled (cancellable, error))
 182         return -1;
 183
 184       res = write (socket_stream->priv->fd, buffer, count);
 185       if (res == -1)
 186         {
 187           if (errno == EINTR)
 188             continue;
 189
 190           g_set_error (error, G_IO_ERROR,
 191                        g_io_error_from_errno (errno),
 192                        _("Error writing to socket: %s"),
 193                        g_strerror (errno));
 194         }
 195
 196       break;
 197     }
 198
 199   return res;
 200 }
 201
 202 static gboolean
 203 g_socket_output_stream_close (GOutputStream *stream,
 204                               GCancellable  *cancellable,
 205                               GError       **error)
 206 {
 207   GSocketOutputStream *socket_stream;
 208   int res;
 209
 210   socket_stream = G_SOCKET_OUTPUT_STREAM (stream);
 211
 212   if (!socket_stream->priv->close_fd_at_close)
 213     return TRUE;
 214
 215   while (1)
 216     {
 217       /* This might block during the close. Doesn't seem to be a way to avoid it though. */
 218       res = close (socket_stream->priv->fd);
 219       if (res == -1)
 220         {
 221           g_set_error (error, G_IO_ERROR,
 222                        g_io_error_from_errno (errno),
 223                        _("Error closing socket: %s"),
 224                        g_strerror (errno));
 225         }
 226       break;
 227     }
 228
 229   return res != -1;
 230 }
 231
 232 typedef struct {
 233   gsize count;
 234   const void *buffer;
 235   GAsyncReadyCallback callback;
 236   gpointer user_data;
 237   GCancellable *cancellable;
 238   GSocketOutputStream *stream;
 239 } WriteAsyncData;
 240
 241 static gboolean
 242 write_async_cb (WriteAsyncData *data,
 243                 GIOCondition condition,
 244                 int fd)
 245 {
 246   GSimpleAsyncResult *simple;
 247   GError *error = NULL;
 248   gssize count_written;
 249
 250   while (1)
 251     {
 252       if (g_cancellable_set_error_if_cancelled (data->cancellable, &error))
 253         {
 254           count_written = -1;
 255           break;
 256         }
 257
 258       count_written = write (data->stream->priv->fd, data->buffer, data->count);
 259       if (count_written == -1)
 260         {
 261           if (errno == EINTR)
 262             continue;
 263
 264           g_set_error (&error, G_IO_ERROR,
 265                        g_io_error_from_errno (errno),
 266                        _("Error reading from socket: %s"),
 267                        g_strerror (errno));
 268         }
 269       break;
 270     }
 271
 272   simple = g_simple_async_result_new (G_OBJECT (data->stream),
 273                                       data->callback,
 274                                       data->user_data,
 275                                       g_socket_output_stream_write_async);
 276
 277   g_simple_async_result_set_op_res_gssize (simple, count_written);
 278
 279   if (count_written == -1)
 280     {
 281       g_simple_async_result_set_from_error (simple, error);
 282       g_error_free (error);
 283     }
 284
 285   /* Complete immediately, not in idle, since we're already in a mainloop callout */
 286   g_simple_async_result_complete (simple);
 287   g_object_unref (simple);
 288
 289   return FALSE;
 290 }
 291
 292 static void
 293 g_socket_output_stream_write_async (GOutputStream      *stream,
 294                                     const void         *buffer,
 295                                     gsize               count,
 296                                     int                 io_priority,
 297                                     GCancellable       *cancellable,
 298                                     GAsyncReadyCallback callback,
 299                                     gpointer            user_data)
 300 {
 301   GSource *source;
 302   GSocketOutputStream *socket_stream;
 303   WriteAsyncData *data;
 304
 305   socket_stream = G_SOCKET_OUTPUT_STREAM (stream);
 306
 307   data = g_new0 (WriteAsyncData, 1);
 308   data->count = count;
 309   data->buffer = buffer;
 310   data->callback = callback;
 311   data->user_data = user_data;
 312   data->cancellable = cancellable;
 313   data->stream = socket_stream;
 314
 315   source = _g_fd_source_new (socket_stream->priv->fd,
 316                              POLLOUT,
 317                              cancellable);
 318
 319   g_source_set_callback (source, (GSourceFunc)write_async_cb, data, g_free);
 320   g_source_attach (source, NULL);
 321
 322   g_source_unref (source);
 323 }
 324
 325 static gssize
 326 g_socket_output_stream_write_finish (GOutputStream *stream,
 327                                      GAsyncResult *result,
 328                                      GError **error)
 329 {
 330   GSimpleAsyncResult *simple;
 331   gssize nwritten;
 332
 333   simple = G_SIMPLE_ASYNC_RESULT (result);
 334   g_assert (g_simple_async_result_get_source_tag (simple) == g_socket_output_stream_write_async);
 335
 336   nwritten = g_simple_async_result_get_op_res_gssize (simple);
 337   return nwritten;
 338 }
 339
 340 typedef struct {
 341   GOutputStream *stream;
 342   GAsyncReadyCallback callback;
 343   gpointer user_data;
 344 } CloseAsyncData;
 345
 346 static gboolean
 347 close_async_cb (CloseAsyncData *data)
 348 {
 349   GSocketOutputStream *socket_stream;
 350   GSimpleAsyncResult *simple;
 351   GError *error = NULL;
 352   gboolean result;
 353   int res;
 354
 355   socket_stream = G_SOCKET_OUTPUT_STREAM (data->stream);
 356
 357   if (!socket_stream->priv->close_fd_at_close)
 358     {
 359       result = TRUE;
 360       goto out;
 361     }
 362
 363   while (1)
 364     {
 365       res = close (socket_stream->priv->fd);
 366       if (res == -1)
 367         {
 368           g_set_error (&error, G_IO_ERROR,
 369                        g_io_error_from_errno (errno),
 370                        _("Error closing socket: %s"),
 371                        g_strerror (errno));
 372         }
 373       break;
 374     }
 375
 376   result = res != -1;
 377
 378  out:
 379   simple = g_simple_async_result_new (G_OBJECT (data->stream),
 380                                       data->callback,
 381                                       data->user_data,
 382                                       g_socket_output_stream_close_async);
 383
 384   if (!result)
 385     {
 386       g_simple_async_result_set_from_error (simple, error);
 387       g_error_free (error);
 388     }
 389
 390   /* Complete immediately, not in idle, since we're already in a mainloop callout */
 391   g_simple_async_result_complete (simple);
 392   g_object_unref (simple);
 393
 394   return FALSE;
 395 }
 396
 397 static void
 398 g_socket_output_stream_close_async (GOutputStream       *stream,
 399                                     int                 io_priority,
 400                                     GCancellable       *cancellable,
 401                                     GAsyncReadyCallback callback,
 402                                     gpointer            user_data)
 403 {
 404   GSource *idle;
 405   CloseAsyncData *data;
 406
 407   data = g_new0 (CloseAsyncData, 1);
 408
 409   data->stream = stream;
 410   data->callback = callback;
 411   data->user_data = user_data;
 412
 413   idle = g_idle_source_new ();
 414   g_source_set_callback (idle, (GSourceFunc)close_async_cb, data, g_free);
 415   g_source_attach (idle, NULL);
 416   g_source_unref (idle);
 417 }
 418
 419 static gboolean
 420 g_socket_output_stream_close_finish (GOutputStream              *stream,
 421                                      GAsyncResult              *result,
 422                                      GError                   **error)
 423 {
 424   /* Failures handled in generic close_finish code */
 425   return TRUE;
 426 }