/*

*

* purple

*

* Purple is the legal property of its developers, whose names are too numerous

* to list here. Please refer to the COPYRIGHT file distributed with this

* source distribution.

*

* This program is free software; you can redistribute it and/or modify

* it under the terms of the GNU General Public License as published by

* the Free Software Foundation; either version 2 of the License, or

* (at your option) any later version.

*

* This program is distributed in the hope that it will be useful,

* but WITHOUT ANY WARRANTY; without even the implied warranty of

* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the

* GNU General Public License for more details.

*

* You should have received a copy of the GNU General Public License

* along with this program; if not, write to the Free Software

* Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02111-1301 USA

*/

#include "internal.h"

#include "queuedoutputstream.h"

/**

* PurpleQueuedOutputStream:

*

* An implementation of #GFilterOutputStream which allows queuing data for

* output. This allows data to be queued while other data is being output.

* Therefore, data doesn't have to be manually stored while waiting for

* stream operations to finish.

*

* To create a queued output stream, use #purple_queued_output_stream_new().

*

* To queue data, use #purple_queued_output_stream_push_bytes_async().

*

* If there's a fatal stream error, it's suggested to clear the remaining

* bytes queued with #purple_queued_output_stream_clear_queue() to avoid

* excessive errors returned in

* #purple_queued_output_stream_push_bytes_async()'s async callback.

*/

struct _PurpleQueuedOutputStream

{

GFilterOutputStream parent;

};

typedef struct

{

GAsyncQueue *queue;

gboolean pending_queued;

} PurpleQueuedOutputStreamPrivate;

G_DEFINE_TYPE_WITH_PRIVATE(PurpleQueuedOutputStream,

purple_queued_output_stream, G_TYPE_FILTER_OUTPUT_STREAM)

/******************************************************************************

* Helpers

*****************************************************************************/

static void purple_queued_output_stream_start_push_bytes_async(GTask *task);

static void

purple_queued_output_stream_push_bytes_async_cb(GObject *source,

GAsyncResult *res, gpointer user_data)

{

GTask *task = G_TASK(user_data);

PurpleQueuedOutputStream *stream = g_task_get_source_object(task);

PurpleQueuedOutputStreamPrivate *priv = purple_queued_output_stream_get_instance_private(stream);

gssize written;

GBytes *bytes;

gsize size;

GError *error = NULL;

written = g_output_stream_write_bytes_finish(G_OUTPUT_STREAM(source),

res, &error);

bytes = g_task_get_task_data(task);

size = g_bytes_get_size(bytes);

if (written < 0) {

/* Error occurred, return error */

g_task_return_error(task, error);

g_clear_object(&task);

} else if (size > written) {

/* Partial write, prepare to send remaining data */

bytes = g_bytes_new_from_bytes(bytes, written, size - written);

g_task_set_task_data(task, bytes,

(GDestroyNotify)g_bytes_unref);

} else {

/* Full write, this task is finished */

g_task_return_boolean(task, TRUE);

g_clear_object(&task);

}

/* If g_task_return_* was called in this function, the callback

* may have cleared the stream. If so, there will be no remaining

* tasks to process here.

*/

if (task == NULL) {

/* Any queued data left? */

task = g_async_queue_try_pop(priv->queue);

}

if (task != NULL) {

/* More to process */

purple_queued_output_stream_start_push_bytes_async(task);

} else {

/* All done */

priv->pending_queued = FALSE;

g_output_stream_clear_pending(G_OUTPUT_STREAM(stream));

}

}

static void

purple_queued_output_stream_start_push_bytes_async(GTask *task)

{

PurpleQueuedOutputStream *stream = g_task_get_source_object(task);

GOutputStream *base_stream;

base_stream = g_filter_output_stream_get_base_stream(

G_FILTER_OUTPUT_STREAM(stream));

g_output_stream_write_bytes_async(base_stream,

g_task_get_task_data(task),

g_task_get_priority(task),

g_task_get_cancellable(task),

purple_queued_output_stream_push_bytes_async_cb,

task);

}

/******************************************************************************

* GObject Implementation

*****************************************************************************/

static void

purple_queued_output_stream_dispose(GObject *object)

