/*
 * libsyncml - A syncml protocol implementation
 * Copyright (C) 2005  Armin Bauer <armin.bauer@opensync.org>
 * Copyright (C) 2007-2009  Michael Bell <michael.bell@opensync.org>
 *
 * This library 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.
 *
 * This library 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 this library; if not, write to the Free Software
 * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307  USA
 *
 */

#include "sml_ds_server_internals.h"

#include <libsyncml/sml_support.h>
#include <libsyncml/sml_error_internals.h>

#include <libsyncml/sml_manager.h>
#include <libsyncml/sml_session.h>
#include <libsyncml/sml_command.h>
#include <libsyncml/sml_elements.h>

/*
static SmlWriteContext *_write_context_find(SmlDsSession *dsession, const gchar *uid, SmlChangeType type)
{
	smlTrace(TRACE_ENTRY, "%s(%p, %s, %i)", __func__, dsession, VA_STRING(uid), type);
	
	g_mutex_lock(dsession->pendingChangesLock);
	GList *c = NULL;
	for (c = dsession->pendingChanges; c; c = c->next) {
		SmlWriteContext *ctx = c->data;
		SmlLocation *loc = sml_data_sync_change_item_get_location(ctx->item);
		const char *uri = sml_location_get_full_uri(loc);
		if (!strcmp(uid, uri) &&
		    sml_data_sync_change_item_get_action(ctx->item) == type) {
			g_mutex_unlock(dsession->pendingChangesLock);
			smlTrace(TRACE_EXIT, "%s: %p", __func__, ctx);
			return ctx;
		}
	}
	g_mutex_unlock(dsession->pendingChangesLock);
	
	smlTrace(TRACE_EXIT_ERROR, "%s: Not found", __func__);
	return NULL;
}
*/

static void _write_context_free(SmlWriteContext *ctx)
{
	smlTrace(TRACE_ENTRY, "%s(%p)", __func__, ctx);

	if (ctx->status)
		smlStatusUnref(ctx->status);
	
	if (ctx->item)
		g_object_unref(ctx->item);
		
	smlSafeFree((gpointer *)&ctx);

	smlTrace(TRACE_EXIT, "%s", __func__);
}

static void _write_context_dispatch(SmlDsSession *dsession, SmlWriteContext *ctx)
{
	smlTrace(TRACE_ENTRY, "%s(%p, %p)", __func__, dsession, ctx);

	if (!ctx->status) {
		smlTrace(TRACE_EXIT, "%s: No status yet", __func__);
		return;
	}
	
	ctx->callback(dsession, ctx->status, ctx->userdata);
	
	g_mutex_lock(dsession->pendingChangesLock);
	dsession->pendingChanges = g_list_remove(dsession->pendingChanges, ctx);
	g_mutex_unlock(dsession->pendingChangesLock);

	_write_context_free(ctx);
		
	smlTrace(TRACE_EXIT, "%s", __func__);
}

static void _alert_reply(SmlSession *session, SmlStatus *status, void *userdata)
{
	smlTrace(TRACE_ENTRY, "%s(%p, %p, %p)", __func__, session, status, userdata);
	SmlDsSession *dsession = userdata;
	
	if (dsession->sentAlertCallback)
		dsession->sentAlertCallback(session, status, dsession->sentAlertCallbackUserdata);
	
	dsession->sentAlertCallback = NULL;
	dsession->sentAlertCallbackUserdata = NULL;
	
	smlTrace(TRACE_EXIT, "%s", __func__);
}

static void _sync_reply(SmlSession *session, SmlStatus *status, void *userdata)
{
	smlTrace(TRACE_ENTRY, "%s(%p, %p, %p)", __func__, session, status, userdata);
	SmlDsSession *dsession = userdata;
	
	if (dsession->sentSyncCallback)
		dsession->sentSyncCallback(session, status, dsession->sentSyncCallbackUserdata);
	
	dsession->sentSyncCallback = NULL;
	dsession->sentSyncCallbackUserdata = NULL;

	smlTrace(TRACE_EXIT, "%s", __func__);
}

static void _change_reply(SmlSession *session, SmlStatus *status, void *userdata)
{
	smlTrace(TRACE_ENTRY, "%s(%p, %p, %p)", __func__, session, status, userdata);
	smlAssert(session);
	smlAssert(status);
	SmlWriteContext *ctx = userdata;
	SmlDsSession *dsession = ctx->session;
	
	/* Please see smlChangeAssemble in sml_xml_parser.c for
	 * a more detailed description on the checked behaviour.
	 * If the received command is an Add or the remote peer
	 * is an OMA DS client then the remote peer must send a
	 * Source. Otherwise a Target must be sent by the remote
	 * peer. We only check the required behaviour and not
	 * the optional one.
	 */
	if (smlStatusGetType(status) == SML_COMMAND_TYPE_ADD ||
	    smlSessionGetType(session) == SML_SESSION_TYPE_CLIENT) {
		/* FIXME: How do we handle remote peers which are clients
		 * FIXME: and only support replace? The assembler will
		 * FIXME: send a replace command with a target instead of
		 * FIXME: an add command with a source.
		 */
		if (!smlStatusGetSourceRef(status)) {
			smlTrace(TRACE_EXIT_ERROR, "%s: Received add status or change reply from an OMA DS server without sourceRef", __func__);
			return;
		}
	} else {
		if (!smlStatusGetTargetRef(status)) {
			smlTrace(TRACE_EXIT_ERROR, "%s: Received delete or modify status or change reply from an OMA DS client without targetRef", __func__);
			return;
		}
	}
	
	ctx->status = status;
	smlStatusRef(status);
	_write_context_dispatch(dsession, ctx);
	
	smlTrace(TRACE_EXIT, "%s", __func__);
}

