/* -*- Mode: C; indent-tabs-mode: t; c-basic-offset: 8; tab-width: 8 -*- */

/*

 * GData Client

 * Copyright (C) Philip Withnall 2009 <philip@tecnocode.co.uk>

 *

 * GData Client is free software; you can redistribute it and/or

 * modify it under the terms of the GNU Lesser General Public

 * License as published by the Free Software Foundation; either

 * version 2.1 of the License, or (at your option) any later version.

 *

 * GData Client 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

 * Lesser General Public License for more details.

 *

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

 * License along with GData Client.  If not, see <http://www.gnu.org/licenses/>.

 */

/**

 * SECTION:gdata-upload-stream

 * @short_description: GData upload stream object

 * @stability: Stable

 * @include: gdata/gdata-upload-stream.h

 *

 * #GDataUploadStream is a #GOutputStream subclass to allow uploading of files from GData services with authorization from a #GDataService under

 * the given #GDataAuthorizationDomain. If authorization is not required to perform the upload, a #GDataAuthorizationDomain doesn't have to be

 * specified.

 *

 * Once a #GDataUploadStream is instantiated with gdata_upload_stream_new(), the standard #GOutputStream API can be used on the stream to upload

 * the file. Network communication may not actually begin until the first call to g_output_stream_write(), so having a #GDataUploadStream around is no

 * guarantee that data is being uploaded.

 *

 * Uploads of a file, or a file with associated metadata (a #GDataEntry) should use #GDataUploadStream, but if you want to simply upload a single

 * #GDataEntry, use gdata_service_insert_entry() instead. #GDataUploadStream is for large streaming uploads.

 *

 * Once an upload is complete, the server's response can be retrieved from the #GDataUploadStream using gdata_upload_stream_get_response(). In order

 * for network communication to be guaranteed to have stopped (and thus the response definitely available), g_output_stream_close() must be called

 * on the #GDataUploadStream first. Otherwise, gdata_upload_stream_get_response() may return saying that the operation is still in progress.

 *

 * If the server returns an error instead of a success response, the error will be returned by g_output_stream_close() as a #GDataServiceError.

 *

 * The entire upload operation can be cancelled using the #GCancellable instance provided to gdata_upload_stream_new(), or returned by

 * gdata_upload_stream_get_cancellable(). Cancelling this at any time will cause all future #GOutputStream method calls to return

 * %G_IO_ERROR_CANCELLED. If any #GOutputStream methods are in the process of being called, they will be cancelled and return %G_IO_ERROR_CANCELLED as

 * soon as possible.

 *

 * Note that cancelling an individual method call (such as a call to g_output_stream_write()) using the #GCancellable parameter of the method will not

 * cancel the upload as a whole — just that particular method call. In the case of g_output_stream_write(), this will cause it to return the number of

 * bytes it has successfully written up to the point of cancellation (up to the requested number of bytes), or return a %G_IO_ERROR_CANCELLED if it

 * had not managed to write any bytes to the network by that point. This is also the behaviour of g_output_stream_write() when the upload operation as

 * a whole is cancelled.

 *

 * In the case of g_output_stream_close(), the call will return immediately if network activity hasn't yet started. If it has, the network activity

 * will be cancelled, regardless of whether the call to g_output_stream_close() is cancelled. Cancelling a pending call to g_output_stream_close()

 * (either using the method's #GCancellable, or by cancelling the upload stream as a whole) will cause it to stop waiting for the network activity to

 * finish, and return %G_IO_ERROR_CANCELLED immediately. Network activity will continue to be shut down in the background.

 *

 * Any outstanding data is guaranteed to be written to the network successfully even if a call to g_output_stream_close() is cancelled. However, if

 * the upload stream as a whole is cancelled using #GDataUploadStream:cancellable, no more data will be sent over the network, and the network

 * connection will be closed immediately. i.e. #GDataUploadStream will do its best to instruct the server to cancel the upload and any associated

 * server-side changes of state.

 *

 * If the server returns an error message (for example, if the user is not correctly authenticated/authorized or doesn't have suitable permissions

 * to upload from the given URI), it will be returned as a #GDataServiceError by g_output_stream_close().

 *

 * <example>

 * 	<title>Uploading from a File</title>

 * 	<programlisting>

 *	GDataService *service;

 *	GDataAuthorizationDomain *domain;

 *	GCancellable *cancellable;

 *	GInputStream *input_stream;

 *	GOutputStream *upload_stream;

 *	GFile *file;

 *	GFileInfo *file_info;

 *	GError *error = NULL;

 *

 *	/* Get the file to upload */

 *	file = get_file_to_upload ();

 *	file_info = g_file_query_info (file, G_FILE_ATTRIBUTE_STANDARD_DISPLAY_NAME "," G_FILE_ATTRIBUTE_STANDARD_CONTENT_TYPE ","

 *	                               G_FILE_ATTRIBUTE_STANDARD_SIZE,

 *	                               G_FILE_QUERY_INFO_NONE, NULL, &error);

 *

 *	if (file_info == NULL) {

 *		g_error ("Error getting file info: %s", error->message);

 *		g_error_free (error);

 *		g_object_unref (file);

 *		return;

 *	}

 *

 *	input_stream = g_file_read (file, NULL, &error);

 *	g_object_unref (file);

 *

 *	if (input_stream == NULL) {

 *		g_error ("Error getting file input stream: %s", error->message);

 *		g_error_free (error);

 *		g_object_unref (file_info);

 *		return;

 *	}

 *

 *	/* Create the upload stream */

 *	service = create_my_service ();

 *	domain = get_my_authorization_domain_from_service (service);

 *	cancellable = g_cancellable_new (); /* cancel this to cancel the entire upload operation */

 *	upload_stream = gdata_upload_stream_new_resumable (service, domain, SOUP_METHOD_POST, upload_uri, NULL,

 *	                                                   g_file_info_get_display_name (file_info), g_file_info_get_content_type (file_info),

 *	                                                   g_file_info_get_size (file_info), cancellable);

 *	g_object_unref (file_info);

 *

 *	/* Perform the upload asynchronously */

 *	g_output_stream_splice_async (upload_stream, input_stream, G_OUTPUT_STREAM_SPLICE_CLOSE_SOURCE | G_OUTPUT_STREAM_SPLICE_CLOSE_TARGET,

 *	                              G_PRIORITY_DEFAULT, NULL, (GAsyncReadyCallback) upload_splice_cb, NULL);

 *

 *	g_object_unref (upload_stream);

 *	g_object_unref (input_stream);

 *	g_object_unref (cancellable);

 *	g_object_unref (domain);

 *	g_object_unref (service);

 *

 *	static void

 *	upload_splice_cb (GOutputStream *upload_stream, GAsyncResult *result, gpointer user_data)

 *	{

 *		gssize length;

 *		GError *error = NULL;

 *

 *		g_output_stream_splice_finish (upload_stream, result, &error);

 *

 *		if (error != NULL && g_error_matches (error, G_IO_ERROR, G_IO_ERROR_CANCELLED) == FALSE)) {

 *			/* Error upload the file; potentially an I/O error (GIOError), or an error response from the server

 *			 * (GDataServiceError). */

 *			g_error ("Error uploading file: %s", error->message);

 *			g_error_free (error);

 *		}

 *

 *		/* If the upload was successful, carry on to parse the result. Note that this will normally be handled by methods like

 *		 * gdata_youtube_service_finish_video_upload(), gdata_picasaweb_service_finish_file_upload() and

 *		 * gdata_documents_service_finish_upload() */

 *		parse_server_result (gdata_upload_stream_get_response (GDATA_UPLOAD_STREAM (upload_stream), &length), length);

 *	}

 * 	</programlisting>

 * </example>

 *

 * Since: 0.5.0

 */

