/*
 * libsyncml - A syncml protocol implementation
 * Copyright (C) 2005  Armin Bauer <armin.bauer@opensync.org>
 * Copyright (C) 2008-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 "tests/support.h"

#include <libsyncml/objects/sml_ds_server.h>

#include <libsyncml/sml_command_internals.h>
#include <libsyncml/sml_transport_internals.h>
#include <libsyncml/objects/sml_ds_server_internals.h>
#include <libsyncml/sml_session_internals.h>
#include <libsyncml/sml_elements_internals.h>
#include <libsyncml/data_sync_api/sml_location.h>
#include <libsyncml/data_sync_api/sml_data_sync_defines.h>

#include <string.h>
#include <unistd.h>

typedef struct SmlSessionThread {
	GMainContext *context;
	GSourceFuncs *functions;
	GSourceFuncs *data_functions;
	GSource *data_source;
	GSource *session_source;
	SmlThread *thread;
	SmlSession *session;
	SmlTransportData **data;
} SmlSessionThread;

typedef struct _server_corr {
	SmlDsSession *dsession;
	SmlDsServer *server;
	SmlSessionThread *thread;
} _server_corr;

gint data_received = 0;
int init_alerts_received = 0;
int alerts_received = 0;
int syncs_received = 0;
int sync_ends_received = 0;
int changes_received = 0;
int replies_received = 0;
int adds_received = 0;
int add_replies_received = 0;
int deletes_received = 0;
int delete_replies_received = 0;
int modifies_received = 0;
int modify_replies_received = 0;
int mappings_received = 0;
int map_replies_received = 0;
int session_ends = 0;

SmlTransportData *to_session = NULL;
SmlTransportData *to_session2 = NULL;

unsigned int defaultMaxMsgSize = 10240;
unsigned int defaultMaxObjSize = 1024000;

//SmlDsSession *serversession = NULL;

gboolean _session_prepare(GSource *source, gint *timeout_)
{
	smlTrace(TRACE_INTERNAL, "%s(%p, %p)", __func__, source, timeout_);
	*timeout_ = 1;
	return FALSE;
}

gboolean _session_check(GSource *source)
{
	SmlSession *session = *((SmlSession **)(source + 1));
	return smlSessionCheck(session);
}

gboolean _session_dispatch(GSource *source, GSourceFunc callback, gpointer user_data)
{
	smlTrace(TRACE_INTERNAL, "%s(%p, %p)", __func__, source, callback);
	SmlSession *session = user_data;
	
	while (smlSessionCheck(session))
		smlSessionDispatch(session);
	
	return TRUE;
}
gboolean _data_prepare(GSource *source, gint *timeout_)
{
	smlTrace(TRACE_INTERNAL, "%s(%p, %p)", __func__, source, timeout_);
	*timeout_ = 1;
	return FALSE;
}

gboolean _data_check(GSource *source)
{
	SmlSessionThread *thread = *((SmlSessionThread **)(source + 1));
	smlAssert(thread != NULL);
	smlAssert(thread->data != NULL);
	
	if (*(thread->data))
		return TRUE;
	
	return FALSE;
}

gboolean _data_dispatch(GSource *source, GSourceFunc callback, gpointer user_data)
{
	smlTrace(TRACE_INTERNAL, "%s(%p, %p)", __func__, source, callback);
	SmlSessionThread *check = user_data;
	GError *error = NULL;
	smlAssert(check != NULL);
	smlAssert(check->data != NULL);
	
	if (*(check->data)) {
		
		SmlTransportData *data = *(check->data);
		*(check->data) = NULL;
		smlTrace(TRACE_ENTRY, "%s(data: %p, real %p, size %i)", __func__, data, data->data, data->size);
		
		SmlParser *parser = smlParserNew(data->type, 0, &error);
		smlAssert(parser != NULL);
		smlAssert(smlParserStart(parser, data->data, data->size, &error));
		
		SmlHeader *header = NULL;
		SmlCred *cred = NULL;
		smlAssert(smlParserGetHeader(parser, &header, &cred, &error));
		
		smlAssert(header != NULL);
		
		smlAssert(smlSessionReceiveHeader(check->session, header, &error));
		
		smlHeaderFree(header);
		
		SmlStatus *reply = smlStatusNew(SML_NO_ERROR, 0, check->session->lastReceivedMessageID, check->session->source, check->session->target, SML_COMMAND_TYPE_HEADER, &error);
		smlAssert(smlSessionSendReply(check->session, reply, &error));
		smlAssert(error == NULL);
		check->session->established = TRUE;
		
		smlStatusUnref(reply);
		
		smlAssert(smlSessionReceiveBody(check->session, parser, &error));
		
		smlParserFree(parser);
		smlTransportDataDeref(data);
		
		smlTrace(TRACE_EXIT, "%s", __func__);
	}
	return TRUE;
}

SmlSessionThread *smlSessionRunAsync(SmlSession *session, SmlTransportData **input)
{
	smlTrace(TRACE_ENTRY, "%s(%p, %p)", __func__, session, input);
	GError *error = NULL;
	
	SmlSessionThread *thread = g_malloc0(sizeof(SmlSessionThread));
	
	thread->context = g_main_context_new();
	g_main_context_ref(thread->context);
	
	thread->functions = g_malloc0(sizeof(GSourceFuncs));
	thread->functions->prepare = _session_prepare;
	thread->functions->check = _session_check;
	thread->functions->dispatch = _session_dispatch;
	thread->functions->finalize = NULL;

	thread->session_source = g_source_new(thread->functions, sizeof(GSource) + sizeof(SmlSession *));
	SmlSession **sessionptr = (SmlSession **)(thread->session_source + 1);
	*sessionptr = session;
	g_source_set_callback(thread->session_source, NULL, session, NULL);
	g_source_attach(thread->session_source, thread->context);
	
	thread->data_functions = g_malloc0(sizeof(GSourceFuncs));
	thread->data_functions->prepare = _data_prepare;
	thread->data_functions->check = _data_check;
	thread->data_functions->dispatch = _data_dispatch;
	thread->data_functions->finalize = NULL;
	thread->data = input;
	thread->session = session;
	
	thread->data_source = g_source_new(thread->data_functions, sizeof(GSource) + sizeof(SmlSession *));
	SmlSessionThread **threadptr = (SmlSessionThread **)(thread->data_source + 1);
	*threadptr = thread;
	g_source_set_callback(thread->data_source, NULL, thread, NULL);
	g_source_attach(thread->data_source, thread->context);
	
	thread->thread = smlThreadNew(thread->context, &error);
	sml_fail_unless(thread->thread != NULL, NULL);
	
	smlThreadStart(thread->thread);
	
	smlTrace(TRACE_EXIT, "%s", __func__);
	return thread;
}

/** @brief Stops a asynchronously running transport
 * 
 * @param tsp The transport
 * 
 */
void smlSessionStop(SmlSessionThread *thread)
{
	smlTrace(TRACE_ENTRY, "%s(%p)", __func__, thread);
	
	smlAssert(thread);

	smlThreadStop(thread->thread);

	g_source_destroy(thread->data_source);
	g_source_destroy(thread->session_source);
	g_source_unref(thread->data_source);
	g_source_unref(thread->session_source);
	
	smlThreadFree(thread->thread);
	
	g_free(thread->functions);
	g_free(thread->data_functions);

	g_main_context_unref(thread->context);
	g_main_context_unref(thread->context);
	
	g_free(thread);
	
	smlTrace(TRACE_EXIT, "%s", __func__);
}

static void _recv_init_alert(SmlDsSession *dsession, void *userdata)
{
	smlTrace(TRACE_ENTRY, "%s(%p, %p)", __func__, dsession, userdata);
	_server_corr *corr = userdata;
	
	corr->dsession = dsession;
	g_atomic_int_add(&init_alerts_received, 1);
	smlDsSessionRef(dsession);
	
	smlTrace(TRACE_EXIT, "%s", __func__);
}

static gboolean _recv_alert(SmlDsSession *dsession, SmlAlertType type, const char *last, const char *next, void *userdata)
{
	smlTrace(TRACE_ENTRY, "%s(%p, %i, %s, %s, %p)", __func__, dsession, type, VA_STRING(last), VA_STRING(next), userdata);
	
	sml_fail_unless(last != NULL, NULL);
	sml_fail_unless(next != NULL, NULL);
	
	g_atomic_int_add(&alerts_received, 1);
	
	smlTrace(TRACE_EXIT, "%s", __func__);
	return TRUE;
}

static void _recv_sync(SmlDsSession *dsession, gsize numChanges, void *userdata)
{
	smlTrace(TRACE_ENTRY, "%s(%p, %d, %p)", __func__,dsession, numChanges, userdata);
	
	g_atomic_int_add(&syncs_received, 1);
		
	smlTrace(TRACE_EXIT, "%s", __func__);
}