SmlDsServer*
smlDsServerNew (const gchar *type,
                SmlLocation *location,
                GError **error)
{
	smlTrace(TRACE_ENTRY, "%s(%s, %p, %p)", __func__, VA_STRING(type), location, error);
	smlAssert(location);
	CHECK_ERROR_REF
	
	SmlDsServer *server = smlTryMalloc0(sizeof(SmlDsServer), error);
	if (!server)
		goto error;

	server->location = location;
	g_object_ref(location);

	server->contenttype = g_strdup(type);
	server->servertype = SML_DS_SERVER;
	
	smlTrace(TRACE_EXIT, "%s: %p", __func__, server);
	return server;
error:
	smlTrace(TRACE_EXIT_ERROR, "%s: %s", __func__, (*error)->message);
	return NULL;
}

SmlDsServer*
smlDsClientNew (const gchar *type,
                SmlLocation *location,
                SmlLocation *target,
                GError **error)
{
	smlTrace(TRACE_ENTRY, "%s(%s, %p, %p, %p)", __func__, VA_STRING(type), location, target, error);
	smlAssert(location);
	smlAssert(target);
	CHECK_ERROR_REF
	
	SmlDsServer *server = smlTryMalloc0(sizeof(SmlDsServer), error);
	if (!server)
		goto error;

	server->location = location;
	g_object_ref(location);
	
	server->target = target;
	g_object_ref(target);

	server->contenttype = g_strdup(type);
	server->servertype = SML_DS_CLIENT;
	
	smlTrace(TRACE_EXIT, "%s: %p", __func__, server);
	return server;
error:
	smlTrace(TRACE_EXIT_ERROR, "%s: %s", __func__, (*error)->message);
	return NULL;
}

SmlDsServerType smlDsServerGetServerType(SmlDsServer *server)
{
	smlAssert(server);
	return server->servertype;
}

void smlDsServerFree(SmlDsServer *server)
{
	smlTrace(TRACE_ENTRY, "%s(%p)", __func__, server);
	smlAssert(server);
	
	if (server->location)
		g_object_unref(server->location);
		
	if (server->target)
		g_object_unref(server->target);
	
	if (server->contenttype)
		smlSafeCFree(&(server->contenttype));
	
	smlSafeFree((gpointer *)&server);
	
	smlTrace(TRACE_EXIT, "%s", __func__);	
}

/** @brief Registers a callback that will get called once a client connects
 * 
 * This function will get called once a client connects to our ds server (which means
 * that it sent a alert to our server). You can then use the smlDsServerRequestAlert() function
 * to get the alert
 * 
 * @param server The DS server
 * @param callback The callback that will receive the alert
 * @param userdata The userdata that will be passed to the alert
 */
void smlDsServerSetConnectCallback(SmlDsServer *server, SmlDsSessionConnectCb callback, void *userdata)
{
	smlTrace(TRACE_ENTRY, "%s(%p, %p, %p)", __func__, server, callback, userdata);
	smlAssert(server);
	
	server->connectCallback = callback;
	server->connectCallbackUserdata = userdata;
	
	smlTrace(TRACE_EXIT, "%s", __func__);	
}

void smlDsServerSetSanSessionCallback(SmlDsServer *server, SmlDsServerSanSessionCb callback, void *userdata)
{
	smlTrace(TRACE_ENTRY, "%s(%p, %p, %p)", __func__, server, callback, userdata);
	smlAssert(server);
	smlAssert(server->servertype == SML_DS_CLIENT);
	
	server->sanSessionCallback = callback;
	server->sanSessionCallbackUserdata = userdata;
	
	smlTrace(TRACE_EXIT, "%s", __func__);	
}

const char *smlDsServerGetLocation(SmlDsServer *server)
{
	smlAssert(server);
	if (server->location)
		return sml_location_get_uri(server->location);
	return NULL;
}

const char *smlDsServerGetContentType(SmlDsServer *server)
{
	smlAssert(server);
	return server->contenttype;
}

gboolean
smlDsServerAddSan (SmlDsServer *server,
                   SmlNotification *san,
                   GError **error)
{
	smlTrace(TRACE_ENTRY, "%s(%p, %p, %p)", __func__, server, san, error);
	smlAssert(server);
	smlAssert(san);
	CHECK_ERROR_REF
	
	if (!smlNotificationNewAlert(san, SML_ALERT_TWO_WAY_BY_SERVER, server->contenttype, sml_location_get_uri(server->location), error))
		goto error;
	
	smlTrace(TRACE_EXIT, "%s", __func__);
	return TRUE;
error:
	smlTrace(TRACE_EXIT_ERROR, "%s: %s", __func__, (*error)->message);
	return FALSE;
}

SmlDsSession *smlDsServerRecvAlert(SmlDsServer *server, SmlSession *session, SmlCommand *cmd)
{
	smlTrace(TRACE_ENTRY, "%s(%p, %p, %p)", __func__, server, session, cmd);
	GError *error = NULL;
	
	SmlDsSession *dsession = smlDsSessionNew(server, session, &error);
	if (!dsession)
		goto error;
	
	smlDsSessionRecvAlert(session, cmd, dsession);
	
	if (server->connectCallback)
		server->connectCallback(dsession, server->connectCallbackUserdata);
	
	smlDsSessionUnref(dsession);
	
	smlTrace(TRACE_EXIT, "%s", __func__);
	return dsession;
error:
	smlSessionDispatchEvent(session, SML_SESSION_EVENT_ERROR, NULL, NULL, NULL, error);
	smlTrace(TRACE_EXIT_ERROR, "%s: %s", __func__, error->message);
	g_error_free(error);
	return NULL;
}