/*

 * We have a network thread which does all the uploading work. We send the message encoded as chunks, but cannot use the SoupMessageBody as a

 * data buffer, since it can only ever be touched by the network thread. Instead, we pass data to the network thread through a GDataBuffer, with

 * the main thread pushing it on as and when write() is called. The network thread cannot block on popping data off the buffer, as it requests fixed-

 * size chunks, and there's no way to notify it that we've reached EOF; so when it gets to popping the last chunk off the buffer, which may well be

 * smaller than its chunk size, it would block for more data and therefore hang. Consequently, the network thread instead pops as much data as it can

 * off the buffer, up to its chunk size, which is a non-blocking operation.

 *

 * The write() and close() operations on the output stream are synchronised with the network thread, so that the write() call only returns once the

 * network thread has written at least as many bytes as were passed to the write() call, and the close() call only returns once all network activity

 * has finished (including receiving the response from the server). Async versions of these calls are provided by GOutputStream.

 *

 * The number of bytes in the various buffers are recorded using:

 *  • message_bytes_outstanding: the number of bytes in the GDataBuffer which are waiting to be written to the SoupMessageBody

 *  • network_bytes_outstanding: the number of bytes which have been written to the SoupMessageBody, and are waiting to be written to the network

 *  • network_bytes_written: the total number of bytes which have been successfully written to the network

 *

 * Mutex locking order:

 *  1. response_mutex

 *  2. write_mutex

 */

#include <config.h>

#include <glib.h>

#include <glib/gi18n-lib.h>

#include <string.h>

#include "gdata-upload-stream.h"

#include "gdata-buffer.h"

#include "gdata-private.h"

#define BOUNDARY_STRING "0003Z5W789deadbeefRTE456KlemsnoZV"

#define MAX_RESUMABLE_CHUNK_SIZE (512 * 1024) /* bytes = 512 KiB */

static void gdata_upload_stream_constructed (GObject *object);

static void gdata_upload_stream_dispose (GObject *object);

static void gdata_upload_stream_finalize (GObject *object);

static void gdata_upload_stream_get_property (GObject *object, guint property_id, GValue *value, GParamSpec *pspec);