static gboolean _recv_changes(SmlDsSession *dsession, SmlDataSyncChangeItem *item, void *userdata, GError **error)
{
	smlTrace(TRACE_ENTRY, "%s(%p, %p, %p, %p)", __func__, dsession, item, userdata, error);

	if (sml_data_sync_change_item_get_action(item) == SML_CHANGE_ADD) {
		g_atomic_int_add(&adds_received, 1);
		g_assert(!strcmp(sml_location_get_uri(sml_data_sync_change_item_get_location(item)), "uid"));
		g_assert(!strncmp(sml_data_sync_change_item_get_data(item), "data", strlen(sml_data_sync_change_item_get_data(item))));
		g_assert(strlen(sml_data_sync_change_item_get_data(item)) == 4);
	} else if (sml_data_sync_change_item_get_action(item) == SML_CHANGE_REPLACE) {
		g_atomic_int_add(&modifies_received, 1);
		g_assert(!strcmp(sml_location_get_uri(sml_data_sync_change_item_get_location(item)), "uid"));
		g_assert(!strncmp(sml_data_sync_change_item_get_data(item), "newdata", strlen(sml_data_sync_change_item_get_data(item))));
		g_assert(strlen(sml_data_sync_change_item_get_data(item)) == 7);
	} else if (sml_data_sync_change_item_get_action(item) == SML_CHANGE_DELETE) {
		g_atomic_int_add(&deletes_received, 1);
		g_assert(!strcmp(sml_location_get_uri(sml_data_sync_change_item_get_location(item)), "uid"));
		g_assert(!sml_data_sync_change_item_get_data(item));
	} else {
		sml_fail_unless(FALSE, "An unexpected SyncML change type %d was received.", sml_data_sync_change_item_get_action(item));
	}

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

static void _recv_map(SmlDsSession *dsession, SmlMapItem *item, void *userdata)
{
	smlTrace(TRACE_ENTRY, "%s(%p, %p, %p)", __func__,dsession, item, userdata);
	
	g_atomic_int_add(&mappings_received, 1);
		
	smlTrace(TRACE_EXIT, "%s", __func__);
}

static void _alert_reply(SmlSession *dsession, SmlStatus *status, void *userdata)
{
	smlTrace(TRACE_ENTRY, "%s(%p, %p, %p)", __func__, dsession, status, userdata);
	
	sml_fail_unless(status != NULL, NULL);
	
	if (GPOINTER_TO_INT(userdata) == 1) {
		sml_fail_unless(smlStatusGetClass(status) == SML_ERRORCLASS_SUCCESS, NULL);
	} else {
		sml_fail_unless(smlStatusGetClass(status) == SML_ERRORCLASS_FATAL, NULL);
	}
	
	g_atomic_int_add(&replies_received, 1);
	
	smlTrace(TRACE_EXIT, "%s", __func__);
}

static void _sync_reply(SmlSession *dsession, SmlStatus *status, void *userdata)
{
	smlTrace(TRACE_ENTRY, "%s(%p, %p, %p)", __func__, dsession, status, userdata);
	
	sml_fail_unless(status != NULL, NULL);
	
	if (GPOINTER_TO_INT(userdata) == 1) {
		sml_fail_unless(smlStatusGetClass(status) == SML_ERRORCLASS_SUCCESS, NULL);
	} else {
		sml_fail_unless(smlStatusGetClass(status) == SML_ERRORCLASS_FATAL, NULL);
	}
	
	g_atomic_int_add(&replies_received, 1);
	
	smlTrace(TRACE_EXIT, "%s", __func__);
}

static void _add_reply(SmlDsSession *session, SmlStatus *status, void *userdata)
{
	smlTrace(TRACE_ENTRY, "%s(%p, %p, %p)", __func__, session, status, userdata);
	
	sml_fail_unless(status != NULL, NULL);
	sml_fail_unless(smlStatusGetClass(status) == SML_ERRORCLASS_SUCCESS, NULL);
	sml_fail_unless(GPOINTER_TO_INT(userdata) == 1 || GPOINTER_TO_INT(userdata) == 2, NULL);

	/* the newuid will only be received by the mapping callback */

	g_atomic_int_add(&add_replies_received, 1);

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

static void _modify_reply(SmlDsSession *session, SmlStatus *status, void *userdata)
{
	smlTrace(TRACE_ENTRY, "%s(%p, %p, %p)", __func__, session, status, userdata);
	
	sml_fail_unless(status != NULL, NULL);
	sml_fail_unless(smlStatusGetClass(status) == SML_ERRORCLASS_SUCCESS, NULL);
	sml_fail_unless(GPOINTER_TO_INT(userdata) == 1, NULL);

	g_atomic_int_add(&modify_replies_received, 1);

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

static void _delete_reply(SmlDsSession *session, SmlStatus *status, void *userdata)
{
	smlTrace(TRACE_ENTRY, "%s(%p, %p, %p)", __func__, session, status, userdata);
	
	sml_fail_unless(status != NULL, NULL);
	sml_fail_unless(smlStatusGetClass(status) == SML_ERRORCLASS_SUCCESS, NULL);
	sml_fail_unless(GPOINTER_TO_INT(userdata) == 1, NULL);

	g_atomic_int_add(&delete_replies_received, 1);
	
	smlTrace(TRACE_EXIT, "%s", __func__);
}

static void _map_reply(SmlSession *dsession, SmlStatus *status, void *userdata)
{
	smlTrace(TRACE_ENTRY, "%s(%p, %p, %p)", __func__, dsession, status, userdata);
	
	sml_fail_unless(status != NULL, NULL);
	sml_fail_unless(smlStatusGetClass(status) == SML_ERRORCLASS_SUCCESS, NULL);
	
	g_atomic_int_add(&map_replies_received, 1);
	
	smlTrace(TRACE_EXIT, "%s", __func__);
}

START_TEST (datastore_server_new)
{
	setup_testbed(NULL);
	
	GError *error = NULL;
	SmlLocation *loc = sml_location_new();
	sml_fail_unless(loc != NULL, NULL);
	sml_location_set_uri(loc, "/vcards");
	SmlDsServer *server = smlDsServerNew(SML_ELEMENT_TEXT_VCARD, loc, &error);
	sml_fail_unless(server != NULL, NULL);
	sml_fail_unless(error == NULL, NULL);
	
	g_object_unref(loc);
	
	smlDsServerFree(server);
}
END_TEST

START_TEST (datastore_client_new)
{
	setup_testbed(NULL);
	
	GError *error = NULL;
	SmlLocation *loc = sml_location_new();
	sml_fail_unless(loc != NULL, NULL);
	sml_location_set_uri(loc, "/vcards");
	SmlDsServer *server = smlDsClientNew(SML_ELEMENT_TEXT_VCARD, loc, loc, &error);
	sml_fail_unless(server != NULL, NULL);
	sml_fail_unless(error == NULL, NULL);
	
	g_object_unref(loc);
	
	smlDsServerFree(server);
}
END_TEST

static void _event_callback(SmlSession *session, SmlSessionEventType type, SmlCommand *command, SmlCommand *parent, SmlStatus *reply, const GError *error, void *userdata)
{
	smlTrace(TRACE_ENTRY, "%s(%p, %p)", __func__, parent, reply);
	
	_server_corr *corr = userdata;
	
	switch (type) {
		case SML_SESSION_EVENT_ESTABLISHED:
		case SML_SESSION_EVENT_HEADER_REPLY:
			break;
		case SML_SESSION_EVENT_CHILD_COMMAND:
		case SML_SESSION_EVENT_COMMAND_START:
			switch (command->type) {
				case SML_COMMAND_TYPE_ALERT:
					if (!corr->dsession)
						corr->dsession = smlDsServerRecvAlert(corr->server, session, command);
					else
						smlDsSessionRecvAlert(session, command, corr->dsession);
						
					break;
				case SML_COMMAND_TYPE_SYNC:
					sml_fail_unless(corr->dsession != NULL, NULL);
					smlDsSessionRecvSync(session, command, corr->dsession);
					break;
				case SML_COMMAND_TYPE_ADD:
				case SML_COMMAND_TYPE_DELETE:
				case SML_COMMAND_TYPE_REPLACE:
					sml_fail_unless(corr->dsession != NULL, NULL);
					smlDsSessionRecvChange(session, command, corr->dsession);
					break;
				case SML_COMMAND_TYPE_MAP:
					sml_fail_unless(corr->dsession != NULL, NULL);
					smlDsSessionRecvMap(session, command, corr->dsession);
					break;
				default:
					sml_fail_unless(FALSE, "An unexpected command type %d was received.", command->type);
					break;
			}
			break;
		case SML_SESSION_EVENT_COMMAND_END:
			switch (command->type) {
				case SML_COMMAND_TYPE_SYNC:
					sml_fail_unless(corr->dsession != NULL, NULL);
					smlDsSessionRecvSync(session, command, corr->dsession);
					break;
				default:
					sml_fail_unless(FALSE, "An unexpected command type %d was received.", command->type);
					break;
			}
			break;
		case SML_SESSION_EVENT_FLUSH:
		case SML_SESSION_EVENT_FINAL:
			break;
		case SML_SESSION_EVENT_END:
			g_atomic_int_add(&session_ends, 1);
			break;
		case SML_SESSION_EVENT_ERROR:
			sml_fail_unless(FALSE, "%s", error->message);
			break;
		default:
			sml_fail_unless(FALSE, "An unexpected session event %d was received.", type);
			break;
	}
	
	smlTrace(TRACE_EXIT, "%s", __func__);
}

static void _data_send_callback(SmlSession *session, SmlTransportData *data, void *userdata)
{
	smlTrace(TRACE_ENTRY, "%s(%p, %p, %p)", __func__, session, data, userdata);
	
	smlTrace(TRACE_INTERNAL, "%s: setting data(data: %p, real %p, size %i)", __func__, data, data->data, data->size);
	smlTransportDataRef(data);
	SmlTransportData **dataptr = userdata;
	*dataptr = data;
	
	data_received++;
	
	smlTrace(TRACE_EXIT, "%s", __func__);
}

/* Send Alert -> C -- Put, Alert --> S -> init_alert, get_alert, recv_alert
 * end session -> S -- Status --> C -> manager event (Session End)
 */
START_TEST (datastore_send_alert)
{
	/* General setup */
	setup_testbed(NULL);
	
	GError *error = NULL;
	SmlLocation *loc = sml_location_new();
	sml_fail_unless(loc != NULL, NULL);
	sml_location_set_uri(loc, "/vcards");

	SmlDsServer *client = smlDsClientNew(SML_ELEMENT_TEXT_VCARD, loc, loc, &error);
	_server_corr *client_corr = g_malloc0(sizeof(_server_corr));
	client_corr->server = client;
	
	SmlDsServer *server = smlDsServerNew(SML_ELEMENT_TEXT_VCARD, loc, &error);
	_server_corr *server_corr = g_malloc0(sizeof(_server_corr));
	server_corr->server = server;
	smlDsServerSetConnectCallback(server, _recv_init_alert, server_corr);
	
	/* Start a new session from the client */
	SmlSession *session = smlSessionNew(SML_SESSION_TYPE_CLIENT, SML_MIMETYPE_XML, SML_VERSION_12, SML_PROTOCOL_SYNCML, loc, loc, 1, 0, &error);
	client_corr->thread = smlSessionRunAsync(session, &to_session);
	SmlSession *session2 = smlSessionNew(SML_SESSION_TYPE_SERVER, SML_MIMETYPE_XML, SML_VERSION_12, SML_PROTOCOL_SYNCML, loc, loc, 1, 0, &error);
	server_corr->thread = smlSessionRunAsync(session2, &to_session2);
	smlSessionSetLocalMaxMsgSize(session2, defaultMaxMsgSize);
	smlSessionSetLocalMaxObjSize(session2, defaultMaxObjSize);
	
	g_object_unref(loc);
	
	smlSessionSetEventCallback(session, _event_callback, client_corr);
	smlSessionSetDataCallback(session, _data_send_callback, &to_session2);
	smlSessionSetEventCallback(session2, _event_callback, server_corr);
	smlSessionSetDataCallback(session2, _data_send_callback, &to_session);
	
	client_corr->dsession = smlDsServerSendAlert(client, session, SML_ALERT_TWO_WAY, "last", "next", _alert_reply, GINT_TO_POINTER(1), &error);
	sml_fail_unless(client_corr->dsession != NULL, NULL);
	sml_fail_unless(error == NULL, NULL);
	
	sml_fail_unless(smlSessionFlush(session, TRUE, &error), NULL);
	sml_fail_unless(error == NULL, NULL);
	
	while (init_alerts_received != 1) {
		usleep(500);
	}
	
	smlDsSessionGetAlert(server_corr->dsession, _recv_alert, GINT_TO_POINTER(1));
	
	while (alerts_received != 1) {
		usleep(500);
	}
	
	sml_fail_unless(smlSessionEnd(session2, &error), NULL);
	sml_fail_unless(error == NULL, NULL);
	
	while (replies_received != 1) {
		usleep(500);
	}
	
	while (session_ends != 2) {
		usleep(500);
	}
	
	smlSessionStop(client_corr->thread);
	smlSessionStop(server_corr->thread);
	
	smlDsSessionUnref(client_corr->dsession);
	smlDsSessionUnref(server_corr->dsession);
	
	sml_fail_unless(data_received == 2, NULL);
	sml_fail_unless(init_alerts_received == 1, NULL);
	
	sml_fail_unless(alerts_received == 1, NULL);
	sml_fail_unless(syncs_received == 0, NULL);
	sml_fail_unless(changes_received == 0, NULL);
	sml_fail_unless(replies_received == 1, NULL);
	sml_fail_unless(session_ends == 2, NULL);
	
	g_free(client_corr);
	g_free(server_corr);
	
	smlSessionUnref(session);
	smlSessionUnref(session2);
		
	smlDsServerFree(server);
	smlDsServerFree(client);
	
}
END_TEST

/* C -- Put, Alert --> S
 * S -- Status, Put, Alert --> C
 * C -- Status --> S 
 * S -- Status --> C */
START_TEST (datastore_send_alert2)
{
	data_received = 0;
	init_alerts_received = 0;
	alerts_received = 0;
	syncs_received = 0;
	sync_ends_received = 0;
	changes_received = 0;
	replies_received = 0;
	adds_received = 0;
	add_replies_received = 0;
	deletes_received = 0;
	delete_replies_received = 0;
	modifies_received = 0;
	modify_replies_received = 0;
	session_ends = 0;
	
	to_session = NULL;
	to_session2 = NULL;

	/* General setup */
	setup_testbed(NULL);
	
	GError *error = NULL;
	SmlLocation *loc = sml_location_new();
	sml_fail_unless(loc != NULL, NULL);
	sml_location_set_uri(loc, "/vcards");
	
	SmlDsServer *client = smlDsClientNew(SML_ELEMENT_TEXT_VCARD, loc, loc, &error);
	_server_corr *client_corr = g_malloc0(sizeof(_server_corr));
	client_corr->server = client;
	
	SmlDsServer *server = smlDsServerNew(SML_ELEMENT_TEXT_VCARD, loc, &error);
	_server_corr *server_corr = g_malloc0(sizeof(_server_corr));
	server_corr->server = server;
	smlDsServerSetConnectCallback(server, _recv_init_alert, server_corr);
	
	
	/* Start a new session from the client */
	SmlSession *session = smlSessionNew(SML_SESSION_TYPE_CLIENT, SML_MIMETYPE_XML, SML_VERSION_12, SML_PROTOCOL_SYNCML, loc, loc, 1, 0, &error);
	client_corr->thread = smlSessionRunAsync(session, &to_session);
	SmlSession *session2 = smlSessionNew(SML_SESSION_TYPE_SERVER, SML_MIMETYPE_XML, SML_VERSION_12, SML_PROTOCOL_SYNCML, loc, loc, 1, 0, &error);
	server_corr->thread = smlSessionRunAsync(session2, &to_session2);
	smlSessionSetLocalMaxMsgSize(session2, defaultMaxMsgSize);
	smlSessionSetLocalMaxObjSize(session2, defaultMaxObjSize);
	
	smlSessionSetEventCallback(session, _event_callback, client_corr);
	smlSessionSetDataCallback(session, _data_send_callback, &to_session2);
	smlSessionSetEventCallback(session2, _event_callback, server_corr);
	smlSessionSetDataCallback(session2, _data_send_callback, &to_session);
	
	client_corr->dsession = smlDsServerSendAlert(client, session, SML_ALERT_TWO_WAY, "last", "next", _alert_reply, GINT_TO_POINTER(1), &error);
	sml_fail_unless(client_corr->dsession != NULL, NULL);
	sml_fail_unless(error == NULL, NULL);
	
	sml_fail_unless(smlSessionFlush(session, TRUE, &error), NULL);
	sml_fail_unless(error == NULL, NULL);
	
	while (init_alerts_received != 1) {
		usleep(500);
	}
	
	smlDsSessionGetAlert(server_corr->dsession, _recv_alert, GINT_TO_POINTER(1));
	smlDsSessionSendAlert(server_corr->dsession, SML_ALERT_TWO_WAY, "last", "next", _alert_reply, GINT_TO_POINTER(1), &error);
	
	g_object_unref(loc);
	
	while (alerts_received != 1) {
		usleep(500);
	}
	
	sml_fail_unless(smlSessionFlush(session2, TRUE, &error), NULL);
	sml_fail_unless(error == NULL, NULL);
	
	while (data_received != 2) {
		usleep(500);
	}
	
	smlDsSessionGetAlert(client_corr->dsession, _recv_alert, GINT_TO_POINTER(1));
	
	
	
	while (alerts_received != 2) {
		smlDsSessionDispatch(client_corr->dsession);
		usleep(500);
	}
	
	sml_fail_unless(smlSessionEnd(session, &error), NULL);
	sml_fail_unless(error == NULL, NULL);
	
	while (session_ends != 2) {
		usleep(500);
	}
	
	smlSessionStop(client_corr->thread);
	smlSessionStop(server_corr->thread);
	
	smlDsSessionUnref(client_corr->dsession);
	smlDsSessionUnref(server_corr->dsession);
	
	sml_fail_unless(data_received == 4, NULL);
	sml_fail_unless(init_alerts_received == 1, NULL);
	
	sml_fail_unless(alerts_received == 2, NULL);
	sml_fail_unless(syncs_received == 0, NULL);
	sml_fail_unless(changes_received == 0, NULL);
	sml_fail_unless(replies_received == 2, NULL);
	sml_fail_unless(session_ends == 2, NULL);
	
	g_free(client_corr);
	g_free(server_corr);
	
	smlSessionUnref(session);
	smlSessionUnref(session2);
		
	smlDsServerFree(server);
	smlDsServerFree(client);
}
END_TEST

/* C -- Put, Alert --> S
 * S -- Status, Put, Alert --> C
 * C -- Status, Sync --> S
 * S -- Status --> S */
START_TEST (datastore_send_empty_sync)
{
	data_received = 0;
	init_alerts_received = 0;
	alerts_received = 0;
	syncs_received = 0;
	sync_ends_received = 0;
	changes_received = 0;
	replies_received = 0;
	adds_received = 0;
	add_replies_received = 0;
	deletes_received = 0;
	delete_replies_received = 0;
	modifies_received = 0;
	modify_replies_received = 0;
	session_ends = 0;
	
	to_session = NULL;
	to_session2 = NULL;
	
	/* General setup */
	setup_testbed(NULL);
	
	GError *error = NULL;
	SmlLocation *loc = sml_location_new();
	sml_fail_unless(loc != NULL, NULL);
	sml_location_set_uri(loc, "/vcards");
	
	
	SmlDsServer *client = smlDsClientNew(SML_ELEMENT_TEXT_VCARD, loc, loc, &error);
	_server_corr *client_corr = g_malloc0(sizeof(_server_corr));
	client_corr->server = client;
	
	
	SmlDsServer *server = smlDsServerNew(SML_ELEMENT_TEXT_VCARD, loc, &error);
	_server_corr *server_corr = g_malloc0(sizeof(_server_corr));
	server_corr->server = server;
	smlDsServerSetConnectCallback(server, _recv_init_alert, server_corr);
	
	
	/* Start a new session from the client */
	SmlSession *session = smlSessionNew(SML_SESSION_TYPE_CLIENT, SML_MIMETYPE_XML, SML_VERSION_12, SML_PROTOCOL_SYNCML, loc, loc, 1, 0, &error);
	client_corr->thread = smlSessionRunAsync(session, &to_session);
	SmlSession *session2 = smlSessionNew(SML_SESSION_TYPE_SERVER, SML_MIMETYPE_XML, SML_VERSION_12, SML_PROTOCOL_SYNCML, loc, loc, 1, 0, &error);
	server_corr->thread = smlSessionRunAsync(session2, &to_session2);
	smlSessionSetLocalMaxMsgSize(session2, defaultMaxMsgSize);
	smlSessionSetLocalMaxObjSize(session2, defaultMaxObjSize);
	
	smlSessionSetEventCallback(session, _event_callback, client_corr);
	smlSessionSetDataCallback(session, _data_send_callback, &to_session2);
	smlSessionSetEventCallback(session2, _event_callback, server_corr);
	smlSessionSetDataCallback(session2, _data_send_callback, &to_session);
	
	client_corr->dsession = smlDsServerSendAlert(client, session, SML_ALERT_TWO_WAY, "last", "next", _alert_reply, GINT_TO_POINTER(1), &error);
	sml_fail_unless(client_corr->dsession != NULL, NULL);
	sml_fail_unless(error == NULL, NULL);
	
	sml_fail_unless(smlSessionFlush(session, TRUE, &error), NULL);
	sml_fail_unless(error == NULL, NULL);
	
	while (init_alerts_received != 1) {
		usleep(500);
	}
	
	smlDsSessionGetAlert(server_corr->dsession, _recv_alert, GINT_TO_POINTER(1));
	smlDsSessionSendAlert(server_corr->dsession, SML_ALERT_TWO_WAY, "last", "next", _alert_reply, GINT_TO_POINTER(1), &error);
	
	
	
	while (alerts_received != 1) {
		smlDsSessionDispatch(server_corr->dsession);
		usleep(500);
	}
	
	sml_fail_unless(smlSessionFlush(session2, TRUE, &error), NULL);
	sml_fail_unless(error == NULL, NULL);
	
	while (data_received != 2) {
		usleep(500);
	}
	
	smlDsSessionGetAlert(client_corr->dsession, _recv_alert, GINT_TO_POINTER(1));
	
	
	
	while (alerts_received != 2) {
		smlDsSessionDispatch(client_corr->dsession);
		usleep(500);
	}
	
	smlDsSessionSendSync(client_corr->dsession, 0, _sync_reply, GINT_TO_POINTER(1), &error);
	smlDsSessionCloseSync(client_corr->dsession, &error);
	
	sml_fail_unless(smlSessionFlush(session, TRUE, &error), NULL);
	sml_fail_unless(error == NULL, NULL);
	
	while (data_received != 3) {
		usleep(500);
	}
	
	smlDsSessionGetSync(server_corr->dsession, _recv_sync, GINT_TO_POINTER(1));
	
	while (syncs_received != 1) {
		smlDsSessionDispatch(server_corr->dsession);
		usleep(500);
	}
	
	sml_fail_unless(smlSessionEnd(session2, &error), NULL);
	sml_fail_unless(error == NULL, NULL);
	
	while (session_ends != 2) {
		usleep(500);
	}
	
	smlSessionStop(client_corr->thread);
	smlSessionStop(server_corr->thread);
	
	smlDsSessionUnref(client_corr->dsession);
	smlDsSessionUnref(server_corr->dsession);
	
	sml_fail_unless(data_received == 4, NULL);
	sml_fail_unless(init_alerts_received == 1, NULL);
	
	sml_fail_unless(alerts_received == 2, NULL);
	sml_fail_unless(syncs_received == 1, NULL);
	sml_fail_unless(changes_received == 0, NULL);
	sml_fail_unless(replies_received == 3, NULL);
	sml_fail_unless(session_ends == 2, NULL);
	
	g_free(client_corr);
	g_free(server_corr);
	
	smlSessionUnref(session);
	smlSessionUnref(session2);
	
	g_object_unref(loc);
	
	smlDsServerFree(server);
	smlDsServerFree(client);
}
END_TEST

/* C -- Put, Alert --> S
 * S -- Status, Put, Alert --> C
 * C -- Status, Sync --> S
 * S -- Status, Sync --> S
 * C -- Status --> S
 * S -- Status --> C */
START_TEST (datastore_reply_empty_sync)
{
	data_received = 0;
	init_alerts_received = 0;
	alerts_received = 0;
	syncs_received = 0;
	sync_ends_received = 0;
	changes_received = 0;
	replies_received = 0;
	adds_received = 0;
	add_replies_received = 0;
	deletes_received = 0;
	delete_replies_received = 0;
	modifies_received = 0;
	modify_replies_received = 0;
	session_ends = 0;
	
	to_session = NULL;
	to_session2 = NULL;
	
	/* General setup */
	setup_testbed(NULL);
	
	GError *error = NULL;
	SmlLocation *loc = sml_location_new();
	sml_fail_unless(loc != NULL, NULL);
	sml_location_set_uri(loc, "/vcards");
	
	
	SmlDsServer *client = smlDsClientNew(SML_ELEMENT_TEXT_VCARD, loc, loc, &error);
	_server_corr *client_corr = g_malloc0(sizeof(_server_corr));
	client_corr->server = client;
	
	
	SmlDsServer *server = smlDsServerNew(SML_ELEMENT_TEXT_VCARD, loc, &error);
	_server_corr *server_corr = g_malloc0(sizeof(_server_corr));
	server_corr->server = server;
	smlDsServerSetConnectCallback(server, _recv_init_alert, server_corr);
	
	
	/* Start a new session from the client */
	SmlSession *session = smlSessionNew(SML_SESSION_TYPE_CLIENT, SML_MIMETYPE_XML, SML_VERSION_12, SML_PROTOCOL_SYNCML, loc, loc, 1, 0, &error);
	client_corr->thread = smlSessionRunAsync(session, &to_session);
	SmlSession *session2 = smlSessionNew(SML_SESSION_TYPE_SERVER, SML_MIMETYPE_XML, SML_VERSION_12, SML_PROTOCOL_SYNCML, loc, loc, 1, 0, &error);
	server_corr->thread = smlSessionRunAsync(session2, &to_session2);
	smlSessionSetLocalMaxMsgSize(session2, defaultMaxMsgSize);
	smlSessionSetLocalMaxObjSize(session2, defaultMaxObjSize);
	
	smlSessionSetEventCallback(session, _event_callback, client_corr);
	smlSessionSetDataCallback(session, _data_send_callback, &to_session2);
	smlSessionSetEventCallback(session2, _event_callback, server_corr);
	smlSessionSetDataCallback(session2, _data_send_callback, &to_session);
	
	client_corr->dsession = smlDsServerSendAlert(client, session, SML_ALERT_TWO_WAY, "last", "next", _alert_reply, GINT_TO_POINTER(1), &error);
	sml_fail_unless(client_corr->dsession != NULL, NULL);
	sml_fail_unless(error == NULL, NULL);
	
	sml_fail_unless(smlSessionFlush(session, TRUE, &error), NULL);
	sml_fail_unless(error == NULL, NULL);
	
	while (init_alerts_received != 1) {
		usleep(500);
	}
	
	smlDsSessionGetAlert(server_corr->dsession, _recv_alert, GINT_TO_POINTER(1));
	smlDsSessionSendAlert(server_corr->dsession, SML_ALERT_TWO_WAY, "last", "next", _alert_reply, GINT_TO_POINTER(1), &error);
	
	
	
	while (alerts_received != 1) {
		smlDsSessionDispatch(server_corr->dsession);
		usleep(500);
	}
	
	sml_fail_unless(smlSessionFlush(session2, TRUE, &error), NULL);
	sml_fail_unless(error == NULL, NULL);
	
	while (data_received != 2) {
		usleep(500);
	}
	
	smlDsSessionGetAlert(client_corr->dsession, _recv_alert, GINT_TO_POINTER(1));
	
	
	
	while (alerts_received != 2) {
		smlDsSessionDispatch(client_corr->dsession);
		usleep(500);
	}
	
	smlDsSessionSendSync(client_corr->dsession, 0, _sync_reply, GINT_TO_POINTER(1), &error);
	smlDsSessionCloseSync(client_corr->dsession, &error);
	
	sml_fail_unless(smlSessionFlush(session, TRUE, &error), NULL);
	sml_fail_unless(error == NULL, NULL);
	
	while (data_received != 3) {
		usleep(500);
	}
	
	smlDsSessionGetSync(server_corr->dsession, _recv_sync, GINT_TO_POINTER(1));
	
	while (syncs_received != 1) {
		smlDsSessionDispatch(server_corr->dsession);
		usleep(500);
	}
	
	smlDsSessionSendSync(server_corr->dsession, 0, _sync_reply, GINT_TO_POINTER(1), &error);
	smlDsSessionCloseSync(server_corr->dsession, &error);
	
	sml_fail_unless(smlSessionFlush(session2, TRUE, &error), NULL);
	sml_fail_unless(error == NULL, NULL);
	
	while (data_received != 4) {
		usleep(500);
	}
	
	smlDsSessionGetSync(client_corr->dsession, _recv_sync, GINT_TO_POINTER(1));
	
	while (syncs_received != 2) {
		smlDsSessionDispatch(client_corr->dsession);
		usleep(500);
	}
	
	sml_fail_unless(smlSessionEnd(session, &error), NULL);
	sml_fail_unless(error == NULL, NULL);
	
	while (session_ends != 2) {
		usleep(500);
	}
	
	smlSessionStop(client_corr->thread);
	smlSessionStop(server_corr->thread);
	
	smlDsSessionUnref(client_corr->dsession);
	smlDsSessionUnref(server_corr->dsession);
	
	sml_fail_unless(data_received == 6, NULL);
	sml_fail_unless(init_alerts_received == 1, NULL);
	
	sml_fail_unless(alerts_received == 2, NULL);
	sml_fail_unless(syncs_received == 2, NULL);
	sml_fail_unless(changes_received == 0, NULL);
	sml_fail_unless(replies_received == 4, NULL);
	sml_fail_unless(session_ends == 2, NULL);
	
	g_free(client_corr);
	g_free(server_corr);
	
	smlSessionUnref(session);
	smlSessionUnref(session2);
		
	g_object_unref(loc);
	
	smlDsServerFree(server);
	smlDsServerFree(client);
}
END_TEST

/* send alerts, send sync from server, send from client with one add */
/* Send no mapitems */
/* C -- Put, Alert --> S
 * S -- Status, Put, Alert --> C
 * C -- Status, Sync, Add --> S
 * S -- Status, Sync --> S
 * C -- Status --> S
 * S -- Status --> C */
START_TEST (datastore_add)
{
	data_received = 0;
	init_alerts_received = 0;
	alerts_received = 0;
	syncs_received = 0;
	sync_ends_received = 0;
	changes_received = 0;
	replies_received = 0;
	adds_received = 0;
	add_replies_received = 0;
	deletes_received = 0;
	delete_replies_received = 0;
	modifies_received = 0;
	modify_replies_received = 0;
	session_ends = 0;
	
	to_session = NULL;
	to_session2 = NULL;
	
	/* General setup */
	setup_testbed(NULL);
	
	GError *error = NULL;
	SmlLocation *loc = sml_location_new();
	sml_fail_unless(loc != NULL, NULL);
	sml_location_set_uri(loc, "/vcards");
	
	
	SmlDsServer *client = smlDsClientNew(SML_ELEMENT_TEXT_VCARD, loc, loc, &error);
	_server_corr *client_corr = g_malloc0(sizeof(_server_corr));
	client_corr->server = client;
	
	
	SmlDsServer *server = smlDsServerNew(SML_ELEMENT_TEXT_VCARD, loc, &error);
	_server_corr *server_corr = g_malloc0(sizeof(_server_corr));
	server_corr->server = server;
	smlDsServerSetConnectCallback(server, _recv_init_alert, server_corr);
	
	
	/* Start a new session from the client */
	SmlSession *session = smlSessionNew(SML_SESSION_TYPE_CLIENT, SML_MIMETYPE_XML, SML_VERSION_12, SML_PROTOCOL_SYNCML, loc, loc, 1, 0, &error);
	client_corr->thread = smlSessionRunAsync(session, &to_session);
	SmlSession *session2 = smlSessionNew(SML_SESSION_TYPE_SERVER, SML_MIMETYPE_XML, SML_VERSION_12, SML_PROTOCOL_SYNCML, loc, loc, 1, 0, &error);
	server_corr->thread = smlSessionRunAsync(session2, &to_session2);
	smlSessionSetLocalMaxMsgSize(session2, defaultMaxMsgSize);
	smlSessionSetLocalMaxObjSize(session2, defaultMaxObjSize);
	
	smlSessionSetEventCallback(session, _event_callback, client_corr);
	smlSessionSetDataCallback(session, _data_send_callback, &to_session2);
	smlSessionSetEventCallback(session2, _event_callback, server_corr);
	smlSessionSetDataCallback(session2, _data_send_callback, &to_session);
	
	client_corr->dsession = smlDsServerSendAlert(client, session, SML_ALERT_TWO_WAY, "last", "next", _alert_reply, GINT_TO_POINTER(1), &error);
	sml_fail_unless(client_corr->dsession != NULL, NULL);
	sml_fail_unless(error == NULL, NULL);
	
	sml_fail_unless(smlSessionFlush(session, TRUE, &error), NULL);
	sml_fail_unless(error == NULL, NULL);
	
	while (init_alerts_received != 1) {
		usleep(500);
	}
	
	smlDsSessionGetAlert(server_corr->dsession, _recv_alert, GINT_TO_POINTER(1));
	smlDsSessionSendAlert(server_corr->dsession, SML_ALERT_TWO_WAY, "last", "next", _alert_reply, GINT_TO_POINTER(1), &error);
	
	
	
	while (alerts_received != 1) {
		smlDsSessionDispatch(server_corr->dsession);
		usleep(500);
	}
	
	sml_fail_unless(smlSessionFlush(session2, TRUE, &error), NULL);
	sml_fail_unless(error == NULL, NULL);
	
	while (data_received != 2) {
		usleep(500);
	}
	
	smlDsSessionGetAlert(client_corr->dsession, _recv_alert, GINT_TO_POINTER(1));
	
	while (alerts_received != 2) {
		smlDsSessionDispatch(client_corr->dsession);
		usleep(500);
	}
	
	smlDsSessionSendSync(client_corr->dsession, 0, _sync_reply, GINT_TO_POINTER(1), &error);
	
	TEST_DATA_SYNC_CHANGE_ITEM_NEW(SML_CHANGE_ADD, "uid", "data", 4, SML_ELEMENT_TEXT_VCARD);
	sml_fail_unless(smlDsSessionQueueChange(client_corr->dsession, TEST_DATA_SYNC_CHANGE_ITEM, _add_reply, GINT_TO_POINTER(2), &error), NULL);
	TEST_DATA_SYNC_CHANGE_ITEM_FREE();
	sml_fail_unless(error == NULL, NULL);
	
	smlDsSessionCloseSync(client_corr->dsession, &error);
	
	sml_fail_unless(smlSessionFlush(session, TRUE, &error), NULL);
	sml_fail_unless(error == NULL, NULL);
	
	while (data_received != 3) {
		usleep(500);
	}
	
	smlDsSessionGetSync(server_corr->dsession, _recv_sync, GINT_TO_POINTER(1));
	smlDsSessionGetChanges(server_corr->dsession, _recv_changes, GINT_TO_POINTER(1));
	smlDsSessionGetMapping(server_corr->dsession, _recv_map, GINT_TO_POINTER(1));
	
	while (syncs_received != 1 || adds_received != 1) {
		smlDsSessionDispatch(server_corr->dsession);
		usleep(500);
	}
	
	smlDsSessionSendSync(server_corr->dsession, 0, _sync_reply, GINT_TO_POINTER(1), &error);
	smlDsSessionCloseSync(server_corr->dsession, &error);
	
	sml_fail_unless(smlSessionFlush(session2, TRUE, &error), NULL);
	sml_fail_unless(error == NULL, NULL);
	
	while (data_received != 4) {
		usleep(500);
	}
	
	smlDsSessionGetSync(client_corr->dsession, _recv_sync, GINT_TO_POINTER(1));
	
	while (syncs_received != 2) {
		smlDsSessionDispatch(client_corr->dsession);
		usleep(500);
	}
	
	sml_fail_unless(smlSessionEnd(session, &error), NULL);
	sml_fail_unless(error == NULL, NULL);
	
	while (session_ends != 2) {
		usleep(500);
	}
	
	smlSessionStop(client_corr->thread);
	smlSessionStop(server_corr->thread);
	
	smlDsSessionUnref(client_corr->dsession);
	smlDsSessionUnref(server_corr->dsession);
	
	sml_fail_unless(data_received == 6, NULL);
	sml_fail_unless(init_alerts_received == 1, NULL);
	
	sml_fail_unless(alerts_received == 2, NULL);
	sml_fail_unless(syncs_received == 2, NULL);
	sml_fail_unless(changes_received == 0, NULL);
	sml_fail_unless(replies_received == 4, NULL);
	sml_fail_unless(session_ends == 2, NULL);
	
	sml_fail_unless(adds_received == 1, NULL);
	sml_fail_unless(add_replies_received == 1, NULL);
	sml_fail_unless(deletes_received == 0, NULL);
	sml_fail_unless(delete_replies_received == 0, NULL);
	sml_fail_unless(modifies_received == 0, NULL);
	sml_fail_unless(modify_replies_received == 0, NULL);
	
	g_free(client_corr);
	g_free(server_corr);
	
	smlSessionUnref(session);
	smlSessionUnref(session2);
		
	g_object_unref(loc);
	
	smlDsServerFree(server);
	smlDsServerFree(client);
}
END_TEST

/* send alerts, send sync from server, send from client with one add */
/* C -- Put, Alert --> S
 * S -- Status, Put, Alert --> C
 * C -- Status, Sync, Add --> S
 * S -- Status, Sync, Add --> S
 * C -- Status, Map --> S
 * S -- Status --> C */
START_TEST (datastore_add2)
{
	data_received = 0;
	init_alerts_received = 0;
	alerts_received = 0;
	syncs_received = 0;
	sync_ends_received = 0;
	changes_received = 0;
	replies_received = 0;
	adds_received = 0;
	add_replies_received = 0;
	deletes_received = 0;
	delete_replies_received = 0;
	modifies_received = 0;
	modify_replies_received = 0;
	session_ends = 0;
	
	to_session = NULL;
	to_session2 = NULL;
	
	/* General setup */
	setup_testbed(NULL);
	
	GError *error = NULL;
	SmlLocation *loc = sml_location_new();
	sml_fail_unless(loc != NULL, NULL);
	sml_location_set_uri(loc, "/vcards");
	
	
	SmlDsServer *client = smlDsClientNew(SML_ELEMENT_TEXT_VCARD, loc, loc, &error);
	_server_corr *client_corr = g_malloc0(sizeof(_server_corr));
	client_corr->server = client;
	
	
	SmlDsServer *server = smlDsServerNew(SML_ELEMENT_TEXT_VCARD, loc, &error);
	_server_corr *server_corr = g_malloc0(sizeof(_server_corr));
	server_corr->server = server;
	smlDsServerSetConnectCallback(server, _recv_init_alert, server_corr);
	
	
	/* Start a new session from the client */
	SmlSession *session = smlSessionNew(SML_SESSION_TYPE_CLIENT, SML_MIMETYPE_XML, SML_VERSION_12, SML_PROTOCOL_SYNCML, loc, loc, 1, 0, &error);
	client_corr->thread = smlSessionRunAsync(session, &to_session);
	SmlSession *session2 = smlSessionNew(SML_SESSION_TYPE_SERVER, SML_MIMETYPE_XML, SML_VERSION_12, SML_PROTOCOL_SYNCML, loc, loc, 1, 0, &error);
	server_corr->thread = smlSessionRunAsync(session2, &to_session2);
	smlSessionSetLocalMaxMsgSize(session2, defaultMaxMsgSize);
	smlSessionSetLocalMaxObjSize(session2, defaultMaxObjSize);
	
	smlSessionSetEventCallback(session, _event_callback, client_corr);
	smlSessionSetDataCallback(session, _data_send_callback, &to_session2);
	smlSessionSetEventCallback(session2, _event_callback, server_corr);
	smlSessionSetDataCallback(session2, _data_send_callback, &to_session);
	
	client_corr->dsession = smlDsServerSendAlert(client, session, SML_ALERT_TWO_WAY, "last", "next", _alert_reply, GINT_TO_POINTER(1), &error);
	sml_fail_unless(client_corr->dsession != NULL, NULL);
	sml_fail_unless(error == NULL, NULL);
	
	sml_fail_unless(smlSessionFlush(session, TRUE, &error), NULL);
	sml_fail_unless(error == NULL, NULL);
	
	while (init_alerts_received != 1) {
		usleep(500);
	}
	
	smlDsSessionGetAlert(server_corr->dsession, _recv_alert, GINT_TO_POINTER(1));
	smlDsSessionSendAlert(server_corr->dsession, SML_ALERT_TWO_WAY, "last", "next", _alert_reply, GINT_TO_POINTER(1), &error);
	
	
	
	while (alerts_received != 1) {
		smlDsSessionDispatch(server_corr->dsession);
		usleep(500);
	}
	
	sml_fail_unless(smlSessionFlush(session2, TRUE, &error), NULL);
	sml_fail_unless(error == NULL, NULL);
	
	while (data_received != 2) {
		usleep(500);
	}
	
	smlDsSessionGetAlert(client_corr->dsession, _recv_alert, GINT_TO_POINTER(1));
	
	
	
	while (alerts_received != 2) {
		smlDsSessionDispatch(client_corr->dsession);
		usleep(500);
	}
	
	smlDsSessionSendSync(client_corr->dsession, 1, _sync_reply, GINT_TO_POINTER(1), &error);
	
	TEST_DATA_SYNC_CHANGE_ITEM_NEW(SML_CHANGE_ADD, "uid", "data", 4, SML_ELEMENT_TEXT_VCARD);
	sml_fail_unless(smlDsSessionQueueChange(client_corr->dsession, TEST_DATA_SYNC_CHANGE_ITEM, _add_reply, GINT_TO_POINTER(2), &error), NULL);
	TEST_DATA_SYNC_CHANGE_ITEM_FREE();
	sml_fail_unless(error == NULL, NULL);
	
	smlDsSessionCloseSync(client_corr->dsession, &error);
	
	sml_fail_unless(smlSessionFlush(session, TRUE, &error), NULL);
	sml_fail_unless(error == NULL, NULL);
	
	while (data_received != 3) {
		usleep(500);
	}
	
	smlDsSessionGetSync(server_corr->dsession, _recv_sync, GINT_TO_POINTER(1));
	smlDsSessionGetChanges(server_corr->dsession, _recv_changes, GINT_TO_POINTER(1));
	smlDsSessionGetMapping(server_corr->dsession, _recv_map, GINT_TO_POINTER(1));
	
	while (syncs_received != 1 || adds_received != 1) {
		smlDsSessionDispatch(server_corr->dsession);
		usleep(500);
	}
	
	smlDsSessionSendSync(server_corr->dsession, 1, _sync_reply, GINT_TO_POINTER(1), &error);
	
	TEST_DATA_SYNC_CHANGE_ITEM_NEW(SML_CHANGE_ADD, "uid", "data", 4, SML_ELEMENT_TEXT_VCARD);
	sml_fail_unless(smlDsSessionQueueChange(server_corr->dsession, TEST_DATA_SYNC_CHANGE_ITEM, _add_reply, GINT_TO_POINTER(2), &error), NULL);
	TEST_DATA_SYNC_CHANGE_ITEM_FREE();
	sml_fail_unless(error == NULL, NULL);
	
	smlDsSessionCloseSync(server_corr->dsession, &error);
	
	sml_fail_unless(smlSessionFlush(session2, TRUE, &error), NULL);
	sml_fail_unless(error == NULL, NULL);
	
	while (data_received != 4) {
		usleep(500);
	}
	
	smlDsSessionGetSync(client_corr->dsession, _recv_sync, GINT_TO_POINTER(1));
	smlDsSessionGetChanges(client_corr->dsession, _recv_changes, GINT_TO_POINTER(1));
	
	while (syncs_received != 2 || adds_received != 2) {
		smlDsSessionDispatch(client_corr->dsession);
		usleep(500);
	}
	
	SmlMapItem *item = sml_map_item_new();
	SmlLocation *local = sml_location_new();
	sml_fail_unless(local != NULL, NULL);
	sml_location_set_uri(local, "uid");
	sml_map_item_set_local(item, local, &error);
	g_object_unref(local);
	local = NULL;
	SmlLocation *remote = sml_location_new();
	sml_fail_unless(remote != NULL, NULL);
	sml_location_set_uri(remote, "newuid");
	sml_map_item_set_remote(item, remote, &error);
	g_object_unref(remote);
	remote = NULL;
	smlDsSessionQueueMap(client_corr->dsession, item, &error);
	smlDsSessionCloseMap(client_corr->dsession, _map_reply, GINT_TO_POINTER(1), &error);
	
	sml_fail_unless(smlSessionFlush(session, TRUE, &error), NULL);
	sml_fail_unless(error == NULL, NULL);
	
	while (data_received != 5 || add_replies_received != 2 || mappings_received < 1) {
		smlDsSessionDispatch(server_corr->dsession);
		usleep(500);
	}
	
	sml_fail_unless(smlSessionEnd(session2, &error), NULL);
	sml_fail_unless(error == NULL, NULL);
	
	while (session_ends != 2) {
		usleep(500);
	}
	
	smlSessionStop(client_corr->thread);
	smlSessionStop(server_corr->thread);
	
	smlDsSessionUnref(client_corr->dsession);
	smlDsSessionUnref(server_corr->dsession);
	
	sml_fail_unless(data_received == 6, NULL);
	sml_fail_unless(init_alerts_received == 1, NULL);
	
	sml_fail_unless(alerts_received == 2, NULL);
	sml_fail_unless(syncs_received == 2, NULL);
	sml_fail_unless(changes_received == 0, NULL);
	sml_fail_unless(replies_received == 4, "4 expected but %d received", replies_received);
	sml_fail_unless(session_ends == 2, NULL);
	
	sml_fail_unless(adds_received == 2, NULL);
	sml_fail_unless(add_replies_received == 2, NULL);
	sml_fail_unless(deletes_received == 0, NULL);
	sml_fail_unless(delete_replies_received == 0, NULL);
	sml_fail_unless(modifies_received == 0, NULL);
	sml_fail_unless(modify_replies_received == 0, NULL);
	sml_fail_unless(mappings_received == 1, NULL);
	
	g_free(client_corr);
	g_free(server_corr);
	
	smlSessionUnref(session);
	smlSessionUnref(session2);
		
	g_object_unref(loc);
		
	smlDsServerFree(server);
	smlDsServerFree(client);
}
END_TEST

/* send alerts, send sync from server, send from client with one add */
/* C -- Put, Alert --> S
 * S -- Status, Put, Alert --> C
 * C -- Status, Sync, Delete --> S
 * S -- Status, Sync, Delete --> S
 * C -- Status --> S
 * S -- Status --> C */
START_TEST (datastore_delete)
{
	data_received = 0;
	init_alerts_received = 0;
	alerts_received = 0;
	syncs_received = 0;
	sync_ends_received = 0;
	changes_received = 0;
	replies_received = 0;
	adds_received = 0;
	add_replies_received = 0;
	deletes_received = 0;
	delete_replies_received = 0;
	modifies_received = 0;
	modify_replies_received = 0;
	session_ends = 0;
	
	to_session = NULL;
	to_session2 = NULL;
	
	/* General setup */
	setup_testbed(NULL);
	
	GError *error = NULL;
	SmlLocation *loc = sml_location_new();
	sml_fail_unless(loc != NULL, NULL);
	sml_location_set_uri(loc, "/vcards");
	
	
	SmlDsServer *client = smlDsClientNew(SML_ELEMENT_TEXT_VCARD, loc, loc, &error);
	_server_corr *client_corr = g_malloc0(sizeof(_server_corr));
	client_corr->server = client;
	
	
	SmlDsServer *server = smlDsServerNew(SML_ELEMENT_TEXT_VCARD, loc, &error);
	_server_corr *server_corr = g_malloc0(sizeof(_server_corr));
	server_corr->server = server;
	smlDsServerSetConnectCallback(server, _recv_init_alert, server_corr);
	
	
	/* Start a new session from the client */
	SmlSession *session = smlSessionNew(SML_SESSION_TYPE_CLIENT, SML_MIMETYPE_XML, SML_VERSION_12, SML_PROTOCOL_SYNCML, loc, loc, 1, 0, &error);
	client_corr->thread = smlSessionRunAsync(session, &to_session);
	SmlSession *session2 = smlSessionNew(SML_SESSION_TYPE_SERVER, SML_MIMETYPE_XML, SML_VERSION_12, SML_PROTOCOL_SYNCML, loc, loc, 1, 0, &error);
	server_corr->thread = smlSessionRunAsync(session2, &to_session2);
	smlSessionSetLocalMaxMsgSize(session2, defaultMaxMsgSize);
	smlSessionSetLocalMaxObjSize(session2, defaultMaxObjSize);
	
	smlSessionSetEventCallback(session, _event_callback, client_corr);
	smlSessionSetDataCallback(session, _data_send_callback, &to_session2);
	smlSessionSetEventCallback(session2, _event_callback, server_corr);
	smlSessionSetDataCallback(session2, _data_send_callback, &to_session);
	
	client_corr->dsession = smlDsServerSendAlert(client, session, SML_ALERT_TWO_WAY, "last", "next", _alert_reply, GINT_TO_POINTER(1), &error);
	sml_fail_unless(client_corr->dsession != NULL, NULL);
	sml_fail_unless(error == NULL, NULL);
	
	sml_fail_unless(smlSessionFlush(session, TRUE, &error), NULL);
	sml_fail_unless(error == NULL, NULL);
	
	while (init_alerts_received != 1) {
		usleep(500);
	}
	
	smlDsSessionGetAlert(server_corr->dsession, _recv_alert, GINT_TO_POINTER(1));
	smlDsSessionSendAlert(server_corr->dsession, SML_ALERT_TWO_WAY, "last", "next", _alert_reply, GINT_TO_POINTER(1), &error);
	
	
	
	while (alerts_received != 1) {
		smlDsSessionDispatch(server_corr->dsession);
		usleep(500);
	}
	
	sml_fail_unless(smlSessionFlush(session2, TRUE, &error), NULL);
	sml_fail_unless(error == NULL, NULL);
	
	while (data_received != 2) {
		usleep(500);
	}
	
	smlDsSessionGetAlert(client_corr->dsession, _recv_alert, GINT_TO_POINTER(1));
	
	
	
	while (alerts_received != 2) {
		smlDsSessionDispatch(client_corr->dsession);
		usleep(500);
	}
	
	smlDsSessionSendSync(client_corr->dsession, 1, _sync_reply, GINT_TO_POINTER(1), &error);
	
	TEST_DATA_SYNC_CHANGE_ITEM_NEW(SML_CHANGE_DELETE, "uid", NULL, 0, SML_ELEMENT_TEXT_VCARD);
	sml_fail_unless(smlDsSessionQueueChange(client_corr->dsession, TEST_DATA_SYNC_CHANGE_ITEM, _delete_reply, GINT_TO_POINTER(1), &error), NULL);
	TEST_DATA_SYNC_CHANGE_ITEM_FREE();
	sml_fail_unless(error == NULL, NULL);
	
	smlDsSessionCloseSync(client_corr->dsession, &error);
	
	sml_fail_unless(smlSessionFlush(session, TRUE, &error), NULL);
	sml_fail_unless(error == NULL, NULL);
	
	while (data_received != 3) {
		usleep(500);
	}
	
	smlDsSessionGetSync(server_corr->dsession, _recv_sync, GINT_TO_POINTER(1));
	smlDsSessionGetChanges(server_corr->dsession, _recv_changes, GINT_TO_POINTER(1));
	smlDsSessionGetMapping(server_corr->dsession, _recv_map, GINT_TO_POINTER(1));
	
	while (syncs_received != 1 || deletes_received != 1) {
		smlDsSessionDispatch(server_corr->dsession);
		usleep(500);
	}
	
	smlDsSessionSendSync(server_corr->dsession, 1, _sync_reply, GINT_TO_POINTER(1), &error);
	
	TEST_DATA_SYNC_CHANGE_ITEM_NEW(SML_CHANGE_DELETE, "uid", NULL, 0, SML_ELEMENT_TEXT_VCARD);
	sml_fail_unless(smlDsSessionQueueChange(server_corr->dsession, TEST_DATA_SYNC_CHANGE_ITEM, _delete_reply, GINT_TO_POINTER(1), &error), NULL);
	TEST_DATA_SYNC_CHANGE_ITEM_FREE();
	sml_fail_unless(error == NULL, NULL);
	
	smlDsSessionCloseSync(server_corr->dsession, &error);
	
	sml_fail_unless(smlSessionFlush(session2, TRUE, &error), NULL);
	sml_fail_unless(error == NULL, NULL);
	
	while (data_received != 4) {
		usleep(500);
	}
	
	smlDsSessionGetSync(client_corr->dsession, _recv_sync, GINT_TO_POINTER(1));
	smlDsSessionGetChanges(client_corr->dsession, _recv_changes, GINT_TO_POINTER(1));
	
	while (syncs_received != 2 || deletes_received != 2) {
		smlDsSessionDispatch(client_corr->dsession);
		usleep(500);
	}
	
	sml_fail_unless(smlSessionFlush(session, TRUE, &error), NULL);
	sml_fail_unless(error == NULL, NULL);
	
	while (data_received != 6 || delete_replies_received != 2 || session_ends != 2) {
		usleep(500);
	}
	
	smlSessionStop(client_corr->thread);
	smlSessionStop(server_corr->thread);
	
	smlDsSessionUnref(client_corr->dsession);
	smlDsSessionUnref(server_corr->dsession);
	
	sml_fail_unless(data_received == 6, NULL);
	sml_fail_unless(init_alerts_received == 1, NULL);
	
	sml_fail_unless(alerts_received == 2, NULL);
	sml_fail_unless(syncs_received == 2, NULL);
	sml_fail_unless(changes_received == 0, NULL);
	sml_fail_unless(replies_received == 4, NULL);
	sml_fail_unless(session_ends == 2, NULL);
	
	sml_fail_unless(adds_received == 0, NULL);
	sml_fail_unless(add_replies_received == 0, NULL);
	sml_fail_unless(deletes_received == 2, NULL);
	sml_fail_unless(delete_replies_received == 2, NULL);
	sml_fail_unless(modifies_received == 0, NULL);
	sml_fail_unless(modify_replies_received == 0, NULL);
	
	g_free(client_corr);
	g_free(server_corr);
	
	smlSessionUnref(session);
	smlSessionUnref(session2);
		
	g_object_unref(loc);
		
	smlDsServerFree(server);
	smlDsServerFree(client);
}
END_TEST

/* send alerts, send sync from client, send from server with one add */
/* Send map */
/* C -- Put, Alert --> S
 * S -- Status, Put, Alert --> C
 * C -- Status, Sync, Replace, Delete, Add --> S
 * S -- Status, Sync, Add, Replace, Delete --> S
 * C -- Status, Map --> S
 * S -- Status --> C */
START_TEST (datastore_map_full)
{
	data_received = 0;
	init_alerts_received = 0;
	alerts_received = 0;
	syncs_received = 0;
	sync_ends_received = 0;
	changes_received = 0;
	replies_received = 0;
	adds_received = 0;
	add_replies_received = 0;
	deletes_received = 0;
	delete_replies_received = 0;
	modifies_received = 0;
	modify_replies_received = 0;
	session_ends = 0;
	
	to_session = NULL;
	to_session2 = NULL;
	
	/* General setup */
	setup_testbed(NULL);
	
	GError *error = NULL;
	SmlLocation *loc = sml_location_new();
	sml_fail_unless(loc != NULL, NULL);
	sml_location_set_uri(loc, "/vcards");
	
	
	SmlDsServer *client = smlDsClientNew(SML_ELEMENT_TEXT_VCARD, loc, loc, &error);
	_server_corr *client_corr = g_malloc0(sizeof(_server_corr));
	client_corr->server = client;
	
	
	SmlDsServer *server = smlDsServerNew(SML_ELEMENT_TEXT_VCARD, loc, &error);
	_server_corr *server_corr = g_malloc0(sizeof(_server_corr));
	server_corr->server = server;
	smlDsServerSetConnectCallback(server, _recv_init_alert, server_corr);
	
	
	/* Start a new session from the client */
	SmlSession *session = smlSessionNew(SML_SESSION_TYPE_CLIENT, SML_MIMETYPE_XML, SML_VERSION_12, SML_PROTOCOL_SYNCML, loc, loc, 1, 0, &error);
	client_corr->thread = smlSessionRunAsync(session, &to_session);
	SmlSession *session2 = smlSessionNew(SML_SESSION_TYPE_SERVER, SML_MIMETYPE_XML, SML_VERSION_12, SML_PROTOCOL_SYNCML, loc, loc, 1, 0, &error);
	server_corr->thread = smlSessionRunAsync(session2, &to_session2);
	smlSessionSetLocalMaxMsgSize(session2, defaultMaxMsgSize);
	smlSessionSetLocalMaxObjSize(session2, defaultMaxObjSize);
	
	smlSessionSetEventCallback(session, _event_callback, client_corr);
	smlSessionSetDataCallback(session, _data_send_callback, &to_session2);
	smlSessionSetEventCallback(session2, _event_callback, server_corr);
	smlSessionSetDataCallback(session2, _data_send_callback, &to_session);
	
	client_corr->dsession = smlDsServerSendAlert(client, session, SML_ALERT_TWO_WAY, "last", "next", _alert_reply, GINT_TO_POINTER(1), &error);
	sml_fail_unless(client_corr->dsession != NULL, NULL);
	sml_fail_unless(error == NULL, NULL);
	
	sml_fail_unless(smlSessionFlush(session, TRUE, &error), NULL);
	sml_fail_unless(error == NULL, NULL);
	
	while (init_alerts_received != 1) {
		usleep(500);
	}
	
	smlDsSessionGetAlert(server_corr->dsession, _recv_alert, GINT_TO_POINTER(1));
	smlDsSessionSendAlert(server_corr->dsession, SML_ALERT_TWO_WAY, "last", "next", _alert_reply, GINT_TO_POINTER(1), &error);
	
	
	
	while (alerts_received != 1) {
		smlDsSessionDispatch(server_corr->dsession);
		usleep(500);
	}
	
	sml_fail_unless(smlSessionFlush(session2, TRUE, &error), NULL);
	sml_fail_unless(error == NULL, NULL);
	
	while (data_received != 2) {
		usleep(500);
	}
	
	smlDsSessionGetAlert(client_corr->dsession, _recv_alert, GINT_TO_POINTER(1));
	
	
	
	while (alerts_received != 2) {
		smlDsSessionDispatch(client_corr->dsession);
		usleep(500);
	}
	
	smlDsSessionSendSync(client_corr->dsession, 0, _sync_reply, GINT_TO_POINTER(1), &error);
	
	TEST_DATA_SYNC_CHANGE_ITEM_NEW(SML_CHANGE_REPLACE, "uid", "newdata", 7, SML_ELEMENT_TEXT_VCARD);
	sml_fail_unless(smlDsSessionQueueChange(client_corr->dsession, TEST_DATA_SYNC_CHANGE_ITEM, _modify_reply, GINT_TO_POINTER(1), &error), NULL);
	TEST_DATA_SYNC_CHANGE_ITEM_FREE();
	sml_fail_unless(error == NULL, NULL);
	
	TEST_DATA_SYNC_CHANGE_ITEM_NEW(SML_CHANGE_DELETE, "uid", NULL, 0, SML_ELEMENT_TEXT_VCARD);
	sml_fail_unless(smlDsSessionQueueChange(client_corr->dsession, TEST_DATA_SYNC_CHANGE_ITEM, _delete_reply, GINT_TO_POINTER(1), &error), NULL);
	TEST_DATA_SYNC_CHANGE_ITEM_FREE();
	sml_fail_unless(error == NULL, NULL);
	
	TEST_DATA_SYNC_CHANGE_ITEM_NEW(SML_CHANGE_ADD, "uid", "data", 4, SML_ELEMENT_TEXT_VCARD);
	sml_fail_unless(smlDsSessionQueueChange(client_corr->dsession, TEST_DATA_SYNC_CHANGE_ITEM, _add_reply, GINT_TO_POINTER(2), &error), NULL);
	TEST_DATA_SYNC_CHANGE_ITEM_FREE();
	sml_fail_unless(error == NULL, NULL);
	
	smlDsSessionCloseSync(client_corr->dsession, &error);
	
	sml_fail_unless(smlSessionFlush(session, TRUE, &error), NULL);
	sml_fail_unless(error == NULL, NULL);
	
	while (data_received != 3) {
		usleep(500);
	}
	
	smlDsSessionGetSync(server_corr->dsession, _recv_sync, GINT_TO_POINTER(1));
	smlDsSessionGetChanges(server_corr->dsession, _recv_changes, GINT_TO_POINTER(1));
	smlDsSessionGetMapping(server_corr->dsession, _recv_map, GINT_TO_POINTER(1));
	
	while (syncs_received != 1 || adds_received != 1 || deletes_received != 1 || modifies_received != 1) {
		smlDsSessionDispatch(server_corr->dsession);
		usleep(500);
	}
	
	smlDsSessionSendSync(server_corr->dsession, 0, _sync_reply, GINT_TO_POINTER(1), &error);
	
	TEST_DATA_SYNC_CHANGE_ITEM_NEW(SML_CHANGE_REPLACE, "uid", "newdata", 7, SML_ELEMENT_TEXT_VCARD);
	sml_fail_unless(smlDsSessionQueueChange(server_corr->dsession, TEST_DATA_SYNC_CHANGE_ITEM, _modify_reply, GINT_TO_POINTER(1), &error), NULL);
	TEST_DATA_SYNC_CHANGE_ITEM_FREE();
	sml_fail_unless(error == NULL, NULL);
	
	TEST_DATA_SYNC_CHANGE_ITEM_NEW(SML_CHANGE_ADD, "uid", "data", 4, SML_ELEMENT_TEXT_VCARD);
	sml_fail_unless(smlDsSessionQueueChange(server_corr->dsession, TEST_DATA_SYNC_CHANGE_ITEM, _add_reply, GINT_TO_POINTER(2), &error), NULL);
	TEST_DATA_SYNC_CHANGE_ITEM_FREE();
	sml_fail_unless(error == NULL, NULL);
	
	TEST_DATA_SYNC_CHANGE_ITEM_NEW(SML_CHANGE_DELETE, "uid", NULL, 0, SML_ELEMENT_TEXT_VCARD);
	sml_fail_unless(smlDsSessionQueueChange(server_corr->dsession, TEST_DATA_SYNC_CHANGE_ITEM, _delete_reply, GINT_TO_POINTER(1), &error), NULL);
	TEST_DATA_SYNC_CHANGE_ITEM_FREE();
	sml_fail_unless(error == NULL, NULL);
	
	smlDsSessionCloseSync(server_corr->dsession, &error);
	
	sml_fail_unless(smlSessionFlush(session2, TRUE, &error), NULL);
	sml_fail_unless(error == NULL, NULL);
	
	while (data_received != 4) {
		usleep(500);
	}
	
	smlDsSessionGetSync(client_corr->dsession, _recv_sync, GINT_TO_POINTER(1));
	smlDsSessionGetChanges(client_corr->dsession, _recv_changes, GINT_TO_POINTER(1));
	
	while (syncs_received != 2 || adds_received != 2 || deletes_received != 2 || modifies_received != 2) {
		smlDsSessionDispatch(client_corr->dsession);
		usleep(500);
	}
	
	SmlMapItem *item = sml_map_item_new();
	SmlLocation *local = sml_location_new();
	sml_fail_unless(local != NULL, NULL);
	sml_location_set_uri(local, "uid");
	sml_map_item_set_local(item, local, &error);
	g_object_unref(local);
	local = NULL;
	SmlLocation *remote = sml_location_new();
	sml_fail_unless(remote != NULL, NULL);
	sml_location_set_uri(remote, "newuid");
	sml_map_item_set_remote(item, remote, &error);
	g_object_unref(remote);
	remote = NULL;
	smlDsSessionQueueMap(client_corr->dsession, item, &error);
	smlDsSessionCloseMap(client_corr->dsession, _map_reply, GINT_TO_POINTER(1), &error);
	
	sml_fail_unless(smlSessionFlush(session, TRUE, &error), NULL);
	sml_fail_unless(error == NULL, NULL);
	
	while (data_received != 5 || add_replies_received != 2 || mappings_received < 1) {
		smlDsSessionDispatch(server_corr->dsession);
		usleep(500);
	}
	
	sml_fail_unless(smlSessionEnd(session2, &error), NULL);
	sml_fail_unless(error == NULL, NULL);
	
	while (session_ends != 2) {
		usleep(500);
	}
	
	smlSessionStop(client_corr->thread);
	smlSessionStop(server_corr->thread);
	
	smlDsSessionUnref(client_corr->dsession);
	smlDsSessionUnref(server_corr->dsession);
	
	sml_fail_unless(data_received == 6, NULL);
	sml_fail_unless(init_alerts_received == 1, NULL);
	
	sml_fail_unless(alerts_received == 2, NULL);
	sml_fail_unless(syncs_received == 2, NULL);
	sml_fail_unless(changes_received == 0, NULL);
	sml_fail_unless(replies_received == 4, NULL);
	sml_fail_unless(session_ends == 2, NULL);
	
	sml_fail_unless(adds_received == 2, NULL);
	sml_fail_unless(add_replies_received == 2, NULL);
	sml_fail_unless(deletes_received == 2, NULL);
	sml_fail_unless(delete_replies_received == 2, NULL);
	sml_fail_unless(modifies_received == 2, NULL);
	sml_fail_unless(modify_replies_received == 2, NULL);
	sml_fail_unless(mappings_received == 1, NULL);
	
	g_free(client_corr);
	g_free(server_corr);
	
	smlSessionUnref(session);
	smlSessionUnref(session2);
		
	g_object_unref(loc);
		
	smlDsServerFree(server);
	smlDsServerFree(client);
	
}
END_TEST

/* C -- Put, Alert --> S
 * S -- Status, Put, Alert --> C
 * C -- Status, Sync, 500 * Add --> S
 * S -- Status, Sync, 500 * Add --> S
 * C -- Status, Map --> S
 * S -- Status --> C */
START_TEST (datastore_add_large)
{
	data_received = 0;
	init_alerts_received = 0;
	alerts_received = 0;
	syncs_received = 0;
	sync_ends_received = 0;
	changes_received = 0;
	replies_received = 0;
	adds_received = 0;
	add_replies_received = 0;
	deletes_received = 0;
	delete_replies_received = 0;
	modifies_received = 0;
	modify_replies_received = 0;
	session_ends = 0;
	
	to_session = NULL;
	to_session2 = NULL;
	
	int num = 50;
	
	/* General setup */
	setup_testbed(NULL);
	
	GError *error = NULL;
	SmlLocation *loc = sml_location_new();
	sml_fail_unless(loc != NULL, NULL);
	sml_location_set_uri(loc, "/vcards");
	
	
	SmlDsServer *client = smlDsClientNew(SML_ELEMENT_TEXT_VCARD, loc, loc, &error);
	_server_corr *client_corr = g_malloc0(sizeof(_server_corr));
	client_corr->server = client;
	
	
	SmlDsServer *server = smlDsServerNew(SML_ELEMENT_TEXT_VCARD, loc, &error);
	_server_corr *server_corr = g_malloc0(sizeof(_server_corr));
	server_corr->server = server;
	smlDsServerSetConnectCallback(server, _recv_init_alert, server_corr);
	
	
	/* Start a new session from the client */
	SmlSession *session = smlSessionNew(SML_SESSION_TYPE_CLIENT, SML_MIMETYPE_XML, SML_VERSION_12, SML_PROTOCOL_SYNCML, loc, loc, 1, 0, &error);
	client_corr->thread = smlSessionRunAsync(session, &to_session);
	SmlSession *session2 = smlSessionNew(SML_SESSION_TYPE_SERVER, SML_MIMETYPE_XML, SML_VERSION_12, SML_PROTOCOL_SYNCML, loc, loc, 1, 0, &error);
	server_corr->thread = smlSessionRunAsync(session2, &to_session2);
	/* All stuff must be sent at once. */
	smlSessionSetLocalMaxMsgSize(session2, defaultMaxObjSize);
	smlSessionSetLocalMaxObjSize(session2, defaultMaxObjSize);
	
	smlSessionSetEventCallback(session, _event_callback, client_corr);
	smlSessionSetDataCallback(session, _data_send_callback, &to_session2);
	smlSessionSetEventCallback(session2, _event_callback, server_corr);
	smlSessionSetDataCallback(session2, _data_send_callback, &to_session);
	
	client_corr->dsession = smlDsServerSendAlert(client, session, SML_ALERT_TWO_WAY, "last", "next", _alert_reply, GINT_TO_POINTER(1), &error);
	sml_fail_unless(client_corr->dsession != NULL, NULL);
	sml_fail_unless(error == NULL, NULL);
	
	sml_fail_unless(smlSessionFlush(session, TRUE, &error), NULL);
	sml_fail_unless(error == NULL, NULL);
	
	while (init_alerts_received != 1) {
		usleep(500);
	}
	
	smlDsSessionGetAlert(server_corr->dsession, _recv_alert, GINT_TO_POINTER(1));
	smlDsSessionSendAlert(server_corr->dsession, SML_ALERT_TWO_WAY, "last", "next", _alert_reply, GINT_TO_POINTER(1), &error);
	
	while (alerts_received != 1) {
		smlDsSessionDispatch(server_corr->dsession);
		usleep(500);
	}
	
	sml_fail_unless(smlSessionFlush(session2, TRUE, &error), NULL);
	sml_fail_unless(error == NULL, NULL);
	
	while (data_received != 2) {
		usleep(500);
	}
	
	smlDsSessionGetAlert(client_corr->dsession, _recv_alert, GINT_TO_POINTER(1));
	
	while (alerts_received != 2) {
		smlDsSessionDispatch(client_corr->dsession);
		usleep(500);
	}
	
	smlDsSessionSendSync(client_corr->dsession, 0, _sync_reply, GINT_TO_POINTER(1), &error);

	int i = 0;
	for (i = 0; i < num; i++) {
		TEST_DATA_SYNC_CHANGE_ITEM_NEW(SML_CHANGE_ADD, "uid", "data", 4, SML_ELEMENT_TEXT_VCARD);
		sml_fail_unless(smlDsSessionQueueChange(client_corr->dsession, TEST_DATA_SYNC_CHANGE_ITEM, _add_reply, GINT_TO_POINTER(2), &error), NULL);
		TEST_DATA_SYNC_CHANGE_ITEM_FREE();
		sml_fail_unless(error == NULL, NULL);
	}
	
	smlDsSessionCloseSync(client_corr->dsession, &error);
	
	sml_fail_unless(smlSessionFlush(session, TRUE, &error), NULL);
	sml_fail_unless(error == NULL, NULL);
	
	while (data_received != 3) {
		usleep(500);
	}
	
	smlDsSessionGetSync(server_corr->dsession, _recv_sync, GINT_TO_POINTER(1));
	smlDsSessionGetChanges(server_corr->dsession, _recv_changes, GINT_TO_POINTER(1));
	smlDsSessionGetMapping(server_corr->dsession, _recv_map, GINT_TO_POINTER(1));
	
	while (syncs_received != 1 || adds_received != num) {
		smlDsSessionDispatch(server_corr->dsession);
		usleep(500);
	}
	
	smlDsSessionSendSync(server_corr->dsession, 0, _sync_reply, GINT_TO_POINTER(1), &error);
	
	for (i = 0; i < num; i++) {
		TEST_DATA_SYNC_CHANGE_ITEM_NEW(SML_CHANGE_ADD, "uid", "data", 4, SML_ELEMENT_TEXT_VCARD);
		sml_fail_unless(smlDsSessionQueueChange(server_corr->dsession, TEST_DATA_SYNC_CHANGE_ITEM, _add_reply, GINT_TO_POINTER(1), &error), NULL);
		TEST_DATA_SYNC_CHANGE_ITEM_FREE();
		sml_fail_unless(error == NULL, NULL);
	}
	
	smlDsSessionCloseSync(server_corr->dsession, &error);
	
	sml_fail_unless(smlSessionFlush(session2, TRUE, &error), NULL);
	sml_fail_unless(error == NULL, NULL);
	
	while (data_received != 4) {
		usleep(500);
	}
	
	smlDsSessionGetSync(client_corr->dsession, _recv_sync, GINT_TO_POINTER(1));
	smlDsSessionGetChanges(client_corr->dsession, _recv_changes, GINT_TO_POINTER(1));
	
	while (syncs_received != 2 || adds_received != num * 2) {
		smlDsSessionDispatch(client_corr->dsession);
		usleep(500);
	}
	
	for (i = 0; i < num; i++) {
		SmlMapItem *item = sml_map_item_new();
		SmlLocation *local = sml_location_new();
		sml_fail_unless(local != NULL, NULL);
		sml_location_set_uri(local, "uid");
		sml_map_item_set_local(item, local, &error);
		g_object_unref(local);
		local = NULL;
		SmlLocation *remote = sml_location_new();
		sml_fail_unless(remote != NULL, NULL);
		sml_location_set_uri(remote, "newuid");
		sml_map_item_set_remote(item, remote, &error);
		g_object_unref(remote);
		remote = NULL;
		smlDsSessionQueueMap(client_corr->dsession, item, &error);
	}
	smlDsSessionCloseMap(client_corr->dsession, _map_reply, GINT_TO_POINTER(1), &error);
	
	sml_fail_unless(smlSessionFlush(session, TRUE, &error), NULL);
	sml_fail_unless(error == NULL, NULL);
	
	while (data_received != 5 || add_replies_received != num * 2 || mappings_received < num) {
		smlDsSessionDispatch(server_corr->dsession);
		usleep(500);
	}
	
	sml_fail_unless(smlSessionEnd(session2, &error), NULL);
	sml_fail_unless(error == NULL, NULL);
	
	while (session_ends != 2) {
		usleep(500);
	}
	
	smlSessionStop(client_corr->thread);
	smlSessionStop(server_corr->thread);
	
	smlDsSessionUnref(client_corr->dsession);
	smlDsSessionUnref(server_corr->dsession);
	
	sml_fail_unless(data_received == 6, NULL);
	sml_fail_unless(init_alerts_received == 1, NULL);
	
	sml_fail_unless(alerts_received == 2, NULL);
	sml_fail_unless(syncs_received == 2, NULL);
	sml_fail_unless(changes_received == 0, NULL);
	sml_fail_unless(replies_received == 4, NULL);
	sml_fail_unless(session_ends == 2, NULL);
	
	sml_fail_unless(adds_received == num * 2, NULL);
	sml_fail_unless(add_replies_received == num * 2, NULL);
	sml_fail_unless(deletes_received == 0, NULL);
	sml_fail_unless(delete_replies_received == 0, NULL);
	sml_fail_unless(modifies_received == 0, NULL);
	sml_fail_unless(modify_replies_received == 0, NULL);
	sml_fail_unless(mappings_received == num, NULL);
	
	g_free(client_corr);
	g_free(server_corr);
	
	smlSessionUnref(session);
	smlSessionUnref(session2);
		
	g_object_unref(loc);
		
	smlDsServerFree(server);
	smlDsServerFree(client);
}
END_TEST

/* C -- Put, Alert --> S
 * S -- Status, Put, Alert --> C
 * C -- Status, Sync, 500 * Add --> S
 * S -- Status, Sync, 500 * Add --> S
 * C -- Status, Map --> S
 * S -- Status --> C */
/* Limit the sending size of both sides */
START_TEST (datastore_limit)
{
	data_received = 0;
	init_alerts_received = 0;
	alerts_received = 0;
	syncs_received = 0;
	sync_ends_received = 0;
	changes_received = 0;
	replies_received = 0;
	adds_received = 0;
	add_replies_received = 0;
	deletes_received = 0;
	delete_replies_received = 0;
	modifies_received = 0;
	modify_replies_received = 0;
	session_ends = 0;
	
	to_session = NULL;
	to_session2 = NULL;
	
	int num = 5;
	unsigned int limit = 1300;
	
	/* General setup */
	setup_testbed(NULL);
	
	GError *error = NULL;
	SmlLocation *loc = sml_location_new();
	sml_fail_unless(loc != NULL, NULL);
	sml_location_set_uri(loc, "/vcards");
	
	
	SmlDsServer *client = smlDsClientNew(SML_ELEMENT_TEXT_VCARD, loc, loc, &error);
	_server_corr *client_corr = g_malloc0(sizeof(_server_corr));
	client_corr->server = client;
	
	
	SmlDsServer *server = smlDsServerNew(SML_ELEMENT_TEXT_VCARD, loc, &error);
	_server_corr *server_corr = g_malloc0(sizeof(_server_corr));
	server_corr->server = server;
	smlDsServerSetConnectCallback(server, _recv_init_alert, server_corr);
	
	
	/* Start a new session from the client */
	SmlSession *session = smlSessionNew(SML_SESSION_TYPE_CLIENT, SML_MIMETYPE_XML, SML_VERSION_12, SML_PROTOCOL_SYNCML, loc, loc, 1, 0, &error);
	client_corr->thread = smlSessionRunAsync(session, &to_session);
	smlSessionSetRemoteMaxMsgSize(session, limit);
	SmlSession *session2 = smlSessionNew(SML_SESSION_TYPE_SERVER, SML_MIMETYPE_XML, SML_VERSION_12, SML_PROTOCOL_SYNCML, loc, loc, 1, 0, &error);
	server_corr->thread = smlSessionRunAsync(session2, &to_session2);
	smlSessionSetRemoteMaxMsgSize(session2, limit);
	smlSessionSetLocalMaxMsgSize(session2, defaultMaxMsgSize);
	smlSessionSetLocalMaxObjSize(session2, defaultMaxObjSize);
	
	smlSessionSetEventCallback(session, _event_callback, client_corr);
	smlSessionSetDataCallback(session, _data_send_callback, &to_session2);
	smlSessionSetEventCallback(session2, _event_callback, server_corr);
	smlSessionSetDataCallback(session2, _data_send_callback, &to_session);
	
	client_corr->dsession = smlDsServerSendAlert(client, session, SML_ALERT_TWO_WAY, "last", "next", _alert_reply, GINT_TO_POINTER(1), &error);
	sml_fail_unless(client_corr->dsession != NULL, NULL);
	sml_fail_unless(error == NULL, NULL);
	
	sml_fail_unless(smlSessionFlush(session, TRUE, &error), NULL);
	sml_fail_unless(error == NULL, NULL);
	
	while (init_alerts_received != 1) {
		usleep(500);
	}
	
	smlDsSessionGetAlert(server_corr->dsession, _recv_alert, GINT_TO_POINTER(1));
	smlDsSessionSendAlert(server_corr->dsession, SML_ALERT_TWO_WAY, "last", "next", _alert_reply, GINT_TO_POINTER(1), &error);
	
	
	
	while (alerts_received != 1) {
		smlDsSessionDispatch(server_corr->dsession);
		usleep(500);
	}
	
	sml_fail_unless(smlSessionFlush(session2, TRUE, &error), NULL);
	sml_fail_unless(error == NULL, NULL);
	
	smlDsSessionGetAlert(client_corr->dsession, _recv_alert, GINT_TO_POINTER(1));
	
	
	while (alerts_received != 2) {
		smlDsSessionDispatch(client_corr->dsession);
		usleep(500);
	}
	
	smlDsSessionSendSync(client_corr->dsession, 0, _sync_reply, GINT_TO_POINTER(1), &error);

	int i = 0;
	for (i = 0; i < num; i++) {
		TEST_DATA_SYNC_CHANGE_ITEM_NEW(SML_CHANGE_ADD, "uid", "data", 4, SML_ELEMENT_TEXT_VCARD);
		sml_fail_unless(smlDsSessionQueueChange(client_corr->dsession, TEST_DATA_SYNC_CHANGE_ITEM, _add_reply, GINT_TO_POINTER(2), &error), NULL);
		TEST_DATA_SYNC_CHANGE_ITEM_FREE();
		sml_fail_unless(error == NULL, NULL);
	}
	
	smlDsSessionCloseSync(client_corr->dsession, &error);
	
	sml_fail_unless(smlSessionFlush(session, TRUE, &error), NULL);
	sml_fail_unless(error == NULL, NULL);
	
	
	smlDsSessionGetSync(server_corr->dsession, _recv_sync, GINT_TO_POINTER(1));
	smlDsSessionGetChanges(server_corr->dsession, _recv_changes, GINT_TO_POINTER(1));
	smlDsSessionGetMapping(server_corr->dsession, _recv_map, GINT_TO_POINTER(1));
	
	while (syncs_received != 1 || adds_received != num) {
		smlDsSessionDispatch(server_corr->dsession);
		usleep(5000);
	}
	
	smlDsSessionSendSync(server_corr->dsession, 0, _sync_reply, GINT_TO_POINTER(1), &error);
	
	for (i = 0; i < num; i++) {
		TEST_DATA_SYNC_CHANGE_ITEM_NEW(SML_CHANGE_ADD, "uid", "data", 4, SML_ELEMENT_TEXT_VCARD);
		sml_fail_unless(smlDsSessionQueueChange(server_corr->dsession, TEST_DATA_SYNC_CHANGE_ITEM, _add_reply, GINT_TO_POINTER(1), &error), NULL);
		TEST_DATA_SYNC_CHANGE_ITEM_FREE();
		sml_fail_unless(error == NULL, NULL);
	}
	
	smlDsSessionCloseSync(server_corr->dsession, &error);
	
	sml_fail_unless(smlSessionFlush(session2, TRUE, &error), NULL);
	sml_fail_unless(error == NULL, NULL);
	
	
	smlDsSessionGetSync(client_corr->dsession, _recv_sync, GINT_TO_POINTER(1));
	smlDsSessionGetChanges(client_corr->dsession, _recv_changes, GINT_TO_POINTER(1));
	
	while (adds_received != num * 2) {
		smlDsSessionDispatch(client_corr->dsession);
		usleep(5000);
	}
	
	for (i = 0; i < num; i++) {
		SmlMapItem *item = sml_map_item_new();
		SmlLocation *local = sml_location_new();
		sml_fail_unless(local != NULL, NULL);
		sml_location_set_uri(local, "uid");
		sml_map_item_set_local(item, local, &error);
		g_object_unref(local);
		local = NULL;
		SmlLocation *remote = sml_location_new();
		sml_fail_unless(remote != NULL, NULL);
		sml_location_set_uri(remote, "newuid");
		sml_map_item_set_remote(item, remote, &error);
		g_object_unref(remote);
		remote = NULL;
		smlDsSessionQueueMap(client_corr->dsession, item, &error);
	}
	smlDsSessionCloseMap(client_corr->dsession, _map_reply, GINT_TO_POINTER(1), &error);
	
	sml_fail_unless(smlSessionFlush(session, TRUE, &error), NULL);
	sml_fail_unless(error == NULL, NULL);
	
	while (add_replies_received != num * 2 || mappings_received < num) {
		smlDsSessionDispatch(server_corr->dsession);
		usleep(500);
	}
	
	sml_fail_unless(smlSessionEnd(session2, &error), NULL);
	sml_fail_unless(error == NULL, NULL);
	
	while (session_ends != 2) {
		usleep(500);
	}
	
	g_object_unref(loc);
	
	smlSessionStop(client_corr->thread);
	smlSessionStop(server_corr->thread);
	
	smlDsSessionUnref(client_corr->dsession);
	smlDsSessionUnref(server_corr->dsession);
	
	//sml_fail_unless(data_received == 6, NULL);
	sml_fail_unless(init_alerts_received == 1, NULL);
	
	sml_fail_unless(alerts_received == 2, NULL);
	sml_fail_unless(syncs_received == 2, NULL);
	sml_fail_unless(changes_received == 0, NULL);
	sml_fail_unless(replies_received == 4, NULL);
	sml_fail_unless(session_ends == 2, NULL);
	
	sml_fail_unless(adds_received == num * 2, NULL);
	sml_fail_unless(add_replies_received == num * 2, NULL);
	sml_fail_unless(deletes_received == 0, NULL);
	sml_fail_unless(delete_replies_received == 0, NULL);
	sml_fail_unless(modifies_received == 0, NULL);
	sml_fail_unless(modify_replies_received == 0, NULL);
	sml_fail_unless(mappings_received == num, NULL);
	
	g_free(client_corr);
	g_free(server_corr);
	
	smlSessionUnref(session);
	smlSessionUnref(session2);
		
	smlDsServerFree(server);
	smlDsServerFree(client);
}
END_TEST

#if 0
/* C -- Put, Alert --> S
 * S -- Status, Put, Alert --> C
 * C -- Status, Sync, 500 * Add --> S
 * S -- Status, Sync, 500 * Add --> S
 * C -- Status, Map --> S
 * S -- Status --> C */
/* Limit the sending size of both sides */
START_TEST (datastore_multi_server)
{
	data_received = 0;
	init_alerts_received = 0;
	alerts_received = 0;
	syncs_received = 0;
	sync_ends_received = 0;
	changes_received = 0;
	replies_received = 0;
	adds_received = 0;
	add_replies_received = 0;
	deletes_received = 0;
	delete_replies_received = 0;
	modifies_received = 0;
	modify_replies_received = 0;
	session_ends = 0;
	
	to_session = NULL;
	to_session2 = NULL;
	
	int num = 50;
	unsigned int limit = 2000;
	
	/* General setup */
	setup_testbed(NULL);
	
	GError *gerror = NULL;
	GError *error = NULL;
	SmlLocation *loc = sml_location_new_with_options("test", NULL, &gerror);
	sml_fail_unless(loc != NULL, NULL);
	sml_fail_unless(gerror == NULL, NULL);
	
	SmlLocation *loc1 = sml_location_new_with_options("vcards", NULL, &gerror);
	sml_fail_unless(loc1 != NULL, NULL);
	sml_fail_unless(gerror == NULL, NULL);
	SmlLocation *loc2 = sml_location_new_with_options("vcals", NULL, &gerror);
	sml_fail_unless(loc2 != NULL, NULL);
	sml_fail_unless(gerror == NULL, NULL);
	SmlLocation *loc3 = sml_location_new_with_options("vnotes", NULL, &gerror);
	sml_fail_unless(loc3 != NULL, NULL);
	sml_fail_unless(gerror == NULL, NULL);
	
	/* The ds clients */
	SmlDsServer *client1 = smlDsClientNew(SML_ELEMENT_TEXT_VCARD, loc1, loc1, &error);
	_server_corr *client_corr1 = g_malloc0(sizeof(_server_corr));
	client_corr1->server = client1;
	
	SmlDsServer *client2 = smlDsClientNew(SML_ELEMENT_TEXT_VCARD, loc2, loc2, &error);
	_server_corr *client_corr2 = g_malloc0(sizeof(_server_corr));
	client_corr2->server = client2;
	
	SmlDsServer *client3 = smlDsClientNew(SML_ELEMENT_TEXT_VCARD, loc3, loc3, &error);
	_server_corr *client_corr3 = g_malloc0(sizeof(_server_corr));
	client_corr3->server = client3;
	
	/* The ds servers */
	SmlDsServer *server1 = smlDsServerNew(SML_ELEMENT_TEXT_VCARD, loc1, &error);
	_server_corr *server_corr1 = g_malloc0(sizeof(_server_corr));
	server_corr1->server = server1;
	smlDsServerSetConnectCallback(server1, _recv_init_alert, server_corr1);
	
	SmlDsServer *server2 = smlDsServerNew(SML_ELEMENT_TEXT_VCARD, loc2, &error);
	_server_corr *server_corr2 = g_malloc0(sizeof(_server_corr));
	server_corr2->server = server2;
	smlDsServerSetConnectCallback(server2, _recv_init_alert, server_corr2);
	
	SmlDsServer *server3 = smlDsServerNew(SML_ELEMENT_TEXT_VCARD, loc3, &error);
	_server_corr *server_corr3 = g_malloc0(sizeof(_server_corr));
	server_corr3->server = server3;
	smlDsServerSetConnectCallback(server3, _recv_init_alert, server_corr3);
	
	
	/* Start a new session from the client */
	SmlSession *session = smlSessionNew(SML_SESSION_TYPE_CLIENT, SML_MIMETYPE_XML, SML_VERSION_12, SML_PROTOCOL_SYNCML, loc, loc, 1, 0, &error);
	client_corr1->thread = smlSessionRunAsync(session, &to_session);
	smlSessionSetSendingMaxMsgSize(session, limit);
	SmlSession *session2 = smlSessionNew(SML_SESSION_TYPE_SERVER, SML_MIMETYPE_XML, SML_VERSION_12, SML_PROTOCOL_SYNCML, loc, loc, 1, 0, &error);
	server_corr->thread = smlSessionRunAsync(session2, &to_session2);
	smlSessionSetSendingMaxMsgSize(session2, limit);
	
	smlSessionSetEventCallback(session, _event_callback, client_corr);
	smlSessionSetDataCallback(session, _data_send_callback, &to_session2);
	smlSessionSetEventCallback(session2, _event_callback, server_corr);
	smlSessionSetDataCallback(session2, _data_send_callback, &to_session);
	
	client_corr->dsession = smlDsServerSendAlert(client, session, SML_ALERT_TWO_WAY, "last", "next", _alert_reply, GINT_TO_POINTER(1), &error);
	sml_fail_unless(client_corr->dsession != NULL, NULL);
	sml_fail_unless(error == NULL, NULL);
	
	sml_fail_unless(smlSessionFlush(session, TRUE, &error), NULL);
	sml_fail_unless(error == NULL, NULL);
	
	while (init_alerts_received != 1) {
		usleep(500);
	}
	
	smlDsSessionGetAlert(server_corr1->dsession, _recv_alert, GINT_TO_POINTER(1));
	smlDsSessionSendAlert(server_corr1->dsession, SML_ALERT_TWO_WAY, "last", "next", _alert_reply, GINT_TO_POINTER(1), &error);
	
	smlDsSessionGetAlert(server_corr2->dsession, _recv_alert, GINT_TO_POINTER(1));
	smlDsSessionSendAlert(server_corr2->dsession, SML_ALERT_TWO_WAY, "last", "next", _alert_reply, GINT_TO_POINTER(1), &error);
	
	smlDsSessionGetAlert(server_corr3->dsession, _recv_alert, GINT_TO_POINTER(1));
	smlDsSessionSendAlert(server_corr4->dsession, SML_ALERT_TWO_WAY, "last", "next", _alert_reply, GINT_TO_POINTER(1), &error);
	
	
	
	while (alerts_received != 3) {
		smlDsSessionDispatch(server_corr1->dsession);
		smlDsSessionDispatch(server_corr2->dsession);
		smlDsSessionDispatch(server_corr3->dsession);
		usleep(500);
	}
	
	sml_fail_unless(smlSessionFlush(session2, TRUE, &error), NULL);
	sml_fail_unless(error == NULL, NULL);
	
	smlDsSessionGetAlert(client_corr1->dsession, _recv_alert, GINT_TO_POINTER(1));
	smlDsSessionGetAlert(client_corr2->dsession, _recv_alert, GINT_TO_POINTER(1));
	smlDsSessionGetAlert(client_corr3->dsession, _recv_alert, GINT_TO_POINTER(1));
	
	
	while (alerts_received != 6) {
		smlDsSessionDispatch(client_corr1->dsession);
		smlDsSessionDispatch(client_corr2->dsession);
		smlDsSessionDispatch(client_corr3->dsession);
		usleep(500);
	}
	
	smlDsSessionSendSync(client_corr1->dsession, 0, _sync_reply, GINT_TO_POINTER(1), &error);
	int i = 0;
	for (i = 0; i < num; i++) {
		TEST_DATA_SYNC_CHANGE_ITEM_NEW(SML_CHANGE_ADD, "uid", "data", 4, SML_ELEMENT_TEXT_VCARD);
		sml_fail_unless(smlDsSessionQueueChange(client_corr1->dsession, TEST_DATA_SYNC_CHANGE_ITEM, _add_reply, GINT_TO_POINTER(2), &error), NULL);
		TEST_DATA_SYNC_CHANGE_ITEM_FREE();
		sml_fail_unless(error == NULL, NULL);
	}
	smlDsSessionCloseSync(client_corr1->dsession, &error);
	
	smlDsSessionSendSync(client_corr2->dsession, 0, _sync_reply, GINT_TO_POINTER(1), &error);
	for (i = 0; i < num; i++) {
		TEST_DATA_SYNC_CHANGE_ITEM_NEW(SML_CHANGE_ADD, "uid", "data", 4, SML_ELEMENT_TEXT_VCARD);
		sml_fail_unless(smlDsSessionQueueChange(client_corr2->dsession, TEST_DATA_SYNC_CHANGE_ITEM, _add_reply, GINT_TO_POINTER(2), &error), NULL);
		TEST_DATA_SYNC_CHANGE_ITEM_FREE();
		sml_fail_unless(error == NULL, NULL);
	}
	smlDsSessionCloseSync(client_corr2->dsession, &error);
	
	smlDsSessionSendSync(client_corr3->dsession, 0, _sync_reply, GINT_TO_POINTER(1), &error);
	for (i = 0; i < num; i++) {
		TEST_DATA_SYNC_CHANGE_ITEM_NEW(SML_CHANGE_ADD, "uid", "data", 4, SML_ELEMENT_TEXT_VCARD);
		sml_fail_unless(smlDsSessionQueueChange(client_corr3->dsession, TEST_DATA_SYNC_CHANGE_ITEM, _add_reply, GINT_TO_POINTER(2), &error), NULL);
		TEST_DATA_SYNC_CHANGE_ITEM_FREE();
		sml_fail_unless(error == NULL, NULL);
	}
	smlDsSessionCloseSync(client_corr3->dsession, &error);
	
	sml_fail_unless(smlSessionFlush(session, TRUE, &error), NULL);
	sml_fail_unless(error == NULL, NULL);
	
	
	smlDsSessionGetSync(server_corr1->dsession, _recv_sync, GINT_TO_POINTER(1));
	smlDsSessionGetChanges(server_corr1->dsession, _recv_changes, GINT_TO_POINTER(1));
	smlDsSessionGetMapping(server_corr1->dsession, _recv_map, GINT_TO_POINTER(1));
	smlDsSessionGetSync(server_corr2->dsession, _recv_sync, GINT_TO_POINTER(1));
	smlDsSessionGetChanges(server_corr2->dsession, _recv_changes, GINT_TO_POINTER(1));
	smlDsSessionGetMapping(server_corr2->dsession, _recv_map, GINT_TO_POINTER(1));
	smlDsSessionGetSync(server_corr3->dsession, _recv_sync, GINT_TO_POINTER(1));
	smlDsSessionGetChanges(server_corr3->dsession, _recv_changes, GINT_TO_POINTER(1));
	smlDsSessionGetMapping(server_corr3->dsession, _recv_map, GINT_TO_POINTER(1));
	
	while (syncs_received != 3 || adds_received != num * 3) {
		smlDsSessionDispatch(server_corr1->dsession);
		smlDsSessionDispatch(server_corr2->dsession);
		smlDsSessionDispatch(server_corr3->dsession);
		usleep(5000);
	}
	
	smlDsSessionSendSync(server_corr1->dsession, 0, _sync_reply, GINT_TO_POINTER(1), &error);
	for (i = 0; i < num; i++) {
		TEST_DATA_SYNC_CHANGE_ITEM_NEW(SML_CHANGE_ADD, "uid", "data", 4, SML_ELEMENT_TEXT_VCARD);
		sml_fail_unless(smlDsSessionQueueChange(server_corr1->dsession, TEST_DATA_SYNC_CHANGE_ITEM, _add_reply, GINT_TO_POINTER(1), &error), NULL);
		TEST_DATA_SYNC_CHANGE_ITEM_FREE();
		sml_fail_unless(error == NULL, NULL);
	}
	smlDsSessionCloseSync(server_corr1->dsession, &error);
	
	smlDsSessionSendSync(server_corr2->dsession, 0, _sync_reply, GINT_TO_POINTER(1), &error);
	for (i = 0; i < num; i++) {
		TEST_DATA_SYNC_CHANGE_ITEM_NEW(SML_CHANGE_ADD, "uid", "data", 4, SML_ELEMENT_TEXT_VCARD);
		sml_fail_unless(smlDsSessionQueueChange(server_corr2->dsession, TEST_DATA_SYNC_CHANGE_ITEM, _add_reply, GINT_TO_POINTER(1), &error), NULL);
		TEST_DATA_SYNC_CHANGE_ITEM_FREE();
		sml_fail_unless(error == NULL, NULL);
	}
	smlDsSessionCloseSync(server_corr2->dsession, &error);
	
	smlDsSessionSendSync(server_corr3->dsession, 0, _sync_reply, GINT_TO_POINTER(1), &error);
	for (i = 0; i < num; i++) {
		TEST_DATA_SYNC_CHANGE_ITEM_NEW(SML_CHANGE_ADD, "uid", "data", 4, SML_ELEMENT_TEXT_VCARD);
		sml_fail_unless(smlDsSessionQueueChange(server_corr3->dsession, TEST_DATA_SYNC_CHANGE_ITEM, _add_reply, GINT_TO_POINTER(1), &error), NULL);
		TEST_DATA_SYNC_CHANGE_ITEM_FREE();
		sml_fail_unless(error == NULL, NULL);
	}
	smlDsSessionCloseSync(server_corr3->dsession, &error);
	
	sml_fail_unless(smlSessionFlush(session2, TRUE, &error), NULL);
	sml_fail_unless(error == NULL, NULL);
	
	
	smlDsSessionGetSync(client_corr1->dsession, _recv_sync, GINT_TO_POINTER(1));
	smlDsSessionGetChanges(client_corr1->dsession, _recv_changes, GINT_TO_POINTER(1));
	smlDsSessionGetSync(client_corr2->dsession, _recv_sync, GINT_TO_POINTER(1));
	smlDsSessionGetChanges(client_corr2->dsession, _recv_changes, GINT_TO_POINTER(1));
	smlDsSessionGetSync(client_corr3->dsession, _recv_sync, GINT_TO_POINTER(1));
	smlDsSessionGetChanges(client_corr3->dsession, _recv_changes, GINT_TO_POINTER(1));
	
	while (adds_received != num * 2 * 3) {
		smlDsSessionDispatch(client_corr1->dsession);
		smlDsSessionDispatch(client_corr2->dsession);
		smlDsSessionDispatch(client_corr3->dsession);
		usleep(5000);
	}
	
	for (i = 0; i < num; i++) {
		smlDsSessionQueueMap(client_corr1->dsession, "uid", "newuid", &error);
	}
	smlDsSessionCloseMap(client_corr1->dsession, _map_reply, GINT_TO_POINTER(1), &error);
	
	for (i = 0; i < num; i++) {
		smlDsSessionQueueMap(client_corr2->dsession, "uid", "newuid", &error);
	}
	smlDsSessionCloseMap(client_corr2->dsession, _map_reply, GINT_TO_POINTER(1), &error);
	
	for (i = 0; i < num; i++) {
		smlDsSessionQueueMap(client_corr3->dsession, "uid", "newuid", &error);
	}
	smlDsSessionCloseMap(client_corr3->dsession, _map_reply, GINT_TO_POINTER(1), &error);
	
	sml_fail_unless(smlSessionFlush(session, TRUE, &error), NULL);
	sml_fail_unless(error == NULL, NULL);
	
	while (add_replies_received != num * 2) {
		usleep(500);
	}
	
	sml_fail_unless(smlSessionEnd(session2, &error), NULL);
	sml_fail_unless(error == NULL, NULL);
	
	while (session_ends != 2) {
		usleep(500);
	}
	
	smlSessionStop(client_corr1->thread);
	smlSessionStop(client_corr2->thread);
	smlSessionStop(client_corr3->thread);
	smlSessionStop(server_corr1->thread);
	smlSessionStop(server_corr2->thread);
	smlSessionStop(server_corr3->thread);
	
	smlDsSessionUnref(client_corr1->dsession);
	smlDsSessionUnref(client_corr2->dsession);
	smlDsSessionUnref(client_corr3->dsession);
	smlDsSessionUnref(server_corr1->dsession);
	smlDsSessionUnref(server_corr2->dsession);
	smlDsSessionUnref(server_corr3->dsession);
	
	//sml_fail_unless(data_received == 6, NULL);
	sml_fail_unless(init_alerts_received == 1, NULL);
	
	sml_fail_unless(alerts_received == 6, NULL);
	sml_fail_unless(syncs_received == 6, NULL);
	sml_fail_unless(changes_received == 0, NULL);
	sml_fail_unless(replies_received == 15, NULL);
	sml_fail_unless(session_ends == 2, NULL);
	
	sml_fail_unless(adds_received == num * 2 * 3, NULL);
	sml_fail_unless(add_replies_received == num * 2 * 3, NULL);
	sml_fail_unless(deletes_received == 0, NULL);
	sml_fail_unless(delete_replies_received == 0, NULL);
	sml_fail_unless(modifies_received == 0, NULL);
	sml_fail_unless(modify_replies_received == 0, NULL);
	
	g_free(client_corr1);
	g_free(client_corr2);
	g_free(client_corr3);
	g_free(server_corr1);
	g_free(server_corr2);
	g_free(server_corr3);
	
	smlSessionUnref(session);
	smlSessionUnref(session2);
		
	smlDsServerFree(server1);
	smlDsServerFree(server2);
	smlDsServerFree(server3);
	smlDsServerFree(client1);
	smlDsServerFree(client2);
	smlDsServerFree(client3);
}
END_TEST
#endif

	/* To test:
	 * - Multiple ds servers
	 * - error behavious
	 * - Slow-sync overwrite
	 * - Aborting
	 * - lots of changes with limit
	 * - large objs with both limits
	 * - missing map
	 * - size too small for one object
	 * - SAN
	 * - SAN 1.2
	 */
	
@SML_TESTCASE_CODE@