SmlDsSession*
smlDsServerSendAlert (SmlDsServer *server,
                      SmlSession *session,
                      SmlAlertType type,
                      const gchar *last,
                      const gchar *next,
                      SmlStatusReplyCb callback,
                      void *userdata,
                      GError **error)
{
	smlTrace(TRACE_ENTRY, "%s(%p, %p, %i, %s, %s, %p)", __func__, server, session, type, VA_STRING(last), VA_STRING(next), error);
	smlAssert(server);
	smlAssert(session);
	CHECK_ERROR_REF
	
	SmlDsSession *dsession = smlDsSessionNew(server, session, error);
	if (!dsession)
		goto error;
	
	if (server->manager) {
		if (!smlManagerObjectRegister(server->manager, SML_COMMAND_TYPE_SYNC, session, server->location, NULL, NULL, smlDsSessionRecvSync, smlDsSessionRecvChange, dsession, error))
			goto error;
		
		if (!smlManagerObjectRegister(server->manager, SML_COMMAND_TYPE_MAP, session, server->location, NULL, NULL, smlDsSessionRecvMap, NULL, dsession, error))
			goto error;
		
		if (!smlManagerObjectRegister(server->manager, SML_COMMAND_TYPE_ALERT, session, server->location, NULL, NULL, smlDsSessionRecvAlert, NULL, dsession, error))
			goto error;
	}
	
	if (!smlDsSessionSendAlert(dsession, type, last, next, callback, userdata, error))
		goto error;
	
	smlTrace(TRACE_EXIT, "%s", __func__);
	return dsession;
error:
	if (dsession)
		smlSafeFree((gpointer *)&dsession);
	smlTrace(TRACE_EXIT_ERROR, "%s: %s", __func__, (*error)->message);
	return NULL;
}

SmlDsSession*
smlDsSessionNew (SmlDsServer *server,
                 SmlSession *session,
                 GError **error)
{
	smlTrace(TRACE_ENTRY, "%s(%p, %p, %p)", __func__, server, session, error);
	smlAssert(server);
	CHECK_ERROR_REF
	
	SmlDsSession *dsession = smlTryMalloc0(sizeof(SmlDsSession), error);
	if (!dsession)
		goto error;

	dsession->server = server;
	dsession->session = session;
	dsession->write_lock = g_mutex_new();
	dsession->lock = g_mutex_new();
	dsession->syncReply = SML_ERROR_UNKNOWN;
	dsession->refCount = 1;
	
	if (server->servertype == SML_DS_CLIENT) {
		dsession->target = server->target;
		g_object_ref(dsession->target);
	}
	
	dsession->location = server->location;
	g_object_ref(dsession->location);

	dsession->pendingChangesLock = g_mutex_new();
	
	smlTrace(TRACE_EXIT, "%s: %p", __func__, dsession);
	return dsession;
error:
	smlTrace(TRACE_EXIT_ERROR, "%s: %s", __func__, (*error)->message);
	return NULL;
}

SmlDsSession *smlDsSessionRef(SmlDsSession *dsession)
{
	smlTrace(TRACE_ENTRY, "%s(%p)", __func__, dsession);
	smlAssert(dsession);
	
	g_atomic_int_inc(&(dsession->refCount));
	
	smlTrace(TRACE_EXIT, "%s: New refcount: %i", __func__, dsession->refCount);
	return dsession;
}

void smlDsSessionUnref(SmlDsSession *dsession)
{
	smlTrace(TRACE_ENTRY, "%s(%p)", __func__, dsession);
	smlAssert(dsession);
	
	if (g_atomic_int_dec_and_test(&(dsession->refCount))) {
		smlTrace(TRACE_INTERNAL, "%s: Refcount == 0!", __func__);
		
		if (dsession->target)
			g_object_unref(dsession->target);
			
		if (dsession->location)
			g_object_unref(dsession->location);
		
		if (dsession->alertCommand)
			smlCommandUnref(dsession->alertCommand);
		
		while (dsession->recvSync) {
			SmlCommand *cmd = dsession->recvSync->data;
			smlCommandUnref(cmd);
			dsession->recvSync = g_list_delete_link(dsession->recvSync, dsession->recvSync);
		}
		
		while (dsession->recvChanges) {
			SmlCommand *cmd = dsession->recvChanges->data;
			smlCommandUnref(cmd);
			dsession->recvChanges = g_list_delete_link(dsession->recvChanges, dsession->recvChanges);
		}
		
		while (dsession->recvMap) {
			SmlCommand *cmd = dsession->recvMap->data;
			smlCommandUnref(cmd);
			dsession->recvMap = g_list_delete_link(dsession->recvMap, dsession->recvMap);
		}
		
		if (dsession->syncCommand)
			smlCommandUnref(dsession->syncCommand);
		
		if (!g_mutex_trylock(dsession->pendingChangesLock)) {
			smlTrace(TRACE_ERROR,
				"%s: somebody still uses this object", __func__);
			g_mutex_lock(dsession->pendingChangesLock);
		}
		while (dsession->pendingChanges) {
			SmlWriteContext *ctx = dsession->pendingChanges->data;
			_write_context_free(ctx);
			dsession->pendingChanges = g_list_delete_link(dsession->pendingChanges, dsession->pendingChanges);
		}
		g_mutex_unlock(dsession->pendingChangesLock);
		g_mutex_free(dsession->pendingChangesLock);
		
		while (dsession->mapItems) {
			SmlMapItem *item = dsession->mapItems->data;
			g_object_unref(item);
			dsession->mapItems = g_list_delete_link(dsession->mapItems, dsession->mapItems);
		}
		
		if (dsession->lock)
			g_mutex_free(dsession->lock);
		if (dsession->write_lock)
			g_mutex_free(dsession->write_lock);
		
		smlSafeFree((gpointer *)&dsession);
	}
	
	smlTrace(TRACE_EXIT, "%s", __func__);
}