static void gdata_upload_stream_set_property (GObject *object, guint property_id, const GValue *value, GParamSpec *pspec);

static gssize gdata_upload_stream_write (GOutputStream *stream, const void *buffer, gsize count, GCancellable *cancellable, GError **error);

static gboolean gdata_upload_stream_flush (GOutputStream *stream, GCancellable *cancellable, GError **error);

static gboolean gdata_upload_stream_close (GOutputStream *stream, GCancellable *cancellable, GError **error);

static void create_network_thread (GDataUploadStream *self, GError **error);

typedef enum {

	STATE_INITIAL_REQUEST, /* initial POST request to the resumable-create-media link (unused for non-resumable uploads) */

	STATE_DATA_REQUESTS, /* one or more subsequent PUT requests (only state used for non-resumable uploads) */

	STATE_FINISHED, /* finished successfully or in error */

} UploadState;

struct _GDataUploadStreamPrivate {

	gchar *method;

	gchar *upload_uri;

	GDataService *service;

	GDataAuthorizationDomain *authorization_domain;

	GDataEntry *entry;

	gchar *slug;

	gchar *content_type;

	goffset content_length; /* -1 for non-resumable uploads; 0 or greater for resumable ones */

	SoupSession *session;

	SoupMessage *message;

	GDataBuffer *buffer;

	GCancellable *cancellable;

	GThread *network_thread;

	UploadState state; /* protected by write_mutex */

	GMutex write_mutex; /* mutex for write operations (specifically, write_finished) */

	/* This persists across all resumable upload chunks. Note that it doesn't count bytes from the entry XML. */

	gsize total_network_bytes_written; /* the number of bytes which have been written to the network in STATE_DATA_REQUESTS */

	/* All of the following apply only to the current resumable upload chunk. */

	gsize message_bytes_outstanding; /* the number of bytes which have been written to the buffer but not libsoup (signalled by write_cond) */

	gsize network_bytes_outstanding; /* the number of bytes which have been written to libsoup but not the network (signalled by write_cond) */

	gsize network_bytes_written; /* the number of bytes which have been written to the network (signalled by write_cond) */

	gsize chunk_size; /* the size of the current chunk (in bytes); 0 iff content_length <= 0; must be <= MAX_RESUMABLE_CHUNK_SIZE */

	GCond write_cond; /* signalled when a chunk has been written (protected by write_mutex) */

	GCond finished_cond; /* signalled when sending the message (and receiving the response) is finished (protected by response_mutex) */

	guint response_status; /* set once we finish receiving the response (SOUP_STATUS_NONE otherwise) (protected by response_mutex) */

	GError *response_error; /* error asynchronously set by the network thread, and picked up by the main thread when appropriate */

	GMutex response_mutex; /* mutex for ->response_error, ->response_status and ->finished_cond */

};

enum {

	PROP_SERVICE = 1,

	PROP_UPLOAD_URI,

	PROP_ENTRY,

	PROP_SLUG,

	PROP_CONTENT_TYPE,

	PROP_METHOD,

	PROP_CANCELLABLE,

	PROP_AUTHORIZATION_DOMAIN,

	PROP_CONTENT_LENGTH,

};

G_DEFINE_TYPE (GDataUploadStream, gdata_upload_stream, G_TYPE_OUTPUT_STREAM)

static void

gdata_upload_stream_class_init (GDataUploadStreamClass *klass)

