libpurple/queuedoutputstream.c

   1 /*
   2  *
   3  * purple
   4  *
   5  * Purple is the legal property of its developers, whose names are too numerous
   6  * to list here.  Please refer to the COPYRIGHT file distributed with this
   7  * source distribution.
   8  *
   9  * This program is free software; you can redistribute it and/or modify
  10  * it under the terms of the GNU General Public License as published by
  11  * the Free Software Foundation; either version 2 of the License, or
  12  * (at your option) any later version.
  13  *
  14  * This program is distributed in the hope that it will be useful,
  15  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  16  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  17  * GNU General Public License for more details.
  18  *
  19  * You should have received a copy of the GNU General Public License
  20  * along with this program; if not, write to the Free Software
  21  * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02111-1301  USA
  22  */
  23
  24 #include "internal.h"
  25 #include "queuedoutputstream.h"
  26
  27 struct _PurpleQueuedOutputStreamPrivate {
  28         GAsyncQueue *queue;
  29         GBytes *next;
  30 };
  31
  32 static GObjectClass *parent_class = NULL;
  33
  34 #define PURPLE_QUEUED_OUTPUT_STREAM_GET_PRIVATE(obj) \
  35                 (G_TYPE_INSTANCE_GET_PRIVATE((obj), \
  36                 PURPLE_TYPE_QUEUED_OUTPUT_STREAM, \
  37                 PurpleQueuedOutputStreamPrivate))
  38
  39 G_DEFINE_TYPE_WITH_CODE(PurpleQueuedOutputStream, purple_queued_output_stream,
  40                 G_TYPE_FILTER_OUTPUT_STREAM,
  41                 G_ADD_PRIVATE(PurpleQueuedOutputStream))
  42
  43 static void purple_queued_output_stream_dispose(GObject *object);
  44 static gboolean purple_queued_output_stream_flush(GOutputStream *stream,
  45                 GCancellable *cancellable, GError **error);
  46 static void purple_queued_output_stream_flush_async(GOutputStream *stream,
  47                 int io_priority, GCancellable *cancellable,
  48                 GAsyncReadyCallback callback, gpointer user_data);
  49 static gboolean purple_queued_output_stream_flush_finish(GOutputStream *stream,
  50                 GAsyncResult *result, GError **error);
  51
  52 static void purple_queued_output_stream_start_flush_async(GTask *task);
  53
  54 static void
  55 purple_queued_output_stream_class_init(PurpleQueuedOutputStreamClass *klass)
  56 {
  57         GObjectClass *object_class;
  58         GOutputStreamClass *ostream_class;
  59
  60         parent_class = g_type_class_peek_parent(klass);
  61
  62         object_class = G_OBJECT_CLASS(klass);
  63         object_class->dispose = purple_queued_output_stream_dispose;
  64
  65         ostream_class = G_OUTPUT_STREAM_CLASS(klass);
  66         ostream_class->flush = purple_queued_output_stream_flush;
  67         ostream_class->flush_async = purple_queued_output_stream_flush_async;
  68         ostream_class->flush_finish = purple_queued_output_stream_flush_finish;
  69 }
  70
  71 PurpleQueuedOutputStream *
  72 purple_queued_output_stream_new(GOutputStream *base_stream)
  73 {
  74         PurpleQueuedOutputStream *stream;
  75
  76         g_return_val_if_fail(G_IS_OUTPUT_STREAM(base_stream), NULL);
  77
  78         stream = g_object_new(PURPLE_TYPE_QUEUED_OUTPUT_STREAM,
  79                         "base-stream", base_stream,
  80                         NULL);
  81
  82         return stream;
  83 }
  84
  85 void
  86 purple_queued_output_stream_push_bytes(PurpleQueuedOutputStream *stream,
  87                 GBytes *bytes)
  88 {
  89         g_return_if_fail(PURPLE_QUEUED_OUTPUT_STREAM(stream));
  90         g_return_if_fail(bytes != NULL);
  91
  92         g_async_queue_push(stream->priv->queue, g_bytes_ref(bytes));
  93 }
  94
  95 static void
  96 purple_queued_output_stream_init(PurpleQueuedOutputStream *stream)
  97 {
  98         stream->priv = PURPLE_QUEUED_OUTPUT_STREAM_GET_PRIVATE(stream);
  99         stream->priv->queue =
 100                         g_async_queue_new_full((GDestroyNotify)g_bytes_unref);
 101 }
 102
 103 static void
 104 purple_queued_output_stream_dispose(GObject *object)
 105 {
 106         PurpleQueuedOutputStream *stream = PURPLE_QUEUED_OUTPUT_STREAM(object);
 107
 108         /* Chain up first in case the stream is flushed */
 109         G_OBJECT_CLASS(parent_class)->dispose(object);
 110
 111         g_clear_pointer(&stream->priv->queue, g_async_queue_unref);
 112         g_clear_pointer(&stream->priv->next, g_bytes_unref);
 113 }
 114
 115 static gboolean
 116 purple_queued_output_stream_flush(GOutputStream *stream,
 117                 GCancellable *cancellable, GError **error)
 118 {
 119         PurpleQueuedOutputStreamPrivate *priv;
 120         GOutputStream *base_stream;
 121         GBytes *bytes;
 122         const void *buffer;
 123         gsize count;
 124         gsize bytes_written = 0;
 125         gboolean ret = TRUE;
 126
 127         priv = PURPLE_QUEUED_OUTPUT_STREAM(stream)->priv;
 128         base_stream = g_filter_output_stream_get_base_stream(
 129                         G_FILTER_OUTPUT_STREAM(stream));
 130
 131         do {
 132                 if (priv->next != NULL) {
 133                         bytes = priv->next;
 134                         priv->next = NULL;
 135                 } else {
 136                         bytes = g_async_queue_try_pop(priv->queue);
 137                 }
 138
 139                 if (bytes == NULL) {
 140                         break;
 141                 }
 142
 143                 buffer = g_bytes_get_data(bytes, &count);
 144
 145                 ret = g_output_stream_write_all(base_stream, buffer, count,
 146                                 &bytes_written, cancellable, error);
 147
 148                 if (!ret) {
 149                         GBytes *queue_bytes;
 150
 151                         if (bytes_written > 0) {
 152                                 queue_bytes = g_bytes_new_from_bytes(bytes,
 153                                                 bytes_written,
 154                                                 count - bytes_written);
 155                         } else {
 156                                 queue_bytes = g_bytes_ref(bytes);
 157                         }
 158
 159                         priv->next = queue_bytes;
 160                 }
 161
 162                 g_bytes_unref(bytes);
 163         } while (ret);
 164
 165         return ret;
 166 }
 167
 168 static void
 169 purple_queued_output_stream_flush_async_cb(GObject *source,
 170                 GAsyncResult *res, gpointer user_data)
 171 {
 172         GTask *task = user_data;
 173         PurpleQueuedOutputStream *stream;
 174         gssize written;
 175         gsize size;
 176         GBytes *old_bytes;
 177         GError *error = NULL;
 178
 179         written = g_output_stream_write_bytes_finish(G_OUTPUT_STREAM(source),
 180                         res, &error);
 181
 182         if (written < 0) {
 183                 g_task_return_error(task, error);
 184                 return;
 185         }
 186
 187         stream = PURPLE_QUEUED_OUTPUT_STREAM(g_task_get_source_object(task));
 188         size = g_bytes_get_size(stream->priv->next);
 189
 190         old_bytes = stream->priv->next;
 191         stream->priv->next = NULL;
 192
 193         if (size > (gsize)written) {
 194                 stream->priv->next = g_bytes_new_from_bytes(old_bytes,
 195                                 written, size - written);
 196         }
 197
 198         g_bytes_unref(old_bytes);
 199
 200         purple_queued_output_stream_start_flush_async(task);
 201 }
 202
 203 static void
 204 purple_queued_output_stream_start_flush_async(GTask *task)
 205 {
 206         PurpleQueuedOutputStream *stream;
 207         GOutputStream *base_stream;
 208
 209         stream = PURPLE_QUEUED_OUTPUT_STREAM(g_task_get_source_object(task));
 210         base_stream = g_filter_output_stream_get_base_stream(
 211                         G_FILTER_OUTPUT_STREAM(stream));
 212
 213         if (stream->priv->next == NULL) {
 214                 stream->priv->next =
 215                                 g_async_queue_try_pop(stream->priv->queue);
 216
 217                 if (stream->priv->next == NULL) {
 218                         g_task_return_boolean(task, TRUE);
 219                         return;
 220                 }
 221         }
 222
 223         g_output_stream_write_bytes_async(base_stream, stream->priv->next,
 224                         g_task_get_priority(task),
 225                         g_task_get_cancellable(task),
 226                         purple_queued_output_stream_flush_async_cb, task);
 227 }
 228
 229 static void
 230 purple_queued_output_stream_flush_async(GOutputStream *stream,
 231                 int io_priority, GCancellable *cancellable,
 232                 GAsyncReadyCallback callback, gpointer user_data)
 233 {
 234         GTask *task;
 235
 236         task = g_task_new(stream, cancellable, callback, user_data);
 237         g_task_set_priority(task, io_priority);
 238
 239         purple_queued_output_stream_start_flush_async(task);
 240 }
 241
 242 static gboolean
 243 purple_queued_output_stream_flush_finish(GOutputStream *stream,
 244                 GAsyncResult *result, GError **error)
 245 {
 246         g_return_val_if_fail(g_task_is_valid(result, stream), FALSE);
 247
 248         return g_task_propagate_boolean(G_TASK(result), error);
 249 }
 250