void smlDsSessionDispatch(SmlDsSession *dsession)
{
	smlTrace(TRACE_ENTRY, "%s(%p)", __func__, dsession);
	GError *error = NULL;
	SmlStatus *reply = NULL;

	/* A dispatch function should never block on a
	 * locked resource. This is like a busy wait
	 * because the thread is not available for this
	 * time which can block the whole system if two
	 * dispatchers use the same thread.
	 */
	if (!g_mutex_trylock(dsession->lock))
	{
		smlTrace(TRACE_EXIT, "%s - DsSession is already locked", __func__);
		return;
	}
	smlTrace(TRACE_INTERNAL, "%s - locked DsSession successfully", __func__);

	if (dsession->alertCommand && dsession->recvAlertCallback) {
		smlTrace(TRACE_INTERNAL, "%s: Dispatching alert", __func__);
		
		SmlErrorType type = SML_NO_ERROR;
		if (!dsession->recvAlertCallback(dsession,
		                                 smlCommandGetAlertType(dsession->alertCommand),
		                                 smlAnchorGetLast(smlCommandGetAnchor(dsession->alertCommand)),
		                                 smlAnchorGetNext(smlCommandGetAnchor(dsession->alertCommand)),
		                                 dsession->recvAlertCallbackUserdata))
			type = SML_ERROR_REQUIRE_REFRESH;
		dsession->recvAlertCallback = NULL;

		reply = smlCommandNewReply(dsession->alertCommand, type, &error);
		if (!reply)
			goto error;
		
		if (!smlSessionSendReply(dsession->session, reply, &error))
			goto error;
		
		
		smlStatusUnref(reply);
		
		smlCommandUnref(dsession->alertCommand);
		dsession->alertCommand = NULL;

		/* unlock the final event at the manager
		 * 
		 * If this is called from a test case then the datastore
		 * is sometimes not registered at a SmlManager.
		 */
		if (dsession->server->manager && dsession->finalLock)
		{
			dsession->finalLock = FALSE;
			smlManagerSessionFinalLockUnref(dsession->server->manager, dsession->session);
		}

	} else if (dsession->recvSync && dsession->recvSyncCallback) {
		smlTrace(TRACE_INTERNAL, "%s: Dispatching sync", __func__);
		
		dsession->recvSyncCallback(dsession, smlCommandGetNumberOfChanges(((SmlCommand *)(dsession->recvSync->data))), dsession->recvSyncCallbackUserdata);
		dsession->recvSyncCallback = NULL;
		
		while (dsession->recvSync) {
			SmlCommand *cmd = dsession->recvSync->data;
			
			smlTrace(TRACE_INTERNAL, "%s: answering sync command with cmdRef %i and msgRef %i",
			         __func__,
			         smlCommandGetID(cmd),
			         smlCommandGetMessageID(cmd));
			reply = smlCommandNewReply(cmd, SML_NO_ERROR, &error);
			if (!reply)
				goto error;
			
			if (!smlSessionSendReply(dsession->session, reply, &error))
				goto error;
				
			smlStatusUnref(reply);
			
			smlCommandUnref(cmd);
			dsession->recvSync = g_list_delete_link(dsession->recvSync, dsession->recvSync);
		}
		
		dsession->syncReply = SML_NO_ERROR;
		
		if (!dsession->recvChanges) {
			/* There are no changes. */

			/* unlock the final event at the manager
			 * 
			 * If this is called from a test case then the datastore
			 * is sometimes not registered at a SmlManager.
			 */
			if (dsession->server->manager && dsession->finalLock)
			{
				dsession->finalLock = FALSE;
				smlManagerSessionFinalLockUnref(dsession->server->manager, dsession->session);
			}
			smlTrace(TRACE_INTERNAL, "%s: final handling done", __func__);
		}
	} else if (dsession->recvChanges && dsession->recvChangeCallback) {
		smlTrace(TRACE_INTERNAL, "%s: Dispatching changes", __func__);
		
		while (dsession->recvChanges) {

			SmlCommand *cmd = dsession->recvChanges->data;
			
			if (cmd) {
				if (!smlCommandGetNumChanges(cmd))
				{
					g_set_error(&error, SML_ERROR, SML_ERROR_GENERIC,
					            "No items found in command.");
					goto error;
				}

				/* All items handled at once because libsyncml
				 * only supports one status code for all items.
				 * So one failed item means that the complete
				 * command is aborted.
				 */

				gsize i;
				for (i=0; i < smlCommandGetNumChanges(cmd); i++)
				{
					SmlDataSyncChangeItem *item = smlCommandGetNthChange(cmd, i);
				        if (!item)
					{
						g_set_error(&error, SML_ERROR, SML_ERROR_GENERIC,
							"Item %i of the command's item list is NULL.", i);
						goto error;
					}
					if (!dsession->recvChangeCallback(dsession,
					                                  item, 
					                                  dsession->recvChangeCallbackUserdata,
					                                  &error))
						goto error;
				}

				if (smlCommandGetChangeType(cmd) == SML_CHANGE_ADD)
					reply = smlCommandNewReply(cmd, SML_ITEM_ADDED, &error);
				else
					reply = smlCommandNewReply(cmd, SML_NO_ERROR, &error);
				
				if (!reply)
					goto error;
				
				if (!smlSessionSendReply(dsession->session, reply, &error))
					goto error;
					
				smlStatusUnref(reply);
			
				smlCommandUnref(cmd);
			}

			dsession->recvChanges = g_list_delete_link(dsession->recvChanges, dsession->recvChanges);
		}

		/* unlock the final event at the manager                                                                                                                    
		 *                                                                                                                                                          
		 * If this is called from a test case then the datastore                                                                                                    
		 * is sometimes not registered at a SmlManager.                                                                                                             
		 */                                                                                                                                                         
		if (dsession->server->manager)                                                                                                                              
		{                                                                                                                                                           
			dsession->finalLock = FALSE;                                                                                                                        
			smlManagerSessionFinalLockUnref(dsession->server->manager, dsession->session);                                                                      
		}                                                                                                                                                           
		else                                                                                                                                                        
			smlTrace(TRACE_INTERNAL, "%s - no manager so should be a test", __func__);
	} else if (dsession->recvMap && dsession->recvMappingCallback) {
		while (dsession->recvMap) {
			SmlCommand *cmd = dsession->recvMap->data;

			smlTrace(TRACE_INTERNAL, "%s: answering map command with cmdRef %i and msgRef %i",
			         __func__,
		        	 smlCommandGetID(cmd),
				 smlCommandGetMessageID(cmd));

			/* handle the mapping */
			gsize i;
			for (i = 0; i < smlCommandGetNumMappings(cmd); i++) {
				SmlMapItem *item = smlCommandGetNthMapping(cmd, i);

				dsession->recvMappingCallback(
						dsession,
						item,
						dsession->recvMappingCallbackUserdata);
			}

			/* send the reply */
			reply = smlCommandNewReply(cmd, SML_NO_ERROR, &error);
			if (!reply)
				goto error;
			
			if (!smlSessionSendReply(dsession->session, reply, &error))
				goto error;
				
			smlStatusUnref(reply);
			
			/* cleanup the command */
			smlCommandUnref(cmd);
			dsession->recvMap = g_list_delete_link(dsession->recvMap, dsession->recvMap);
		}
	} else {
		
		smlTrace(TRACE_INTERNAL, "%s: no actions/callbacks to be done", __func__); 
	}

	smlTrace(TRACE_EXIT, "%s()", __func__);
	g_mutex_unlock(dsession->lock);
	
	return;
error:
	if (reply)
		smlStatusUnref(reply);
	smlSessionDispatchEvent(dsession->session, SML_SESSION_EVENT_ERROR, NULL, NULL, NULL, error);
	smlTrace(TRACE_EXIT_ERROR, "%s: Unable to dispatch: %s", __func__, error->message);
	g_error_free(error);
	g_mutex_unlock(dsession->lock);
}