{

	GObjectClass *gobject_class = G_OBJECT_CLASS (klass);

	GOutputStreamClass *stream_class = G_OUTPUT_STREAM_CLASS (klass);

	g_type_class_add_private (klass, sizeof (GDataUploadStreamPrivate));

	gobject_class->constructed = gdata_upload_stream_constructed;

	gobject_class->dispose = gdata_upload_stream_dispose;

	gobject_class->finalize = gdata_upload_stream_finalize;

	gobject_class->get_property = gdata_upload_stream_get_property;

	gobject_class->set_property = gdata_upload_stream_set_property;

	/* We use the default implementations of the async functions, which just run

	 * our implementation of the sync function in a thread. */

	stream_class->write_fn = gdata_upload_stream_write;

	stream_class->flush = gdata_upload_stream_flush;

	stream_class->close_fn = gdata_upload_stream_close;

	/**

	 * GDataUploadStream:service:

	 *

	 * The service which is used to authorize the upload, and to which the upload relates.

	 *

	 * Since: 0.5.0

	 */

	g_object_class_install_property (gobject_class, PROP_SERVICE,

	                                 g_param_spec_object ("service",

	                                                      "Service", "The service which is used to authorize the upload.",

	                                                      GDATA_TYPE_SERVICE,

	                                                      G_PARAM_CONSTRUCT_ONLY | G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));

	/**

	 * GDataUploadStream:authorization-domain:

	 *

	 * The authorization domain for the upload, against which the #GDataService:authorizer for the #GDataDownloadStream:service should be

	 * authorized. This may be %NULL if authorization is not needed for the upload.

	 *

	 * Since: 0.9.0

	 */

	g_object_class_install_property (gobject_class, PROP_AUTHORIZATION_DOMAIN,

	                                 g_param_spec_object ("authorization-domain",

	                                                      "Authorization domain", "The authorization domain for the upload.",

	                                                      GDATA_TYPE_AUTHORIZATION_DOMAIN,

	                                                      G_PARAM_CONSTRUCT_ONLY | G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));

	/**

	 * GDataUploadStream:method:

	 *

	 * The HTTP request method to use when uploading the file.

	 *

	 * Since: 0.7.0

	 */

	g_object_class_install_property (gobject_class, PROP_METHOD,

	                                 g_param_spec_string ("method",

	                                                      "Method", "The HTTP request method to use when uploading the file.",

	                                                      NULL,

	                                                      G_PARAM_CONSTRUCT_ONLY | G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));

	/**

	 * GDataUploadStream:upload-uri:

	 *

	 * The URI to upload the data and metadata to. This must be HTTPS.

	 *

	 * Since: 0.5.0

	 */

	g_object_class_install_property (gobject_class, PROP_UPLOAD_URI,

	                                 g_param_spec_string ("upload-uri",

	                                                      "Upload URI", "The URI to upload the data and metadata to.",

	                                                      NULL,

	                                                      G_PARAM_CONSTRUCT_ONLY | G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));

	/**

	 * GDataUploadStream:entry:

	 *

	 * The entry used for metadata to upload.

	 *

	 * Since: 0.5.0

	 */

	g_object_class_install_property (gobject_class, PROP_ENTRY,

	                                 g_param_spec_object ("entry",

	                                                      "Entry", "The entry used for metadata to upload.",

	                                                      GDATA_TYPE_ENTRY,

	                                                      G_PARAM_CONSTRUCT_ONLY | G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));

	/**

	 * GDataUploadStream:slug:

	 *

	 * The slug of the file being uploaded. This is usually the display name of the file (i.e. as returned by g_file_info_get_display_name()).

	 *

	 * Since: 0.5.0

	 */

	g_object_class_install_property (gobject_class, PROP_SLUG,

	                                 g_param_spec_string ("slug",

	                                                      "Slug", "The slug of the file being uploaded.",

	                                                      NULL,

	                                                      G_PARAM_CONSTRUCT_ONLY | G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));

	/**

	 * GDataUploadStream:content-length:

	 *

	 * The content length (in bytes) of the file being uploaded (i.e. as returned by g_file_info_get_size()). Note that this does not include the

	 * length of the XML serialisation of #GDataUploadStream:entry, if set.

	 *

	 * If this is -1 the upload will be non-resumable; if it is non-negative, the upload will be resumable.

	 *

	 * Since: 0.13.0

	 */

	g_object_class_install_property (gobject_class, PROP_CONTENT_LENGTH,

	                                 g_param_spec_int64 ("content-length",

	                                                     "Content length", "The content length (in bytes) of the file being uploaded.",

	                                                     -1, G_MAXINT64, -1,

	                                                     G_PARAM_CONSTRUCT_ONLY | G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));

	/**

	 * GDataUploadStream:content-type:

	 *

	 * The content type of the file being uploaded (i.e. as returned by g_file_info_get_content_type()).

	 *

	 * Since: 0.5.0

	 */

	g_object_class_install_property (gobject_class, PROP_CONTENT_TYPE,

	                                 g_param_spec_string ("content-type",

	                                                      "Content type", "The content type of the file being uploaded.",

	                                                      NULL,

	                                                      G_PARAM_CONSTRUCT_ONLY | G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));

	/**

	 * GDataUploadStream:cancellable:

	 *

	 * An optional cancellable used to cancel the entire upload operation. If a #GCancellable instance isn't provided for this property at

	 * construction time (i.e. to gdata_upload_stream_new()), one will be created internally and can be retrieved using

	 * gdata_upload_stream_get_cancellable() and used to cancel the upload operation with g_cancellable_cancel() just as if it was passed to

	 * gdata_upload_stream_new().

	 *

	 * If the upload operation is cancelled using this #GCancellable, any ongoing network activity will be stopped, and any pending or future calls

	 * to #GOutputStream API on the #GDataUploadStream will return %G_IO_ERROR_CANCELLED. Note that the #GCancellable objects which can be passed

	 * to individual #GOutputStream operations will not cancel the upload operation proper if cancelled — they will merely cancel that API call.

	 * The only way to cancel the upload operation completely is using #GDataUploadStream:cancellable.

	 *

	 * Since: 0.8.0

	 */

	g_object_class_install_property (gobject_class, PROP_CANCELLABLE,

	                                 g_param_spec_object ("cancellable",

	                                                      "Cancellable", "An optional cancellable used to cancel the entire upload operation.",

	                                                      G_TYPE_CANCELLABLE,

	                                                      G_PARAM_CONSTRUCT_ONLY | G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));

}

static void