{

PurpleQueuedOutputStream *stream = PURPLE_QUEUED_OUTPUT_STREAM(object);

PurpleQueuedOutputStreamPrivate *priv = purple_queued_output_stream_get_instance_private(stream);

g_clear_pointer(&priv->queue, g_async_queue_unref);

G_OBJECT_CLASS(purple_queued_output_stream_parent_class)->dispose(object);

}

static void

purple_queued_output_stream_class_init(PurpleQueuedOutputStreamClass *klass)

{

GObjectClass *obj_class = G_OBJECT_CLASS(klass);

obj_class->dispose = purple_queued_output_stream_dispose;

}

static void

purple_queued_output_stream_init(PurpleQueuedOutputStream *stream)

{

PurpleQueuedOutputStreamPrivate *priv = purple_queued_output_stream_get_instance_private(stream);

priv->queue = g_async_queue_new_full((GDestroyNotify)g_bytes_unref);

priv->pending_queued = FALSE;

}

/******************************************************************************

* Public API

*****************************************************************************/

PurpleQueuedOutputStream *

purple_queued_output_stream_new(GOutputStream *base_stream)

{

PurpleQueuedOutputStream *stream;

g_return_val_if_fail(G_IS_OUTPUT_STREAM(base_stream), NULL);

stream = g_object_new(PURPLE_TYPE_QUEUED_OUTPUT_STREAM,

"base-stream", base_stream,

NULL);

return stream;

}

void

purple_queued_output_stream_push_bytes_async(PurpleQueuedOutputStream *stream,

GBytes *bytes, int io_priority, GCancellable *cancellable,

GAsyncReadyCallback callback, gpointer user_data)

{

GTask *task;

gboolean set_pending;

GError *error = NULL;

PurpleQueuedOutputStreamPrivate *priv = NULL;

g_return_if_fail(PURPLE_IS_QUEUED_OUTPUT_STREAM(stream));

g_return_if_fail(bytes != NULL);

priv = purple_queued_output_stream_get_instance_private(stream);

task = g_task_new(stream, cancellable, callback, user_data);

g_task_set_task_data(task, g_bytes_ref(bytes),

(GDestroyNotify)g_bytes_unref);

g_task_set_source_tag(task,

purple_queued_output_stream_push_bytes_async);

g_task_set_priority(task, io_priority);

set_pending = g_output_stream_set_pending(

G_OUTPUT_STREAM(stream), &error);

/* Since we're allowing queuing requests without blocking,

* it's not an error to be pending while processing queued operations.

*/

if (!set_pending && (!g_error_matches(error,

G_IO_ERROR, G_IO_ERROR_PENDING) ||

!priv->pending_queued)) {

g_task_return_error(task, error);

g_object_unref(task);

return;

}

priv->pending_queued = TRUE;

if (set_pending) {

/* Start processing if there were no pending operations */

purple_queued_output_stream_start_push_bytes_async(task);

} else {

/* Otherwise queue the data */

g_async_queue_push(priv->queue, task);

}

}

gboolean

purple_queued_output_stream_push_bytes_finish(PurpleQueuedOutputStream *stream,

GAsyncResult *result, GError **error)

{

g_return_val_if_fail(PURPLE_IS_QUEUED_OUTPUT_STREAM(stream), FALSE);

g_return_val_if_fail(g_task_is_valid(result, stream), FALSE);

g_return_val_if_fail(g_async_result_is_tagged(result,

purple_queued_output_stream_push_bytes_async), FALSE);

return g_task_propagate_boolean(G_TASK(result), error);

}

void

purple_queued_output_stream_clear_queue(PurpleQueuedOutputStream *stream)

{

GTask *task;

PurpleQueuedOutputStreamPrivate *priv = NULL;

g_return_if_fail(PURPLE_IS_QUEUED_OUTPUT_STREAM(stream));

priv = purple_queued_output_stream_get_instance_private(stream);

while ((task = g_async_queue_try_pop(priv->queue)) != NULL) {

g_task_return_new_error(task, G_IO_ERROR, G_IO_ERROR_CANCELLED,

"PurpleQueuedOutputStream queue cleared");

g_object_unref(task);

}

}