gboolean smlDsSessionCheck(SmlDsSession *dsession)
{
	if ((dsession->alertCommand && dsession->recvAlertCallback) || \
		(dsession->recvSync && dsession->recvSyncCallback) || \
		(dsession->recvChanges && dsession->recvChangeCallback) || \
		(dsession->recvMap && dsession->recvMappingCallback))
		return TRUE;
	return FALSE;
}

void smlDsSessionRecvAlert(SmlSession *session, SmlCommand *cmd, void *userdata)
{
	smlTrace(TRACE_ENTRY, "%s(%p, %p, %p)", __func__, session, cmd, userdata);
	SmlDsSession *dsession = userdata;
	GError *error = NULL;
	smlAssert(dsession->location);
	
	g_mutex_lock(dsession->lock);
	
	if (!smlCommandGetTarget(cmd) || !smlCommandGetSource(cmd)) {
		SmlStatus *reply = smlCommandNewReply(cmd, SML_ERROR_BAD_REQUEST, &error);
		if (!reply)
			goto error;
		
		if (!smlSessionSendReply(session, reply, &error)) {
			smlStatusUnref(reply);
			goto error;
		}
		
		smlStatusUnref(reply);
			
		smlTrace(TRACE_EXIT, "%s: Alert had no target or source", __func__);
		return;
	}
	
	if (!sml_location_is_equal(dsession->location, smlCommandGetTarget(cmd))) {
		SmlStatus *reply = smlCommandNewReply(cmd, SML_ERROR_NOT_FOUND, &error);
		if (!reply)
			goto error;
		
		if (!smlSessionSendReply(session, reply, &error)) {
			smlStatusUnref(reply);
			goto error;
		}
		
		smlStatusUnref(reply);
		
		smlTrace(TRACE_EXIT, "%s: Alert does not match our location", __func__);
		return;
	}
	
	smlCommandRef(cmd);
	
	if (!dsession->target) {
		dsession->target = smlCommandGetSource(cmd);
		g_object_ref(dsession->target);
	}
	
	dsession->alertCommand = cmd;

	/* Lock the final event at the manager.
	 * 
	 * If this is called from a test case then the datastore
	 * is sometimes not registered at a SmlManager.
	 */
	if (dsession->server->manager)
	{
		smlManagerSessionFinalLockRef(dsession->server->manager, session);
		dsession->finalLock = TRUE;
	}
	else
		smlTrace(TRACE_INTERNAL, "%s - no manager so should be a test", __func__);
	
	g_mutex_unlock(dsession->lock);
	
	smlTrace(TRACE_EXIT, "%s", __func__);
	return;

error:
	g_mutex_unlock(dsession->lock);
	smlSessionDispatchEvent(session, SML_SESSION_EVENT_ERROR, NULL, NULL, NULL, error);
	smlTrace(TRACE_EXIT_ERROR, "%s: %s", __func__, error->message);
	g_error_free(error);
}

void smlDsSessionRecvSync(SmlSession *session, SmlCommand *cmd, void *userdata)
{
	smlTrace(TRACE_ENTRY, "%s(%p, %p, %p)", __func__, session, cmd, userdata);
	SmlDsSession *dsession = userdata;
	GError *error = NULL;

	/* Create a write lock on the complete DsSession object to avoid race
	 * conditions. If the write lock is already present then this is the
	 * end of the sync block and the write lock must only be removed.
	 */
	if (g_mutex_trylock(dsession->write_lock))
	{
		/* write lock succeeded so lock the complete DsSession */
		g_mutex_lock(dsession->lock);
		smlTrace(TRACE_INTERNAL, "%s - write lock enabled", __func__);
		/* Lock the final event at the manager.
		 * 
		 * If this is called from a test case then the datastore
		 * is sometimes not registered at a SmlManager.
		 */
		if (dsession->server->manager)
		{
			smlManagerSessionFinalLockRef(dsession->server->manager, session);
			dsession->finalLock = TRUE;
			dsession->emptySync = TRUE;
		}
		else
			smlTrace(TRACE_INTERNAL, "%s - no manager so should be a test", __func__);
	} else {
		/* detected an old write lock so this is the closing sync call */
		if (dsession->finalLock && dsession->emptySync)
		{
			smlTrace(TRACE_INTERNAL, "%s - empty sync of moreData ?", __func__);
			dsession->finalLock = FALSE;
			smlManagerSessionFinalLockUnref(dsession->server->manager, session);
		}
		g_mutex_unlock(dsession->lock);
		g_mutex_unlock(dsession->write_lock);
		smlTrace(TRACE_EXIT, "%s - removed write lock", __func__);
		return;
	}

	if (dsession->syncReply == SML_ERROR_UNKNOWN) {
		smlTrace(TRACE_INTERNAL, "%s: Storing sync command with cmdRef %i and msgRef %i",
		          __func__,
		          smlCommandGetID(cmd),
		          smlCommandGetMessageID(cmd));
		smlCommandRef(cmd);
		dsession->recvSync = g_list_append(dsession->recvSync, cmd);
	} else {
		smlTrace(TRACE_INTERNAL, "%s: Using stored sync reply on cmd with cmdRef %i and msgRef %i",
		          __func__,
		          smlCommandGetID(cmd),
		          smlCommandGetMessageID(cmd));
		SmlStatus *reply = smlCommandNewReply(cmd, dsession->syncReply, &error);
		if (!reply)
			goto error;
		
		if (!smlSessionSendReply(dsession->session, reply, &error))
			goto error;
		
		smlStatusUnref(reply);
	}

	smlTrace(TRACE_EXIT, "%s", __func__);
	return;
	
error:
	g_mutex_unlock(dsession->lock);
	g_mutex_unlock(dsession->write_lock);
	smlSessionDispatchEvent(session, SML_SESSION_EVENT_ERROR, NULL, NULL, NULL, error);
	smlTrace(TRACE_EXIT_ERROR, "%s: %s", __func__, error->message);
	g_error_free(error);
}