gdata_upload_stream_init (GDataUploadStream *self)

{

	self->priv = G_TYPE_INSTANCE_GET_PRIVATE (self, GDATA_TYPE_UPLOAD_STREAM, GDataUploadStreamPrivate);

	self->priv->buffer = gdata_buffer_new ();

	g_mutex_init (&(self->priv->write_mutex));

	g_cond_init (&(self->priv->write_cond));

	g_cond_init (&(self->priv->finished_cond));

	g_mutex_init (&(self->priv->response_mutex));

}

static SoupMessage *

build_message (GDataUploadStream *self, const gchar *method, const gchar *upload_uri)

{

	SoupMessage *new_message;

	SoupURI *_uri;

	/* Build the message */

	_uri = soup_uri_new (upload_uri);

	soup_uri_set_port (_uri, _gdata_service_get_https_port ());

	new_message = soup_message_new_from_uri (method, _uri);

	soup_uri_free (_uri);

	/* We don't want to accumulate chunks */

	soup_message_body_set_accumulate (new_message->request_body, FALSE);

	return new_message;

}

static void

gdata_upload_stream_constructed (GObject *object)

{

	GDataUploadStreamPrivate *priv;

	GDataServiceClass *service_klass;

	SoupURI *uri = NULL;

	/* Chain up to the parent class */

	G_OBJECT_CLASS (gdata_upload_stream_parent_class)->constructed (object);

	priv = GDATA_UPLOAD_STREAM (object)->priv;

	/* The upload URI must be HTTPS. */

	uri = soup_uri_new (priv->upload_uri);

	g_assert_cmpstr (soup_uri_get_scheme (uri), ==, SOUP_URI_SCHEME_HTTPS);

	soup_uri_free (uri);

	/* Create a #GCancellable for the entire upload operation if one wasn't specified for #GDataUploadStream:cancellable during construction */

	if (priv->cancellable == NULL)

		priv->cancellable = g_cancellable_new ();

	/* Build the message */

	priv->message = build_message (GDATA_UPLOAD_STREAM (object), priv->method, priv->upload_uri);

	if (priv->slug != NULL)

		soup_message_headers_append (priv->message->request_headers, "Slug", priv->slug);

	if (priv->content_length == -1) {

		/* Non-resumable upload */

		soup_message_headers_set_encoding (priv->message->request_headers, SOUP_ENCODING_CHUNKED);

		/* The Content-Type should be multipart/related if we're also uploading the metadata (entry != NULL),

		 * and the given content_type otherwise. */

		if (priv->entry != NULL) {

			gchar *first_part_header, *upload_data;

			gchar *second_part_header;

			GDataParsableClass *parsable_klass;

			parsable_klass = GDATA_PARSABLE_GET_CLASS (priv->entry);

			g_assert (parsable_klass->get_content_type != NULL);

			soup_message_headers_set_content_type (priv->message->request_headers, "multipart/related; boundary=" BOUNDARY_STRING, NULL);

			if (g_strcmp0 (parsable_klass->get_content_type (), "application/json") == 0) {

				upload_data = gdata_parsable_get_json (GDATA_PARSABLE (priv->entry));

			} else {

				upload_data = gdata_parsable_get_xml (GDATA_PARSABLE (priv->entry));

			}

			/* Start by writing out the entry; then the thread has something to write to the network when it's created */

			first_part_header = g_strdup_printf ("--" BOUNDARY_STRING "\n"

			                                     "Content-Type: %s; charset=UTF-8\n\n",

			                                     parsable_klass->get_content_type ());

			second_part_header = g_strdup_printf ("\n--" BOUNDARY_STRING "\n"

			                                      "Content-Type: %s\n"

			                                      "Content-Transfer-Encoding: binary\n\n",

			                                      priv->content_type);

			/* Push the message parts onto the message body; we can skip the buffer, since the network thread hasn't yet been created,

			 * so we're the sole thread accessing the SoupMessage. */

			soup_message_body_append (priv->message->request_body,

			                          SOUP_MEMORY_TAKE,

			                          first_part_header,

			                          strlen (first_part_header));

			soup_message_body_append (priv->message->request_body,

			                          SOUP_MEMORY_TAKE, upload_data,

			                          strlen (upload_data));

			soup_message_body_append (priv->message->request_body,

			                          SOUP_MEMORY_TAKE,

			                          second_part_header,

			                          strlen (second_part_header));

			first_part_header = NULL;

			upload_data = NULL;

			second_part_header = NULL;

			priv->network_bytes_outstanding = priv->message->request_body->length;

		} else {

			soup_message_headers_set_content_type (priv->message->request_headers, priv->content_type, NULL);

		}

		/* Non-resumable uploads start with the data requests immediately. */

		priv->state = STATE_DATA_REQUESTS;

	} else {

		gchar *content_length_str;

		/* Resumable upload's initial request */

		soup_message_headers_set_encoding (priv->message->request_headers, SOUP_ENCODING_CONTENT_LENGTH);

		soup_message_headers_replace (priv->message->request_headers, "X-Upload-Content-Type", priv->content_type);

		content_length_str = g_strdup_printf ("%" G_GOFFSET_FORMAT, priv->content_length);

		soup_message_headers_replace (priv->message->request_headers, "X-Upload-Content-Length", content_length_str);

		g_free (content_length_str);

		if (priv->entry != NULL) {

			GDataParsableClass *parsable_klass;

			gchar *content_type, *upload_data;

			parsable_klass = GDATA_PARSABLE_GET_CLASS (priv->entry);

			g_assert (parsable_klass->get_content_type != NULL);

			if (g_strcmp0 (parsable_klass->get_content_type (), "application/json") == 0) {

				upload_data = gdata_parsable_get_json (GDATA_PARSABLE (priv->entry));

			} else {

				upload_data = gdata_parsable_get_xml (GDATA_PARSABLE (priv->entry));

			}

			content_type = g_strdup_printf ("%s; charset=UTF-8",

			                                parsable_klass->get_content_type ());

			soup_message_headers_set_content_type (priv->message->request_headers,

			                                       content_type,

			                                       NULL);

			g_free (content_type);

			soup_message_body_append (priv->message->request_body,

			                          SOUP_MEMORY_TAKE,

			                          upload_data,

			                          strlen (upload_data));

			upload_data = NULL;

			priv->network_bytes_outstanding = priv->message->request_body->length;

		} else {

			soup_message_headers_set_content_length (priv->message->request_headers, 0);

		}

		/* Resumable uploads always start with an initial request, which either contains the XML or is empty. */

		priv->state = STATE_INITIAL_REQUEST;

		priv->chunk_size = MIN (priv->content_length, MAX_RESUMABLE_CHUNK_SIZE);

	}

	/* Make sure the headers are set. HACK: This should actually be in build_message(), but we have to work around

	 * http://code.google.com/a/google.com/p/apps-api-issues/issues/detail?id=3033 in GDataDocumentsService's append_query_headers(). */

	service_klass = GDATA_SERVICE_GET_CLASS (priv->service);

	if (service_klass->append_query_headers != NULL) {

		service_klass->append_query_headers (priv->service, priv->authorization_domain, priv->message);

	}

	/* If the entry exists and has an ETag, we assume we're updating the entry, so we can set the If-Match header */

	if (priv->entry != NULL && gdata_entry_get_etag (priv->entry) != NULL)

		soup_message_headers_append (priv->message->request_headers, "If-Match", gdata_entry_get_etag (priv->entry));

	/* Uploading doesn't actually start until the first call to write() */

}