void smlDsSessionRecvChange(SmlSession *session, SmlCommand *cmd, void *userdata)
{
	smlTrace(TRACE_ENTRY, "%s(%p, %p, %p)", __func__, session, cmd, userdata);
	SmlDsSession *dsession = userdata;

	/* It is not necessary to lock the DsSesion object
	 * because RecvSync already enabled a global write_lock.
	 */
	dsession->recvChanges = g_list_append(dsession->recvChanges, cmd);
	smlCommandRef(cmd);

	/* This is important for fast and consistent final signalling. */
	dsession->emptySync = FALSE;

	smlTrace(TRACE_EXIT, "%s", __func__);
}

/* FIXME: this is synchronous behaviour !? */
void smlDsSessionRecvMap(SmlSession *session, SmlCommand *cmd, void *userdata)
{
	smlTrace(TRACE_ENTRY, "%s(%p, %p, %p)", __func__, session, cmd, userdata);
	SmlDsSession *dsession = userdata;
	
	g_mutex_lock(dsession->lock);
	
	dsession->recvMap = g_list_append(dsession->recvMap, cmd);
	smlCommandRef(cmd);
	
	g_mutex_unlock(dsession->lock);
	
	smlTrace(TRACE_EXIT, "%s", __func__);
	return;
}

/** @brief Gets a already received alert
 * 
 * This function will get a already received alert or register a callback that will be
 * called once the alert is received. If the alert already was waiting the callback is 
 * called immediatly.
 * 
 * @param server The DS server
 * @param callback The callback that will receive the alert
 * @param userdata The userdata that will be passed to the alert
 * @param error A pointer to a error struct
 * @returns TRUE if the call was successful, FALSE otherwise
 */
void smlDsSessionGetAlert(SmlDsSession *dsession, SmlDsSessionAlertCb callback, void *userdata)
{
	smlTrace(TRACE_ENTRY, "%s(%p, %p, %p)", __func__, dsession, callback, userdata);
	smlAssert(dsession);
	smlAssert(callback);
	
	dsession->recvAlertCallback = callback;
	dsession->recvAlertCallbackUserdata = userdata;
	
	smlDsSessionDispatch(dsession);
	
	smlTrace(TRACE_EXIT, "%s", __func__);
}

/** @brief Sends the alert to the remote side
 * 
 * This function will get a already received alert or register a callback that will be
 * called once the alert is received. If the alert already was waiting the callback is 
 * called immediatly.
 * 
 * @param server The DS server
 * @param callback The callback that will receive the alert
 * @param userdata The userdata that will be passed to the alert
 * @param error A pointer to a error struct
 * @returns TRUE if the call was successful, FALSE otherwise
 */
gboolean
smlDsSessionSendAlert (SmlDsSession *dsession,
                       SmlAlertType type,
                       const gchar *last,
                       const gchar *next,
                       SmlStatusReplyCb callback,
                       void *userdata,
                       GError **error)
{
	smlTrace(TRACE_ENTRY, "%s(%p, %i, %s, %s, %p)", __func__, dsession, type, VA_STRING(last), VA_STRING(next), error);
	smlAssert(dsession);
	CHECK_ERROR_REF
	
	SmlCommand *alert = smlCommandNewAlert(type, dsession->target, dsession->location, next, last, NULL, error);
	if (!alert)
		goto error;
	
	dsession->sentAlertCallback = callback;
	dsession->sentAlertCallbackUserdata = userdata;
	
	if (!smlSessionSendCommand(dsession->session, alert, NULL, _alert_reply, dsession, error))
		goto error;
	
	smlCommandUnref(alert);
	
	smlTrace(TRACE_EXIT, "%s", __func__);
	return TRUE;
error:
	smlTrace(TRACE_EXIT_ERROR, "%s: %s", __func__, (*error)->message);
	return FALSE;
}

void smlDsSessionGetMapping(
		SmlDsSession *dsession,
		SmlDsSessionMapCb mapCallback,
		void *userdata)
{
	smlTrace(TRACE_ENTRY, "%s", __func__);
	dsession->recvMappingCallback = mapCallback;
	dsession->recvMappingCallbackUserdata = userdata;
	smlTrace(TRACE_EXIT, "%s", __func__);
}

/** @brief Gets a already received sync command
 * 
 * This function will get a already received sync command or register a callback that will be
 * called once the sync is received. If the sync already was waiting the callback is 
 * called immediatly. The read callback is called with every subcommand of the sync.
 * 
 * @param server The DS server
 * @param chgCallback The callback that will receive all the changes (subcommands of sync)
 * @param syncCallback The callback that will receive the sync command
 * @param userdata The userdata that will be passed to the sync and change callbacks
 * @param error A pointer to a error struct
 * @returns TRUE if the call was successful, FALSE otherwise
 */
void smlDsSessionGetChanges(SmlDsSession *dsession, SmlDsSessionChangeCb chgCallback, void *userdata)
{
	smlTrace(TRACE_ENTRY, "%s(%p, %p, %p)", __func__, dsession, chgCallback, userdata);
	smlAssert(dsession);
	smlAssert(chgCallback);
	
	dsession->recvChangeCallback = chgCallback;
	dsession->recvChangeCallbackUserdata = userdata;
	
	smlDsSessionDispatch(dsession);
	
	smlTrace(TRACE_EXIT, "%s", __func__);
}