static void

gdata_upload_stream_dispose (GObject *object)

{

	GDataUploadStreamPrivate *priv = GDATA_UPLOAD_STREAM (object)->priv;

	/* Close the stream before unreffing things like priv->service, which stops crashes like bgo#602156 if the stream is unreffed in the middle

	 * of network operations */

	g_output_stream_close (G_OUTPUT_STREAM (object), NULL, NULL);

	if (priv->cancellable != NULL)

		g_object_unref (priv->cancellable);

	priv->cancellable = NULL;

	if (priv->service != NULL)

		g_object_unref (priv->service);

	priv->service = NULL;

	if (priv->authorization_domain != NULL)

		g_object_unref (priv->authorization_domain);

	priv->authorization_domain = NULL;

	if (priv->message != NULL)

		g_object_unref (priv->message);

	priv->message = NULL;

	if (priv->entry != NULL)

		g_object_unref (priv->entry);

	priv->entry = NULL;

	/* Chain up to the parent class */

	G_OBJECT_CLASS (gdata_upload_stream_parent_class)->dispose (object);

}

static void

gdata_upload_stream_finalize (GObject *object)

{

	GDataUploadStreamPrivate *priv = GDATA_UPLOAD_STREAM (object)->priv;

	g_mutex_clear (&(priv->response_mutex));

	g_cond_clear (&(priv->finished_cond));

	g_cond_clear (&(priv->write_cond));

	g_mutex_clear (&(priv->write_mutex));

	gdata_buffer_free (priv->buffer);

	g_clear_error (&(priv->response_error));

	g_free (priv->upload_uri);

	g_free (priv->method);

	g_free (priv->slug);

	g_free (priv->content_type);

	/* Chain up to the parent class */

	G_OBJECT_CLASS (gdata_upload_stream_parent_class)->finalize (object);

}

static void

gdata_upload_stream_get_property (GObject *object, guint property_id, GValue *value, GParamSpec *pspec)

{

	GDataUploadStreamPrivate *priv = GDATA_UPLOAD_STREAM (object)->priv;

	switch (property_id) {

		case PROP_SERVICE:

			g_value_set_object (value, priv->service);

			break;

		case PROP_AUTHORIZATION_DOMAIN:

			g_value_set_object (value, priv->authorization_domain);

			break;

		case PROP_METHOD:

			g_value_set_string (value, priv->method);

			break;

		case PROP_UPLOAD_URI:

			g_value_set_string (value, priv->upload_uri);

			break;

		case PROP_ENTRY:

			g_value_set_object (value, priv->entry);

			break;

		case PROP_SLUG:

			g_value_set_string (value, priv->slug);

			break;

		case PROP_CONTENT_TYPE:

			g_value_set_string (value, priv->content_type);

			break;

		case PROP_CONTENT_LENGTH:

			g_value_set_int64 (value, priv->content_length);

			break;

		case PROP_CANCELLABLE:

			g_value_set_object (value, priv->cancellable);

			break;

		default:

			/* We don't have any other property... */

			G_OBJECT_WARN_INVALID_PROPERTY_ID (object, property_id, pspec);

			break;

	}

}

static void

gdata_upload_stream_set_property (GObject *object, guint property_id, const GValue *value, GParamSpec *pspec)

{

	GDataUploadStreamPrivate *priv = GDATA_UPLOAD_STREAM (object)->priv;

	switch (property_id) {

		case PROP_SERVICE:

			priv->service = g_value_dup_object (value);

			priv->session = _gdata_service_get_session (priv->service);

			break;

		case PROP_AUTHORIZATION_DOMAIN:

			priv->authorization_domain = g_value_dup_object (value);

			break;

		case PROP_METHOD:

			priv->method = g_value_dup_string (value);

			break;

		case PROP_UPLOAD_URI:

			priv->upload_uri = g_value_dup_string (value);

			break;

		case PROP_ENTRY:

			priv->entry = g_value_dup_object (value);

			break;

		case PROP_SLUG:

			priv->slug = g_value_dup_string (value);

			break;

		case PROP_CONTENT_TYPE:

			priv->content_type = g_value_dup_string (value);

			break;

		case PROP_CONTENT_LENGTH:

			priv->content_length = g_value_get_int64 (value);

			break;

		case PROP_CANCELLABLE:

			/* Construction only */

			priv->cancellable = g_value_dup_object (value);

			break;

		default:

			/* We don't have any other property... */

			G_OBJECT_WARN_INVALID_PROPERTY_ID (object, property_id, pspec);

			break;

	}

}

typedef struct {

	GDataUploadStream *upload_stream;

	gboolean *cancelled;

} CancelledData;

static void

write_cancelled_cb (GCancellable *cancellable, CancelledData *data)

{

	GDataUploadStreamPrivate *priv = data->upload_stream->priv;

	/* Signal the gdata_upload_stream_write() function that it should stop blocking and cancel */

	g_mutex_lock (&(priv->write_mutex));

	*(data->cancelled) = TRUE;

	g_cond_signal (&(priv->write_cond));

	g_mutex_unlock (&(priv->write_mutex));

}

static gssize

gdata_upload_stream_write (GOutputStream *stream, const void *buffer, gsize count, GCancellable *cancellable, GError **error_out)

{

	GDataUploadStreamPrivate *priv = GDATA_UPLOAD_STREAM (stream)->priv;

	gssize length_written = -1;

	gulong cancelled_signal = 0, global_cancelled_signal = 0;

	gboolean cancelled = FALSE; /* must only be touched with ->write_mutex held */

	gsize old_total_network_bytes_written;

	CancelledData data;

	GError *error = NULL;

	/* Listen for cancellation events */

	data.upload_stream = GDATA_UPLOAD_STREAM (stream);

	data.cancelled = &cancelled;

	global_cancelled_signal = g_cancellable_connect (priv->cancellable, (GCallback) write_cancelled_cb, &data, NULL);

	if (cancellable != NULL)

		cancelled_signal = g_cancellable_connect (cancellable, (GCallback) write_cancelled_cb, &data, NULL);

	/* Check for an error and return if necessary */

	g_mutex_lock (&(priv->write_mutex));

	if (cancelled == TRUE) {

		g_assert (g_cancellable_set_error_if_cancelled (cancellable, &error) == TRUE ||

		          g_cancellable_set_error_if_cancelled (priv->cancellable, &error) == TRUE);

		g_mutex_unlock (&(priv->write_mutex));

		length_written = -1;

		goto done;

	}

	g_mutex_unlock (&(priv->write_mutex));

	/* Increment the number of bytes outstanding for the new write, and keep a record of the old number written so we know if the write's

	 * finished before we reach write_cond. */

	old_total_network_bytes_written = priv->total_network_bytes_written;

	priv->message_bytes_outstanding += count;

	/* Handle the more common case of the network thread already having been created first */

	if (priv->network_thread != NULL) {

		/* Push the new data into the buffer */

		gdata_buffer_push_data (priv->buffer, buffer, count);

		goto write;

	}

	/* Write out the first chunk of data, so there's guaranteed to be something in the buffer */

	gdata_buffer_push_data (priv->buffer, buffer, count);

	/* Create the thread and let the writing commence! */

	create_network_thread (GDATA_UPLOAD_STREAM (stream), &error);

	if (priv->network_thread == NULL) {

		length_written = -1;

		goto done;

	}

write:

	g_mutex_lock (&(priv->write_mutex));

	/* Wait for it to be written */

	while (priv->total_network_bytes_written - old_total_network_bytes_written < count && cancelled == FALSE && priv->state != STATE_FINISHED) {

		g_cond_wait (&(priv->write_cond), &(priv->write_mutex));

	}

	length_written = MIN (count, priv->total_network_bytes_written - old_total_network_bytes_written);

	/* Check for an error and return if necessary */

	if (cancelled == TRUE && length_written < 1) {

		/* Cancellation. */

		g_assert (g_cancellable_set_error_if_cancelled (cancellable, &error) == TRUE ||

		          g_cancellable_set_error_if_cancelled (priv->cancellable, &error) == TRUE);

		length_written = -1;

	} else if (priv->state == STATE_FINISHED && (length_written < 0 || (gsize) length_written < count)) {

		/* Resumable upload error. */

		g_set_error (&error, G_IO_ERROR, G_IO_ERROR_FAILED, _("Error received from server after uploading a resumable upload chunk."));

		length_written = -1;

	}

	g_mutex_unlock (&(priv->write_mutex));

done:

	/* Disconnect from the cancelled signals. Note that we have to do this with @write_mutex not held, as g_cancellable_disconnect() blocks

	 * until any outstanding cancellation callbacks return, and they will block on @write_mutex. */

	if (cancelled_signal != 0)

		g_cancellable_disconnect (cancellable, cancelled_signal);

	if (global_cancelled_signal != 0)

		g_cancellable_disconnect (priv->cancellable, global_cancelled_signal);

	g_assert (error != NULL || length_written > 0);

	if (error != NULL) {

		g_propagate_error (error_out, error);

	}

	return length_written;

}