void smlDsSessionGetSync(SmlDsSession *dsession, SmlDsSessionSyncCb syncCallback, void *userdata)
{
	smlTrace(TRACE_ENTRY, "%s(%p, %p, %p)", __func__, dsession, syncCallback, userdata);
	smlAssert(dsession);
	smlAssert(syncCallback);
	
	dsession->recvSyncCallback = syncCallback;
	dsession->recvSyncCallbackUserdata = userdata;
	
	smlDsSessionDispatch(dsession);
	
	smlTrace(TRACE_EXIT, "%s", __func__);
}

/** @brief Start the sync command to send to the other side
 * 
 * This function will start the sync command with which the changes will be sent to the other side.
 * After this command you can start to queue the changes. After you are done queueing changes, you
 * have to end the sync command with smlDsServerCloseSync().
 * 
 * @param server The DS server
 * @param callback The callback that will the answer to the sync command
 * @param userdata The userdata that will be passed to the alert
 * @param error A pointer to a error struct
 * @returns TRUE if the call was successful, FALSE otherwise
 */
gboolean
smlDsSessionSendSync (SmlDsSession *dsession,
                      gsize num_changes,
                      SmlStatusReplyCb callback,
                      void *userdata,
                      GError **error)
{
	smlTrace(TRACE_ENTRY, "%s(%p, %i, %p, %p, %p)", __func__, dsession, num_changes, callback, userdata, error);
	smlAssert(dsession);
	CHECK_ERROR_REF
	
	if (dsession->syncCommand) {
		g_set_error(error, SML_ERROR, SML_ERROR_GENERIC,
		            "There already was a sync command started");
		goto error;
	}
	
	dsession->sentSyncCallback = callback;
	dsession->sentSyncCallbackUserdata = userdata;
	
	dsession->syncCommand = smlCommandNewSync(dsession->target, dsession->location, num_changes, error);
	if (!dsession->syncCommand)
		goto error;
	
	if (!smlSessionStartCommand(dsession->session, dsession->syncCommand, NULL, _sync_reply, dsession, error))
		goto error;
	
	smlTrace(TRACE_EXIT, "%s", __func__);
	return TRUE;
error:
	smlTrace(TRACE_EXIT_ERROR, "%s: %s", __func__, (*error)->message);
	return FALSE;
}

//Send a change to the remote side
gboolean
smlDsSessionQueueChange (SmlDsSession *dsession,
                         SmlDataSyncChangeItem *item,
                         SmlDsSessionChangeStatusCb callback,
                         void *userdata,
                         GError **error)
{
	smlTrace(TRACE_ENTRY, "%s(%p, %p, %p, %p, %p)", __func__, dsession, item, callback, userdata, error);
	smlAssert(dsession);
	CHECK_ERROR_REF
	
	SmlCommand *cmd = NULL;
	SmlWriteContext *ctx = NULL;

	if (!dsession->syncCommand) {
		g_set_error(error, SML_ERROR, SML_ERROR_GENERIC,
		            "You have to start a sync command first");
		goto error;
	}
		
	cmd = smlCommandNewChange(item, error);
	if (!cmd)
		goto error;
	
	ctx = smlTryMalloc0(sizeof(SmlWriteContext), error);
	if (!ctx)
		goto error;
	
	ctx->callback = callback;
	ctx->userdata = userdata;
	ctx->item = item;
	g_object_ref(item);
	ctx->session = dsession;
	
	g_mutex_lock(dsession->pendingChangesLock);
	dsession->pendingChanges = g_list_append(dsession->pendingChanges, ctx);
	g_mutex_unlock(dsession->pendingChangesLock);
	
	if (!smlSessionSendCommand(dsession->session, cmd, dsession->syncCommand, _change_reply, ctx, error))
		goto error;
	ctx = NULL;
	
	smlCommandUnref(cmd);
	
	smlTrace(TRACE_EXIT, "%s", __func__);
	return TRUE;
error:
	if (ctx) {
		g_object_unref(ctx->item);
		smlSafeFree((gpointer *)&ctx);
	}
	if (cmd)
		smlCommandUnref(cmd);
	smlTrace(TRACE_EXIT_ERROR, "%s: %s", __func__, (*error)->message);
	return FALSE;
}

/** @brief Closes the sync command
 * 
 * @param server The DS server
 * @param error A pointer to a error struct
 * @returns TRUE if the call was successful, FALSE otherwise
 */
gboolean
smlDsSessionCloseSync (SmlDsSession *dsession,
                       GError **error)
{
	smlTrace(TRACE_ENTRY, "%s(%p, %p)", __func__, dsession, error);
	smlAssert(dsession);
	CHECK_ERROR_REF
	
	if (!dsession->syncCommand) {
		g_set_error(error, SML_ERROR, SML_ERROR_GENERIC,
		            "There already was a sync command started");
		goto error;
	}
	
	if (!smlSessionEndCommand(dsession->session, NULL, error))
		goto error;
	
	smlCommandUnref(dsession->syncCommand);
	dsession->syncCommand = NULL;
	
	smlTrace(TRACE_EXIT, "%s", __func__);
	return TRUE;
error:
	smlTrace(TRACE_EXIT_ERROR, "%s: %s", __func__, (*error)->message);
	return FALSE;
}

gboolean
smlDsSessionQueueMap (SmlDsSession *dsession,
                      SmlMapItem *item,
                      GError **error)
{
	smlTrace(TRACE_ENTRY, "%s(%p, %p, %p)", __func__, dsession, item, error);
	smlAssert(dsession);
	CHECK_ERROR_REF
	
	dsession->mapItems = g_list_append(dsession->mapItems, item);
	g_object_ref(item);
	
	smlTrace(TRACE_EXIT, "%s", __func__);
	return TRUE;
}

/** @brief Closes the map command
 * 
 * This will tell the ds server that you are not going to queue
 * any more map commands.
 * 
 * @param server The DS server
 * @param error A pointer to a error struct
 * @returns TRUE if the call was successful, FALSE otherwise
 */
gboolean
smlDsSessionCloseMap (SmlDsSession *dsession,
                      SmlStatusReplyCb callback,
                      void *userdata,
                      GError **error)
{
	smlTrace(TRACE_ENTRY, "%s(%p, %p, %p, %p)", __func__, dsession, callback, userdata, error);
	smlAssert(dsession);
	CHECK_ERROR_REF
	
	if (!dsession->mapItems) {
		smlTrace(TRACE_EXIT, "%s: No mapitems", __func__);
		return TRUE;
	}
	
	SmlCommand *cmd = smlCommandNewMap(dsession->server->target, dsession->server->location, error);
	if (!cmd)
		goto error;
	
	while (dsession->mapItems) {
		SmlMapItem *item = dsession->mapItems->data;
		if (!smlCommandAddMapItem(cmd, item, error))
			goto error;
		dsession->mapItems = g_list_remove(dsession->mapItems, item);
		g_object_unref(item);
	}
	
	if (!smlSessionSendCommand(dsession->session, cmd, NULL, callback, userdata, error))
		goto error;
	
	smlCommandUnref(cmd);
	
	smlTrace(TRACE_EXIT, "%s", __func__);
	return TRUE;
error:
	if (cmd)
		smlCommandUnref(cmd);
	smlTrace(TRACE_EXIT_ERROR, "%s: %s", __func__, (*error)->message);
	return FALSE;
}

const gchar *smlDsSessionGetLocation(SmlDsSession *dsession)
{
	smlAssert(dsession);
	return smlDsServerGetLocation(dsession->server);
}

SmlDsServer *smlDsSessionGetServer(SmlDsSession *dsession)
{
	smlAssert(dsession);
	return dsession->server;
}

SmlSession *smlDsSessionGetSession(SmlDsSession *dsession)
{
	smlAssert(dsession);
	return dsession->session;
}

const gchar *smlDsSessionGetContentType(SmlDsSession *dsession)
{
	smlAssert(dsession);
	return smlDsServerGetContentType(dsession->server);
}

static void _recv_manager_alert(SmlSession *session, SmlCommand *cmd, void *userdata)
{
	smlTrace(TRACE_ENTRY, "%s(%p, %p, %p)", __func__, session, cmd, userdata);
	SmlDsServer *server = userdata;
	GError *error = NULL;
	
	/* The error is already signaled by smlDsServerRecvAlert. */
	SmlDsSession *dsession = smlDsServerRecvAlert(server, session, cmd);
	if (!dsession)
		goto error;
	
	if (!smlManagerObjectRegister(server->manager, SML_COMMAND_TYPE_SYNC, session, server->location, NULL, NULL, smlDsSessionRecvSync, smlDsSessionRecvChange, dsession, &error))
		goto error;
	
	if (!smlManagerObjectRegister(server->manager, SML_COMMAND_TYPE_MAP, session, server->location, NULL, NULL, smlDsSessionRecvMap, NULL, dsession, &error))
		goto error;
	
	smlTrace(TRACE_EXIT, "%s", __func__);
	return;

error:
	if (dsession)
		smlSafeFree((gpointer *)&dsession);
	smlTrace(TRACE_EXIT_ERROR, "%s: %s", __func__, error?error->message:"NULL");
	if (error) {
		smlSessionDispatchEvent(session, SML_SESSION_EVENT_ERROR, NULL, NULL, NULL, error);
		g_error_free(error);
	}
}

static void _recv_manager_san(SmlSession *session, SmlCommand *cmd, void *userdata)
{
	smlTrace(TRACE_ENTRY, "%s(%p, %p, %p)", __func__, session, cmd, userdata);
	SmlDsServer *server = userdata;
	GError *error = NULL;

	/* ensure that the session is in client mode */
	smlSessionSetType(session, SML_SESSION_TYPE_CLIENT);

	/* fix target */
	g_object_unref(server->target);
	server->target = smlCommandGetSource(cmd);
	g_object_ref(server->target);

	/* handle the SAN */
	if (server->sanSessionCallback) {
		SmlErrorType type = server->sanSessionCallback(
						server,
						session,
						smlCommandGetAlertType(cmd),
						server->sanSessionCallbackUserdata);
		
		SmlStatus *reply = smlCommandNewReply(cmd, type, &error);
		if (!reply)
			goto error;
		
		if (!smlSessionSendReply(session, reply, &error))
			goto error;
			
		smlStatusUnref(reply);
	} else {
		smlTrace(TRACE_INTERNAL, "%s: SAN ignored", __func__);
		
		SmlStatus *reply = smlCommandNewReply(cmd, SML_ERROR_NOT_IMPLEMENTED, &error);
		if (!reply)
			goto error;
		
		if (!smlSessionSendReply(session, reply, &error))
			goto error;
			
		smlStatusUnref(reply);
	}
	
	smlTrace(TRACE_EXIT, "%s", __func__);
	return;
	
error:
	smlTrace(TRACE_EXIT_ERROR, "%s: %s", __func__, error->message);
	smlSessionDispatchEvent(session, SML_SESSION_EVENT_ERROR, NULL, NULL, NULL, error);
	g_error_free(error);
}

gboolean
smlDsServerRegister (SmlDsServer *server,
                     SmlManager *manager,
                     GError **error)
{
	smlTrace(TRACE_ENTRY, "%s(%p, %p, %p)", __func__, server, manager, error);
	smlAssert(server);
	smlAssert(manager);
	CHECK_ERROR_REF
	
	if (!smlManagerObjectRegister(manager, SML_COMMAND_TYPE_ALERT, NULL, server->location, NULL, NULL, _recv_manager_alert, NULL, server, error))
		goto error;
	
	if (!smlManagerObjectRegister(manager, SML_COMMAND_TYPE_ALERT, NULL, NULL, NULL,server->contenttype, _recv_manager_san, NULL, server, error))
		goto error;
	
	server->manager = manager;
	
	smlTrace(TRACE_EXIT, "%s", __func__);
	return TRUE;
error:
	smlTrace(TRACE_EXIT_ERROR, "%s: %s", __func__, (*error)->message);
	return FALSE;
}

SmlLocation *smlDsSessionGetTarget(SmlDsSession *dsession)
{
	smlAssert(dsession);
	return dsession->target;
}