static void

flush_cancelled_cb (GCancellable *cancellable, CancelledData *data)

{

	GDataUploadStreamPrivate *priv = data->upload_stream->priv;

	/* Signal the gdata_upload_stream_flush() function that it should stop blocking and cancel */

	g_mutex_lock (&(priv->write_mutex));

	*(data->cancelled) = TRUE;

	g_cond_signal (&(priv->write_cond));

	g_mutex_unlock (&(priv->write_mutex));

}

/* Block until ->network_bytes_outstanding reaches zero. Cancelling the cancellable passed to gdata_upload_stream_flush() breaks out of the wait(),

 * but doesn't stop the network thread from continuing to write the remaining bytes to the network.

 * The wrapper function, g_output_stream_flush(), calls g_output_stream_set_pending() before calling this function, and calls

 * g_output_stream_clear_pending() afterwards, so we don't need to worry about other operations happening concurrently. We also don't need to worry

 * about being called after the stream has been closed (though the network activity could finish before or during this function). */

static gboolean

gdata_upload_stream_flush (GOutputStream *stream, GCancellable *cancellable, GError **error)

{

	GDataUploadStreamPrivate *priv = GDATA_UPLOAD_STREAM (stream)->priv;

	gulong cancelled_signal = 0, global_cancelled_signal = 0;

	gboolean cancelled = FALSE; /* must only be touched with ->write_mutex held */

	gboolean success = TRUE;

	CancelledData data;

	/* Listen for cancellation events */

	data.upload_stream = GDATA_UPLOAD_STREAM (stream);

	data.cancelled = &cancelled;

	global_cancelled_signal = g_cancellable_connect (priv->cancellable, (GCallback) flush_cancelled_cb, &data, NULL);

	if (cancellable != NULL)

		cancelled_signal = g_cancellable_connect (cancellable, (GCallback) flush_cancelled_cb, &data, NULL);

	/* Create the thread if it hasn't been created already. This can happen if flush() is called immediately after creating the stream. */

	if (priv->network_thread == NULL) {

		create_network_thread (GDATA_UPLOAD_STREAM (stream), error);

		if (priv->network_thread == NULL) {

			success = FALSE;

			goto done;

		}

	}

	/* Start the flush operation proper */

	g_mutex_lock (&(priv->write_mutex));

	/* Wait for all outstanding bytes to be written to the network */

	while (priv->network_bytes_outstanding > 0 && cancelled == FALSE && priv->state != STATE_FINISHED) {

		g_cond_wait (&(priv->write_cond), &(priv->write_mutex));

	}

	/* Check for an error and return if necessary */

	if (cancelled == TRUE) {

		g_assert (g_cancellable_set_error_if_cancelled (cancellable, error) == TRUE ||

		          g_cancellable_set_error_if_cancelled (priv->cancellable, error) == TRUE);

		success = FALSE;

	} else if (priv->state == STATE_FINISHED && priv->network_bytes_outstanding > 0) {

		/* Resumable upload error. */

		g_set_error (error, G_IO_ERROR, G_IO_ERROR_FAILED, _("Error received from server after uploading a resumable upload chunk."));

		success = FALSE;

	}

	g_mutex_unlock (&(priv->write_mutex));