opensync-commits — Mailing list that will receive the commit information

[Opensync-commits] r899 - trunk/libsyncml

[<dg...@su...>]

Author: bellmich
Date: Tue Jan 27 16:39:11 2009
New Revision: 899
URL: http://libsyncml.opensync.org/changeset/899

Log:
If an error reference is set once then this error
reference cannot be used for other function calls.

Modified:
   trunk/libsyncml/sml_manager.c

Modified: trunk/libsyncml/sml_manager.c
==============================================================================
--- trunk/libsyncml/sml_manager.c	Tue Jan 27 16:28:34 2009	(r898)
+++ trunk/libsyncml/sml_manager.c	Tue Jan 27 16:39:11 2009	(r899)
@@ -1231,8 +1231,6 @@
 		
 		object->childCallback(session, cmd, object->commandCallbackUserdata);
 	} else {
-		smlErrorSet(error, SML_ERROR_GENERIC, "Unable to find child command handler");
-		
 		SmlStatus *reply = smlCommandNewReply(cmd, SML_ERROR_NOT_FOUND, error);
 		if (!reply)
 			goto error;
@@ -1244,6 +1242,8 @@
 		
 		smlStatusUnref(reply);
 		
+		smlErrorSet(error, SML_ERROR_GENERIC, "Unable to find child command handler");
+		
 		goto error;
 	}
 	

[Opensync-commits] r898 - trunk/libsyncml/data_sync_api

[<dg...@su...>]

Author: bellmich
Date: Tue Jan 27 16:28:34 2009
New Revision: 898
URL: http://libsyncml.opensync.org/changeset/898

Log:
A link requires some link data. Otherwise the link is useless.

Modified:
   trunk/libsyncml/data_sync_api/transport_http_client.c

Modified: trunk/libsyncml/data_sync_api/transport_http_client.c
==============================================================================
--- trunk/libsyncml/data_sync_api/transport_http_client.c	Tue Jan 27 10:30:06 2009	(r897)
+++ trunk/libsyncml/data_sync_api/transport_http_client.c	Tue Jan 27 16:28:34 2009	(r898)
@@ -78,7 +78,6 @@
 	char *sessionString = smlManagerGetNewSessionID(dsObject->manager);
 	SmlLocation *target = smlLocationNew(dsObject->url, NULL, error);
 	SmlLocation *source = smlLocationNew(dsObject->identifier, NULL, error);
-	SmlLink *link = smlLinkNew(dsObject->tsp, NULL, error);
 	dsObject->session = smlSessionNew(SML_SESSION_TYPE_CLIENT,
 					SML_MIMETYPE_XML,
 					dsObject->version,
@@ -99,7 +98,7 @@
 	smlSessionRef(dsObject->session);
 
 	/* register all the add-ons */
-	if (!smlManagerSessionAdd(dsObject->manager, dsObject->session, link, error))
+	if (!smlManagerSessionAdd(dsObject->manager, dsObject->session, NULL, error))
 		goto error;
 	if (!smlDevInfAgentRegisterSession(dsObject->agent, dsObject->manager, dsObject->session, error))
         	goto error;

[Opensync-commits] r897 - trunk/libsyncml/data_sync_api

[<dg...@su...>]

Author: bellmich
Date: Tue Jan 27 10:30:06 2009
New Revision: 897
URL: http://libsyncml.opensync.org/changeset/897

Log:
enum changes are critical if the order changes.
Hardcoded values make the header files and shared
libraries robust against new elements.

Modified:
   trunk/libsyncml/data_sync_api/standard.h

Modified: trunk/libsyncml/data_sync_api/standard.h
==============================================================================
--- trunk/libsyncml/data_sync_api/standard.h	Tue Jan 20 11:16:53 2009	(r896)
+++ trunk/libsyncml/data_sync_api/standard.h	Tue Jan 27 10:30:06 2009	(r897)
@@ -38,13 +38,13 @@
 /*! @brief These are all possible events which the high level API sends.
  */
 typedef enum SmlDataSyncEventType {
-	SML_DATA_SYNC_EVENT_ERROR,
-	SML_DATA_SYNC_EVENT_CONNECT,
-	SML_DATA_SYNC_EVENT_GOT_ALL_ALERTS,
-	SML_DATA_SYNC_EVENT_GOT_ALL_CHANGES,
-	SML_DATA_SYNC_EVENT_GOT_ALL_MAPPINGS,
-	SML_DATA_SYNC_EVENT_DISCONNECT,
-	SML_DATA_SYNC_EVENT_FINISHED,
+	SML_DATA_SYNC_EVENT_ERROR = 0,
+	SML_DATA_SYNC_EVENT_CONNECT = 1,
+	SML_DATA_SYNC_EVENT_GOT_ALL_ALERTS = 2,
+	SML_DATA_SYNC_EVENT_GOT_ALL_CHANGES = 3,
+	SML_DATA_SYNC_EVENT_GOT_ALL_MAPPINGS = 4,
+	SML_DATA_SYNC_EVENT_DISCONNECT = 5,
+	SML_DATA_SYNC_EVENT_FINISHED = 6,
 } SmlDataSyncEventType;
 
 typedef struct SmlDataSyncObject SmlDataSyncObject;

[Opensync-commits] r5234 - in branches/timeout: . opensync/ipc tests tests/ipc-tests

[<dg...@su...>]

Author: Graham Cobb
Date: Mon Jan 26 01:28:59 2009
New Revision: 5234
URL: http://www.opensync.org/changeset/5234

Log:
Done: 4) Handle timeout before reply queue set up by generating an
error message on the command queue

Unit test (ipc_timeout_noreplyq) added

Note: this changes behaviour: when a queue is disconnected, all pending 
commands (i.e. unreplied, but with a message handler) will be replied with
an error response.  The message handler callbacks will all be called before 
the call to osync_queue_disconnect returns.

Still to do:

6) Implement a mechanism to protect the queue itself: i.e. is the
receiver still reading the queue?

Modified:
   branches/timeout/ChangeLog
   branches/timeout/opensync/ipc/opensync_queue.c
   branches/timeout/opensync/ipc/opensync_queue_private.h
   branches/timeout/tests/CMakeLists.txt
   branches/timeout/tests/ipc-tests/check_ipc.c

Modified: branches/timeout/ChangeLog
==============================================================================
--- branches/timeout/ChangeLog	Mon Jan 26 01:23:47 2009	(r5233)
+++ branches/timeout/ChangeLog	Mon Jan 26 01:28:59 2009	(r5234)
@@ -1,5 +1,32 @@
+2009-01-25  Graham Cobb  <g+...@co...>
+
+	* opensync/engine/opensync_sink_engine.c (osync_sink_engine_new): Ref objengine [Bug #1052]
+	(osync_sink_engine_unref): Unref objengine [Bug #1052]
+
+	* opensync/engine/opensync_obj_engine.c (_osync_obj_engine_*_callback): Unref sinkengine [Bug #1052]
+	(osync_obj_engine_command): Ref sinkengine every time it is used for a callback [Bug #1052]
+
+	* opensync/ipc/opensync_queue.c (osync_queue_disconnect): Empty pending
+	queue before performing disconnect.  Pending messages with callbacks
+	will get called with an error message.
+	(_incoming_check): Do not action incoming queue if a disconnect is in progress.
+	This avoids entries being added to pending queue while we are trying to empty it.
+	(osync_queue_send_message_with_timeout): Do not allow sending messages which require
+	adding entries to the reply queue pending list if the reply queue is being disconnected.
+
+	* opensync/ipc/opensync_queue_private.h: Add disc_in_progress flag to queue.
+
 2009-01-24  Graham Cobb  <g+...@co...>
 
+	* opensync/ipc/opensync_queue.c (_osync_queue_generate_error): Add 
+	_osync_queue_generate_error
+	(_osync_send_timeout_response): If there is no reply queue, 
+	call _osync_queue_generate_error
+
+	* tests/CMakeLists.txt: Add ipc_timeout_noreplyq
+
+	* tests/ipc-tests/check_ipc.c: Add ipc_timeout_noreplyq
+
 	* opensync/ipc/opensync_queue.c (osync_queue_remove_cross_link): Add
 	osync_queue_remove_cross_link
 	(osync_queue_disconnect): Call osync_queue_remove_cross_link

Modified: branches/timeout/opensync/ipc/opensync_queue.c
==============================================================================
--- branches/timeout/opensync/ipc/opensync_queue.c	Mon Jan 26 01:23:47 2009	(r5233)
+++ branches/timeout/opensync/ipc/opensync_queue.c	Mon Jan 26 01:28:59 2009	(r5234)
@@ -175,12 +175,37 @@
 
 	if (g_async_queue_length(queue->incoming) > 0
 	    && (queue->pendingLimit == 0 
-		|| queue->pendingCount < queue->pendingLimit) )
+		|| queue->pendingCount < queue->pendingLimit) 
+	    && !queue->disc_in_progress )
 		return TRUE;
 	
 	return FALSE;
 }
 
+static gboolean _osync_queue_generate_error(OSyncQueue *queue, OSyncMessageCommand errcode, OSyncError **error)
+{
+	OSyncMessage *message;
+
+	queue->connected = FALSE;
+	
+	/* Now we can send the hup message, and wake up the consumer thread so
+	 * it can pickup the messages in the incoming queue */
+	message = osync_message_new(errcode, 0, error);
+	if (!message) {
+		return FALSE;
+	}
+	osync_trace(TRACE_INTERNAL, "Generating incoming error message %p(%s), id= %lli", message, osync_message_get_commandstr(message), osync_message_get_id(message));
+	
+	osync_message_ref(message);
+
+	g_async_queue_push(queue->incoming, message);
+	
+	if (queue->incomingContext)
+		g_main_context_wakeup(queue->incomingContext);
+
+	return TRUE;
+}
+
 static void _osync_send_timeout_response(OSyncMessage *errormsg, void *user_data)
 {
 	OSyncQueue *queue = user_data;
@@ -191,9 +216,13 @@
 		osync_queue_send_message_with_timeout(queue->reply_queue, NULL, errormsg, 0, NULL);
 	} else {
 		osync_message_unref(errormsg);
-		/* TODO: as we can't send the timeout response, disconnect the queue to signal to
-		   the sender that they aren't going to get a response.  Note: we can't just
-		   call osync_queue_disconnect as that tries to kill the thread and hits a deadlock! */
+		/* As we can't send the timeout response, create an error and drop it in the incoming queue
+		   so the higher layer will disconnect. Note: we can't just call osync_queue_disconnect 
+		   as that tries to kill the thread and hits a deadlock! */
+		if (!_osync_queue_generate_error(queue, OSYNC_MESSAGE_QUEUE_ERROR, NULL)) {
+			osync_trace(TRACE_EXIT_ERROR, "%s: cannot even generate error on incoming queue", __func__);
+			return;
+		}
 		osync_trace(TRACE_EXIT_ERROR, "%s: cannot find reply queue to send timeout error", __func__);
 		return;
 	}
@@ -570,18 +599,8 @@
 		return TRUE;
 	case OSYNC_QUEUE_EVENT_HUP:
 	case OSYNC_QUEUE_EVENT_ERROR:
-		queue->connected = FALSE;
-			
-		/* Now we can send the hup message, and wake up the consumer thread so
-		 * it can pickup the messages in the incoming queue */
-		message = osync_message_new(OSYNC_MESSAGE_QUEUE_HUP, 0, &error);
-		if (!message)
+		if (!_osync_queue_generate_error(queue, OSYNC_MESSAGE_QUEUE_HUP, &error))
 			goto error;
-			
-		g_async_queue_push(queue->incoming, message);
-			
-		if (queue->incomingContext)
-			g_main_context_wakeup(queue->incomingContext);
 		return FALSE;
 	}
 	
@@ -1000,6 +1019,52 @@
 	osync_trace(TRACE_ENTRY, "%s(%p, %p)", __func__, queue, error);
 	osync_assert(queue);
 
+	/* Before doing the real disconnect, we empty the pending queue by creating HUP
+	   errors for anything on it. In order to make sure it empties, the disconnect 
+	   is marked as in progress so that no more entries get put on it (e.g. if a 
+	   callback tries to send another message). */
+
+	g_mutex_lock(queue->pendingLock);
+
+	queue->disc_in_progress = TRUE;
+
+	while (queue->pendingCount > 0) {
+		OSyncPendingMessage *pending = queue->pendingReplies->data;
+		OSyncError *error = NULL;
+		OSyncError *huperr = NULL;
+		OSyncMessage *errormsg = NULL;
+		
+		queue->pendingReplies = osync_list_remove(queue->pendingReplies, pending);
+		queue->pendingCount--;
+
+		/* Call the callback of the pending message */
+		if (pending->callback) {
+			osync_error_set(&huperr, OSYNC_ERROR_IO_ERROR, "Disconnect.");
+			errormsg = osync_message_new_errorreply(NULL, huperr, &error);
+			osync_error_unref(&huperr);
+			osync_message_set_id(errormsg, pending->id);
+			
+			/* Unlock the pending lock during the callback */
+			g_mutex_unlock(queue->pendingLock);
+			
+			osync_trace(TRACE_INTERNAL, "%s: Reporting disconnect error for message %lli", __func__, pending->id);
+
+			pending->callback(errormsg, pending->user_data);
+			if (errormsg != NULL)
+				osync_message_unref(errormsg);
+		
+			/* Lock again */
+			g_mutex_lock(queue->pendingLock);
+		}
+		
+		// TODO: Refcounting for OSyncPendingMessage
+		if (pending->timeout_info)
+			g_free(pending->timeout_info);
+
+		osync_free(pending);
+	}
+	g_mutex_unlock(queue->pendingLock);
+
 	osync_queue_remove_cross_link(queue);
 
 	g_mutex_lock(queue->disconnectLock);
@@ -1038,6 +1103,10 @@
 	queue->fd = -1;
 	queue->connected = FALSE;
 	g_mutex_unlock(queue->disconnectLock);
+
+	g_mutex_lock(queue->pendingLock);
+	queue->disc_in_progress = FALSE;
+	g_mutex_unlock(queue->pendingLock);
 	
 	osync_trace(TRACE_EXIT, "%s", __func__);
 	return TRUE;
@@ -1221,6 +1290,13 @@
 		GTimeVal current_time;
 		long long int id = 0;
 		osync_assert(replyqueue);
+
+		g_mutex_lock(replyqueue->pendingLock);
+		if (replyqueue->disc_in_progress) {
+			osync_error_set(error, OSYNC_ERROR_IO_ERROR, "Disconnect in progress.");
+			goto error;
+		}
+
 		pending = osync_try_malloc0(sizeof(OSyncPendingMessage), error);
 		if (!pending)
 			goto error;
@@ -1243,7 +1319,6 @@
 		pending->callback = osync_message_get_handler(message);
 		pending->user_data = osync_message_get_handler_data(message);
 		
-		g_mutex_lock(replyqueue->pendingLock);
 		replyqueue->pendingReplies = osync_list_append(replyqueue->pendingReplies, pending);
 		replyqueue->pendingCount++;
 		g_mutex_unlock(replyqueue->pendingLock);

Modified: branches/timeout/opensync/ipc/opensync_queue_private.h
==============================================================================
--- branches/timeout/opensync/ipc/opensync_queue_private.h	Mon Jan 26 01:23:47 2009	(r5233)
+++ branches/timeout/opensync/ipc/opensync_queue_private.h	Mon Jan 26 01:28:59 2009	(r5234)
@@ -96,6 +96,9 @@
 
 	/* Queue for receiving commands we are replying to */
 	OSyncQueue *cmd_queue;
+
+	/* Disconnect in progress */
+	gboolean disc_in_progress;
 };
 
 

Modified: branches/timeout/tests/CMakeLists.txt
==============================================================================
--- branches/timeout/tests/CMakeLists.txt	Mon Jan 26 01:23:47 2009	(r5233)
+++ branches/timeout/tests/CMakeLists.txt	Mon Jan 26 01:28:59 2009	(r5234)
@@ -249,6 +249,7 @@
 OSYNC_TESTCASE(ipc ipc_loop_with_timeout)
 OSYNC_TESTCASE(ipc ipc_late_reply)
 OSYNC_TESTCASE(ipc ipc_loop_timeout_with_idle)
+OSYNC_TESTCASE(ipc ipc_timeout_noreplyq)
 ENDIF (NOT WIN32)
 
 BUILD_CHECK_TEST( mapping mapping-tests/check_mapping.c ${TEST_TARGET_LIBRARIES} )

Modified: branches/timeout/tests/ipc-tests/check_ipc.c
==============================================================================
--- branches/timeout/tests/ipc-tests/check_ipc.c	Mon Jan 26 01:23:47 2009	(r5233)
+++ branches/timeout/tests/ipc-tests/check_ipc.c	Mon Jan 26 01:28:59 2009	(r5234)
@@ -2089,10 +2089,13 @@
 
 static void _message_handler(OSyncMessage *message, void *user_data)
 {
+	osync_trace(TRACE_ENTRY, "%s(%p, %p)", __func__, message, user_data);
+	osync_trace(TRACE_INTERNAL, "%s",osync_message_get_commandstr(message));
 	if (osync_message_is_error(message))
 		num_callback_timeout++;
 	else
 		num_callback++;
+	osync_trace(TRACE_EXIT, "%s", __func__);
 }
 
 char *data5 = "this is another test string";
@@ -2803,6 +2806,185 @@
 }
 END_TEST
 
+void client_handler6(OSyncMessage *message, void *user_data)
+{
+	osync_trace(TRACE_ENTRY, "%s(%p, %p)", __func__, message, user_data);
+	OSyncError *error = NULL;
+	
+	osync_assert(GPOINTER_TO_INT(user_data) ==1);
+
+	if (osync_message_get_command(message) == OSYNC_MESSAGE_QUEUE_ERROR) {
+		osync_queue_disconnect(client_queue, NULL);
+		osync_trace(TRACE_EXIT, "%s: disconnect", __func__);
+		return;
+	}
+
+	osync_assert(osync_message_get_command(message) == OSYNC_MESSAGE_INITIALIZE);
+
+	int int1;
+	long long int longint1;
+	char *string;
+	char databuf[strlen(data5) + 1];
+	
+	osync_message_read_int(message, &int1);
+	osync_message_read_string(message, &string);
+	osync_message_read_long_long_int(message, &longint1);
+	osync_message_read_data(message, databuf, strlen(data5) + 1);
+	
+	osync_assert(int1 == 4000000);
+	osync_assert(!strcmp(string, "this is a test string"));
+	osync_assert(longint1 == 400000000);
+	osync_assert(!strcmp(databuf, data5));
+	
+	/* TIMEOUT TIMEOUT TIMEOUT (no reply...) */
+	
+	/* Proper code would reply to this message, but for testing
+	   purposes we don't reply and simulate a "timeout" situation */
+		
+	osync_trace(TRACE_EXIT, "%s", __func__);
+}
+START_TEST (ipc_timeout_noreplyq)
+{	
+	/* This testcase is inteded to test timeout before the command and reply queues are cross-linked.
+	   Client got forked and listens for messages from Server and replies.
+
+	   To simulate a "timeout" situation the Client doesn't reply to one of the Server messages.
+
+	   As there is no reply queue, an error will be sent to the **client**, who then disconnects
+	   so an error (although not a timeout) ends up sent to the server.
+	   */
+
+	char *testbed = setup_testbed(NULL);
+	osync_testing_file_remove("/tmp/testpipe-server");
+	osync_testing_file_remove("/tmp/testpipe-client");
+
+	num_callback_timeout = 0;
+	num_callback = 0;
+	
+	OSyncError *error = NULL;
+	server_queue = osync_queue_new("/tmp/testpipe-server", &error);
+	client_queue = osync_queue_new("/tmp/testpipe-client", &error);
+	OSyncMessage *message = NULL;
+	
+	osync_queue_create(server_queue, &error);
+	fail_unless(error == NULL, NULL);
+	
+	osync_queue_create(client_queue, &error);
+	fail_unless(error == NULL, NULL);
+	
+	pid_t cpid = fork();
+	if (cpid == 0) { //Child
+		
+		GMainContext *context = g_main_context_new();
+		OSyncThread *thread = osync_thread_new(context, &error);
+	
+		osync_queue_set_message_handler(client_queue, client_handler6, GINT_TO_POINTER(1));
+		osync_queue_set_pending_limit(client_queue, OSYNC_QUEUE_PENDING_LIMIT);
+		
+		osync_queue_setup_with_gmainloop(client_queue, context);
+		
+		osync_thread_start(thread);
+
+		osync_assert(osync_queue_connect(client_queue, OSYNC_QUEUE_RECEIVER, &error));
+		osync_assert(error == NULL);
+
+		osync_assert(osync_queue_connect(server_queue, OSYNC_QUEUE_SENDER, &error));
+		osync_assert(error == NULL);
+
+		/* Do not cross-link */
+		/*osync_queue_cross_link(client_queue, server_queue);*/
+		
+		message = osync_queue_get_message(server_queue);
+		
+		osync_assert(osync_message_get_command(message) == OSYNC_MESSAGE_QUEUE_HUP);
+		
+		osync_message_unref(message);
+	
+		if (osync_queue_disconnect(server_queue, &error) != TRUE || error != NULL)
+			exit(1);
+		osync_queue_unref(server_queue);
+		
+		osync_assert(osync_queue_disconnect(client_queue, &error));
+		osync_assert(error == NULL);
+		
+		osync_thread_stop(thread);
+		osync_thread_unref(thread);
+		
+		osync_queue_unref(client_queue);
+		
+		g_free(testbed);
+		
+		exit(0);
+	} else {
+		GMainContext *context = g_main_context_new();
+		OSyncThread *thread = osync_thread_new(context, &error);
+		
+		osync_queue_set_message_handler(server_queue, server_handler4, GINT_TO_POINTER(1));
+		
+		osync_queue_setup_with_gmainloop(server_queue, context);
+		
+		osync_thread_start(thread);
+
+		fail_unless(osync_queue_connect(client_queue, OSYNC_QUEUE_SENDER, &error), NULL);
+		fail_unless(error == NULL, NULL);
+		
+		fail_unless(osync_queue_connect(server_queue, OSYNC_QUEUE_RECEIVER, &error), NULL);
+		fail_unless(error == NULL, NULL);
+		
+		message = osync_message_new(OSYNC_MESSAGE_INITIALIZE, 0, &error);
+		fail_unless(message != NULL, NULL);
+		fail_unless(!osync_error_is_set(&error), NULL);
+
+		osync_message_set_handler(message, _message_handler, NULL);
+		
+		osync_message_write_int(message, 4000000);
+		osync_message_write_string(message, "this is a test string");
+		osync_message_write_long_long_int(message, 400000000);
+		osync_message_write_data(message, data5, strlen(data5) + 1);
+		
+		// Send with timeout of one second
+		fail_unless(osync_queue_send_message_with_timeout(client_queue, server_queue, message, 1, &error), NULL);
+		fail_unless(!osync_error_is_set(&error), NULL);
+
+		osync_message_unref(message);
+		
+		while (!(message = osync_queue_get_message(client_queue))) {
+			g_usleep(10000);
+		}
+		
+		fail_unless(osync_message_get_command(message) == OSYNC_MESSAGE_QUEUE_HUP);
+		osync_message_unref(message);
+		
+		osync_queue_disconnect(client_queue, &error);
+		fail_unless(error == NULL, NULL);
+		
+		osync_queue_disconnect(server_queue, &error);
+		fail_unless(error == NULL, NULL);
+		
+		int status = 0;
+		wait(&status);
+		fail_unless(WEXITSTATUS(status) == 0, NULL);
+	}
+	
+	fail_unless(osync_testing_file_exists("/tmp/testpipe-client") == TRUE, NULL);
+	
+	fail_unless(osync_queue_remove(client_queue, &error), NULL);
+	fail_unless(osync_queue_remove(server_queue, &error), NULL);
+	fail_unless(!osync_error_is_set(&error), NULL);
+
+	/* Check if the timeout handler replied with an error */
+	fail_unless(num_callback_timeout == 1, NULL);
+	fail_unless(num_callback == 0, NULL);
+	
+	fail_unless(osync_testing_file_exists("/tmp/testpipe-client") == FALSE, NULL);
+
+	osync_queue_unref(client_queue);
+	osync_queue_unref(server_queue);
+	
+	destroy_testbed(testbed);
+}
+END_TEST
+
 OSYNC_TESTCASE_START("ipc")
 OSYNC_TESTCASE_ADD(ipc_new)
 OSYNC_TESTCASE_ADD(ipc_ref)
@@ -2832,5 +3014,6 @@
 OSYNC_TESTCASE_ADD(ipc_late_reply)
 OSYNC_TESTCASE_ADD(ipc_loop_with_timeout)
 OSYNC_TESTCASE_ADD(ipc_loop_timeout_with_idle)
+OSYNC_TESTCASE_ADD(ipc_timeout_noreplyq)
 OSYNC_TESTCASE_END
 

[Opensync-commits] r5233 - branches/timeout/opensync/engine

[<dg...@su...>]

Author: Graham Cobb
Date: Mon Jan 26 01:23:47 2009
New Revision: 5233
URL: http://www.opensync.org/changeset/5233

Log:
Bug #1052: add refs in obj_engine and sink_engine

Modified:
   branches/timeout/opensync/engine/opensync_obj_engine.c
   branches/timeout/opensync/engine/opensync_sink_engine.c

Modified: branches/timeout/opensync/engine/opensync_obj_engine.c
==============================================================================
--- branches/timeout/opensync/engine/opensync_obj_engine.c	Sun Jan 25 14:47:52 2009	(r5232)
+++ branches/timeout/opensync/engine/opensync_obj_engine.c	Mon Jan 26 01:23:47 2009	(r5233)
@@ -118,6 +118,7 @@
 	} else
 		osync_trace(TRACE_INTERNAL, "Not yet: %i", osync_bitcount(engine->sink_errors | engine->sink_connects));
 	
+	osync_sink_engine_unref(sinkengine);
 	osync_trace(TRACE_EXIT, "%s", __func__);
 }
 
@@ -148,6 +149,7 @@
 	} else
 		osync_trace(TRACE_INTERNAL, "Not yet: %i", osync_bitcount(engine->sink_errors | engine->sink_connect_done));
 	
+	osync_sink_engine_unref(sinkengine);
 	osync_trace(TRACE_EXIT, "%s", __func__);
 }
 
@@ -191,6 +193,8 @@
 	
 	_osync_obj_engine_generate_event_disconnected(engine, error);
 
+	osync_sink_engine_unref(sinkengine);
+
 	osync_trace(TRACE_EXIT, "%s", __func__);
 }
 
@@ -322,9 +326,12 @@
 
 static void _osync_obj_engine_read_ignored_callback(OSyncClientProxy *proxy, void *userdata, OSyncError *error)
 {
+	OSyncSinkEngine *sinkengine = userdata;
+	OSyncObjEngine *engine = sinkengine->engine;
 	/* TODO: Share _generate_read_event fucntion with _osync_obj_engine_read_callback?
 	   To report errors .. and handle _timeout problems of _read_ignored call.
 	*/
+	osync_sink_engine_unref(sinkengine);
 }
 
 
@@ -356,6 +363,7 @@
 	} else
 		osync_trace(TRACE_INTERNAL, "Not yet: %i", osync_bitcount(engine->sink_errors | engine->sink_get_changes));
 	
+	osync_sink_engine_unref(sinkengine);
 	osync_trace(TRACE_EXIT, "%s", __func__);
 }
 
@@ -586,6 +594,7 @@
 	} else
 		osync_trace(TRACE_INTERNAL, "Not yet: %i", osync_bitcount(engine->sink_errors | engine->sink_sync_done));
 	
+	osync_sink_engine_unref(sinkengine);
 	osync_trace(TRACE_EXIT, "%s", __func__);
 }
 
@@ -935,6 +944,7 @@
 
 			if (!osync_client_proxy_connect(sinkengine->proxy, _osync_obj_engine_connect_callback, sinkengine, engine->objtype, engine->slowsync, error))
 				goto error;
+			osync_sink_engine_ref(sinkengine); // Note that connect callback has a reference
 		}
 		break;
 	case OSYNC_ENGINE_COMMAND_CONNECT_DONE:
@@ -946,6 +956,7 @@
 
 			if (!osync_client_proxy_connect_done(sinkengine->proxy, _osync_obj_engine_connect_done_callback, sinkengine, engine->objtype, error))
 				goto error;
+			osync_sink_engine_ref(sinkengine); // Note that connect_done callback has a reference
 		}
 		break;
 	case OSYNC_ENGINE_COMMAND_READ:
@@ -964,6 +975,7 @@
 
 				if (!osync_client_proxy_read(sinkengine->proxy, _osync_obj_engine_read_ignored_callback, sinkengine, change, error))
 					goto error;
+				osync_sink_engine_ref(sinkengine); // Note that read_ignored callback has a reference
 			}
 		}
 
@@ -998,6 +1010,7 @@
 
 			if (!osync_client_proxy_get_changes(sinkengine->proxy, _osync_obj_engine_read_callback, sinkengine, engine->objtype, engine->slowsync, error))
 				goto error;
+			osync_sink_engine_ref(sinkengine); // Note that read callback has a reference
 		}
 
 		break;
@@ -1107,6 +1120,7 @@
 
 			if (!osync_client_proxy_sync_done(sinkengine->proxy, _osync_obj_engine_sync_done_callback, sinkengine, engine->objtype, error))
 				goto error;
+			osync_sink_engine_ref(sinkengine); // Note that sync_done callback has a reference
 		}
 		break;
 	case OSYNC_ENGINE_COMMAND_DISCONNECT:;
@@ -1126,6 +1140,7 @@
 
 			if (!osync_client_proxy_disconnect(sinkengine->proxy, _osync_obj_engine_disconnect_callback, sinkengine, engine->objtype, error))
 				goto error;
+			osync_sink_engine_ref(sinkengine); // Note that disconnect callback has a reference
 		}
 			
 		/* If no client needs to be disconnected, we MUST NOT expected any 

Modified: branches/timeout/opensync/engine/opensync_sink_engine.c
==============================================================================
--- branches/timeout/opensync/engine/opensync_sink_engine.c	Sun Jan 25 14:47:52 2009	(r5232)
+++ branches/timeout/opensync/engine/opensync_sink_engine.c	Mon Jan 26 01:23:47 2009	(r5233)
@@ -55,6 +55,7 @@
 	sinkengine->proxy = proxy;
 	
 	sinkengine->engine = objengine;
+	osync_obj_engine_ref(objengine);
 	
 	osync_trace(TRACE_EXIT, "%s: %p", __func__, sinkengine);
 	return sinkengine;
@@ -92,6 +93,8 @@
 			engine->entries = osync_list_remove(engine->entries, engine->entries->data);
 		}
 		
+		osync_obj_engine_unref(engine->engine);
+
 		osync_free(engine);
 	}
 }

[Opensync-commits] r5232 - plugins/opie-sync/src

[<dg...@su...>]

Author: paule
Date: Sun Jan 25 14:47:52 2009
New Revision: 5232
URL: http://www.opensync.org/changeset/5232

Log:
Date created XMLField for events is called Created not DateCreated

Modified:
   plugins/opie-sync/src/opie_format.c

Modified: plugins/opie-sync/src/opie_format.c
==============================================================================
--- plugins/opie-sync/src/opie_format.c	Sun Jan 25 14:20:51 2009	(r5231)
+++ plugins/opie-sync/src/opie_format.c	Sun Jan 25 14:47:52 2009	(r5232)
@@ -929,7 +929,7 @@
 				{
 					time_t createtime = (time_t)atoi(iprop->children->content);
 					char *createvtime = osync_time_unix2vtime(&createtime);
-					out_xmlfield = osync_xmlfield_new(out_xmlformat, "DateCreated", error);
+					out_xmlfield = osync_xmlfield_new(out_xmlformat, "Created", error);
 					osync_xmlfield_set_key_value(out_xmlfield, "Content", iprop->children->content);
 					g_free(createvtime);
 				}
@@ -1086,7 +1086,7 @@
 		else if(!strcmp("Location", fieldname)) {
 			xmlfield_key_to_attr(in_xmlfield, "Content", on_event, "location");
 		}
-		else if(!strcmp("DateCreated", fieldname)) {
+		else if(!strcmp("Created", fieldname)) {
 			xmlfield_vtime_to_attr_time_t(in_xmlfield, on_event, "created");
 		}
 		else if(!strcmp("DateStarted", fieldname)) {

[Opensync-commits] r5231 - plugins/opie-sync/src

[<dg...@su...>]

Author: paule
Date: Sun Jan 25 14:20:51 2009
New Revision: 5231
URL: http://www.opensync.org/changeset/5231

Log:
Use BlogUrl for HomeWebPage and Url for BusinessWebPage in contacts (previously Url was linked to HomeWebPage and BusinessWebPage got discarded)

Modified:
   plugins/opie-sync/src/opie_format.c

Modified: plugins/opie-sync/src/opie_format.c
==============================================================================
--- plugins/opie-sync/src/opie_format.c	Sun Jan 25 14:00:26 2009	(r5230)
+++ plugins/opie-sync/src/opie_format.c	Sun Jan 25 14:20:51 2009	(r5231)
@@ -158,12 +158,13 @@
 				}
 				else if(!strcasecmp(iprop->name, "HomeWebPage"))
 				{
-					out_xmlfield = osync_xmlfield_new(out_xmlformat, "Url", error);
+					out_xmlfield = osync_xmlfield_new(out_xmlformat, "BlogUrl", error);
 					osync_xmlfield_set_key_value(out_xmlfield, "Content", iprop->children->content);
 				}
 				else if(!strcasecmp(iprop->name, "BusinessWebPage"))
 				{
-					/* FIXME handle this field */
+					out_xmlfield = osync_xmlfield_new(out_xmlformat, "Url", error);
+					osync_xmlfield_set_key_value(out_xmlfield, "Content", iprop->children->content);
 				}
 				else if(!strcasecmp(iprop->name, "Spouse"))
 				{
@@ -472,6 +473,9 @@
 			xmlfield_key_to_attr(in_xmlfield, "Content", on_contact, "Anniversary");
 		}
 		else if(!strcmp("Url", fieldname)) {
+			xmlfield_key_to_attr(in_xmlfield, "Content", on_contact, "BusinessWebPage");
+		}
+		else if(!strcmp("BlogUrl", fieldname)) {
 			xmlfield_key_to_attr(in_xmlfield, "Content", on_contact, "HomeWebPage");
 		}
 		else if(!strcmp("FormattedName", fieldname)) {
@@ -493,7 +497,6 @@
 // TODO: Entries to be handled
 /*   unsigned int rid; */
 /*   unsigned int rinfo; */
-/*   char* business_webpage; */
 /*   int gender; */
 /*   char* children; */
 /*   GList* anons; */

[Opensync-commits] r5230 - plugins/opie-sync/src

[<dg...@su...>]

Author: paule
Date: Sun Jan 25 14:00:26 2009
New Revision: 5230
URL: http://www.opensync.org/changeset/5230

Log:
Add explicit check that a UID could be retrieved, in case of entries on the device written by other software without one (!)

Modified:
   plugins/opie-sync/src/opie_sync.c

Modified: plugins/opie-sync/src/opie_sync.c
==============================================================================
--- plugins/opie-sync/src/opie_sync.c	Sun Jan 25 13:47:14 2009	(r5229)
+++ plugins/opie-sync/src/opie_sync.c	Sun Jan 25 14:00:26 2009	(r5230)
@@ -347,6 +347,10 @@
 
 			/* Retrieve and set the uid of the object */
 			char *uid = opie_xml_get_tagged_uid(item_node);
+			if(!uid) {
+				osync_trace(TRACE_EXIT_ERROR, "%s: unable to get UID from entry");
+				return;
+			}
 			osync_change_set_uid(change, uid);
 			g_free(uid);
 			

[Opensync-commits] r5229 - plugins/opie-sync/src

[<dg...@su...>]

Author: paule
Date: Sun Jan 25 13:47:14 2009
New Revision: 5229
URL: http://www.opensync.org/changeset/5229

Log:
Fix for anchor API changes; fix for change to discovery function signature

Modified:
   plugins/opie-sync/src/opie_sync.c

Modified: plugins/opie-sync/src/opie_sync.c
==============================================================================
--- plugins/opie-sync/src/opie_sync.c	Sat Jan 24 19:13:57 2009	(r5228)
+++ plugins/opie-sync/src/opie_sync.c	Sun Jan 25 13:47:14 2009	(r5229)
@@ -298,10 +298,12 @@
 
 	if(env->objtype == OPIE_OBJECT_TYPE_NOTE) {
 		/* Check if the notes resource has changed since the last sync */
-		char *anchorpath = g_strdup_printf("%s/anchor.db", osync_plugin_info_get_configdir(info));
-		if (!osync_anchor_compare(anchorpath, "notespath", env->plugin_env->notes_path))
+		OSyncAnchor *anchor = osync_objtype_sink_get_anchor(sink);
+		osync_bool anchor_result = FALSE;
+		if(!osync_anchor_compare(anchor, env->plugin_env->notes_path, &anchor_result, &error))
+			goto error;
+		if(!anchor_result)
 			osync_objtype_sink_set_slowsync(sink, TRUE);
-		g_free(anchorpath);
 	}
 	
 	osync_context_report_success(ctx);
@@ -561,9 +563,9 @@
 	}
 	
 	if(env->objtype == OPIE_OBJECT_TYPE_NOTE) {
-		char *anchorpath = g_strdup_printf("%s/anchor.db", osync_plugin_info_get_configdir(info));
- 		osync_anchor_update(anchorpath, "notespath", env->plugin_env->notes_path);
-		g_free(anchorpath);
+		OSyncAnchor *anchor = osync_objtype_sink_get_anchor(sink);
+ 		if(!osync_anchor_update(anchor, env->plugin_env->notes_path, &error))
+			goto error;
 	}
 	
 	if (!osync_hashtable_save(env->hashtable, &error))
@@ -662,6 +664,7 @@
 	env->note_env = opie_sync_create_sink_env(env, info, "note", OPIE_FORMAT_XML_NOTE, OPIE_OBJECT_TYPE_NOTE, NULL, "notes", "note", error);
 	if(!env->note_env)
 		goto error_free_env;
+	osync_objtype_sink_enable_anchor(env->note_env->sink, TRUE);
 	
 	env->qcopconn = NULL;
 	env->connected = FALSE;
@@ -708,7 +711,7 @@
 	osync_trace(TRACE_EXIT, "%s", __func__);
 }
 
-static osync_bool opie_sync_discover(void *data, OSyncPluginInfo *info, OSyncError **error)
+static osync_bool opie_sync_discover(OSyncPluginInfo *info, void *data, OSyncError **error)
 {
 	osync_trace(TRACE_ENTRY, "%s(%p, %p, %p)", __func__, data, info, error);
 	
@@ -728,6 +731,7 @@
 	//osync_version_set_hardwareversion(version, "hardwareversion");
 	osync_plugin_info_set_version(info, version);
 	osync_version_unref(version);
+	
 
 	/* FIXME define capabilities */
 

[Opensync-commits] r5228 - in branches/timeout: . opensync/ipc

[<dg...@su...>]

Author: Graham Cobb
Date: Sat Jan 24 19:13:57 2009
New Revision: 5228
URL: http://www.opensync.org/changeset/5228

Log:
Done: 3) Add references for cross-linked pointers between queues and
remove linkages (and unref) when queue disconnected

All tests passed

This leaves:

4) Handle timeout before reply queue set up by generating an
error message on the command queue
6) Implement a mechanism to protect the queue itself: i.e. is the
receiver still reading the queue?

Modified:
   branches/timeout/ChangeLog
   branches/timeout/opensync/ipc/opensync_queue.c
   branches/timeout/opensync/ipc/opensync_queue_internals.h

Modified: branches/timeout/ChangeLog
==============================================================================
--- branches/timeout/ChangeLog	Sat Jan 24 01:29:49 2009	(r5227)
+++ branches/timeout/ChangeLog	Sat Jan 24 19:13:57 2009	(r5228)
@@ -1,3 +1,11 @@
+2009-01-24  Graham Cobb  <g+...@co...>
+
+	* opensync/ipc/opensync_queue.c (osync_queue_remove_cross_link): Add
+	osync_queue_remove_cross_link
+	(osync_queue_disconnect): Call osync_queue_remove_cross_link
+
+	* opensync/ipc/opensync_queue_internals.h: Add osync_queue_remove_cross_link
+
 2009-01-23  Graham Cobb  <g+...@co...>
 
 	* opensync/client/opensync_client.c (osync_client_set_incoming_queue): Call

Modified: branches/timeout/opensync/ipc/opensync_queue.c
==============================================================================
--- branches/timeout/opensync/ipc/opensync_queue.c	Sat Jan 24 01:29:49 2009	(r5227)
+++ branches/timeout/opensync/ipc/opensync_queue.c	Sat Jan 24 19:13:57 2009	(r5228)
@@ -1000,6 +1000,8 @@
 	osync_trace(TRACE_ENTRY, "%s(%p, %p)", __func__, queue, error);
 	osync_assert(queue);
 
+	osync_queue_remove_cross_link(queue);
+
 	g_mutex_lock(queue->disconnectLock);
 	if (queue->thread) {
 		osync_thread_stop(queue->thread);
@@ -1068,11 +1070,31 @@
 	osync_assert(reply_queue->type == OSYNC_QUEUE_SENDER);
 
 	cmd_queue->reply_queue = reply_queue;
+	osync_queue_ref(reply_queue);
 	reply_queue->cmd_queue = cmd_queue;
+	osync_queue_ref(cmd_queue);
 	
 	osync_trace(TRACE_EXIT, "%s", __func__);
 }
 
+void osync_queue_remove_cross_link(OSyncQueue *queue)
+{
+	/* Remove the cross links from this queue and all queues linked
+	   from it, recursively */
+	if (queue->cmd_queue) {
+		OSyncQueue *linked_queue = queue->cmd_queue;
+		queue->cmd_queue = NULL;
+		osync_queue_remove_cross_link(linked_queue);
+		osync_queue_unref(linked_queue);
+	}
+	if (queue->reply_queue) {
+		OSyncQueue *linked_queue = queue->reply_queue;
+		queue->reply_queue = NULL;
+		osync_queue_remove_cross_link(linked_queue);
+		osync_queue_unref(linked_queue);
+	}
+}
+
 void osync_queue_set_pending_limit(OSyncQueue *queue, unsigned int limit)
 {
 	/* The pending limit is used on queues which receive commands and run timeouts.

Modified: branches/timeout/opensync/ipc/opensync_queue_internals.h
==============================================================================
--- branches/timeout/opensync/ipc/opensync_queue_internals.h	Sat Jan 24 01:29:49 2009	(r5227)
+++ branches/timeout/opensync/ipc/opensync_queue_internals.h	Sat Jan 24 19:13:57 2009	(r5228)
@@ -99,6 +99,17 @@
 OSYNC_TEST_EXPORT void osync_queue_cross_link(OSyncQueue *cmd_queue, OSyncQueue *reply_queue);
 
 /**
+ * @brief Remove cross links between command queues and reply queues
+ * 
+ * Removes the cross-links from this queue and all queues linked
+ * from it, recursively
+ * 
+ * @param queue The queue to unlink
+ * 
+ */
+OSYNC_TEST_EXPORT void osync_queue_remove_cross_link(OSyncQueue *queue);
+
+/**
  * @brief Set pending limit on queue
  * 
  * This should be used on queues used to receive incoming commands to

[Opensync-commits] r5227 - in branches/timeout: . opensync/client opensync/ipc tests tests/ipc-tests

[<dg...@su...>]

Author: Graham Cobb
Date: Sat Jan 24 01:29:49 2009
New Revision: 5227
URL: http://www.opensync.org/changeset/5227

Log:
Completed:
2) Limit the number of pending entries so the timeout is not
dependent on the number of outstanding requests

And added unit test.

That leaves (at least):

3) Add references for cross-linked pointers between queues and
remove linkages (and unref) when ??? (queue disconnected?)
4) Handle timeout before reply queue set up by generating an
error message on the command queue
6) Implement a mechanism to protect the queue itself: i.e. is the
receiver still reading the queue?

Modified:
   branches/timeout/ChangeLog
   branches/timeout/opensync/client/opensync_client.c
   branches/timeout/opensync/ipc/opensync_queue.c
   branches/timeout/opensync/ipc/opensync_queue_internals.h
   branches/timeout/opensync/ipc/opensync_queue_private.h
   branches/timeout/tests/CMakeLists.txt
   branches/timeout/tests/ipc-tests/check_ipc.c

Modified: branches/timeout/ChangeLog
==============================================================================
--- branches/timeout/ChangeLog	Fri Jan 23 15:07:37 2009	(r5226)
+++ branches/timeout/ChangeLog	Sat Jan 24 01:29:49 2009	(r5227)
@@ -1,3 +1,26 @@
+2009-01-23  Graham Cobb  <g+...@co...>
+
+	* opensync/client/opensync_client.c (osync_client_set_incoming_queue): Call
+	osync_queue_set_pending_limit
+
+	* tests/ipc-tests/check_ipc.c: Call osync_queue_set_pending_limit in timeout tests.
+
+	* opensync/ipc/opensync_queue_internals.h: Add osync_queue_set_pending_limit
+	and OSYNC_QUEUE_PENDING_LIMIT
+
+	* opensync/ipc/opensync_queue.c (_incoming_check): Check pendingLimit not exceeded.
+	Add increments/decrements of pendingCount whenever pendingReplies is manipulated.
+	(osync_queue_set_pending_limit): Add osync_queue_set_pending_limit
+
+	* opensync/ipc/opensync_queue_private.h: Add pendingCount, pendingLimit to OSyncQueue
+
+	* tests/CMakeLists.txt: Add ipc_loop_timeout_with_idle.
+
+	* tests/ipc-tests/check_ipc.c: Rename callback_handler as callback_handler_check_reply.
+	Rename server_handler3 as server_handler_abort.
+	Rename client_handler6 as client_handler_sleep.
+	Add ipc_loop_timeout_with_idle.
+
 2009-01-22  Graham Cobb  <g+...@co...>
 
 	* tests/CMakeLists.txt: Add ipc_loop_with_timeout

Modified: branches/timeout/opensync/client/opensync_client.c
==============================================================================
--- branches/timeout/opensync/client/opensync_client.c	Fri Jan 23 15:07:37 2009	(r5226)
+++ branches/timeout/opensync/client/opensync_client.c	Sat Jan 24 01:29:49 2009	(r5227)
@@ -1611,6 +1611,7 @@
 	osync_queue_set_message_handler(incoming, _osync_client_message_handler, client);
 	osync_queue_setup_with_gmainloop(incoming, client->context);
 	client->incoming = osync_queue_ref(incoming);
+	osync_queue_set_pending_limit(incoming, OSYNC_QUEUE_PENDING_LIMIT);
 }
 
 void osync_client_set_outgoing_queue(OSyncClient *client, OSyncQueue *outgoing)

Modified: branches/timeout/opensync/ipc/opensync_queue.c
==============================================================================
--- branches/timeout/opensync/ipc/opensync_queue.c	Fri Jan 23 15:07:37 2009	(r5226)
+++ branches/timeout/opensync/ipc/opensync_queue.c	Sat Jan 24 01:29:49 2009	(r5227)
@@ -128,6 +128,7 @@
 				 gets called twice! */
 
 			queue->pendingReplies = osync_list_remove(queue->pendingReplies, pending);
+			queue->pendingCount--;
 			/* Unlock the pending lock since the messages might be sent during the callback */
 			g_mutex_unlock(queue->pendingLock);
 
@@ -163,7 +164,18 @@
 static gboolean _incoming_check(GSource *source)
 {
 	OSyncQueue *queue = *((OSyncQueue **)(source + 1));
-	if (g_async_queue_length(queue->incoming) > 0)
+
+	/* As well as checking there is something on the incoming queue, we check
+	   that we are not blocked because we have too many pending commands.
+	   This check is only done if the pendingLimit has been set.
+
+	   Note, to avoid taking out the pending lock in order to count
+	   the length of the pending queue we use the pendingCount counter.
+	   This better not get out of line! */
+
+	if (g_async_queue_length(queue->incoming) > 0
+	    && (queue->pendingLimit == 0 
+		|| queue->pendingCount < queue->pendingLimit) )
 		return TRUE;
 	
 	return FALSE;
@@ -214,6 +226,7 @@
 			   gets called twice! */
 			
 			queue->pendingReplies = osync_list_remove(queue->pendingReplies, pending);
+			queue->pendingCount--;
 
 			/* Unlock the pending lock since the messages might be sent during the callback */
 			g_mutex_unlock(queue->pendingLock);
@@ -247,8 +260,11 @@
 	OSyncError *error = NULL;
 	
 	osync_trace(TRACE_ENTRY, "%s(%p)", __func__, user_data);
+	osync_trace(TRACE_INTERNAL, "queue->pendingCount = %d, queue->pendingLimit = %d", queue->pendingCount, queue->pendingLimit);
+
+	while ( (queue->pendingLimit == 0 || queue->pendingCount < queue->pendingLimit)
+		&& (message = g_async_queue_try_pop(queue->incoming)) ) {
 
-	while ((message = g_async_queue_try_pop(queue->incoming))) {
 		/* We check if the message is a reply to something */
 		osync_trace(TRACE_INTERNAL, "Dispatching %p:%i(%s), timeout=%d, id=%lli", message, 
 			    osync_message_get_cmd(message), osync_message_get_commandstr(message), 
@@ -290,6 +306,7 @@
 		
 				g_mutex_lock(queue->pendingLock);
 				queue->pendingReplies = osync_list_append(queue->pendingReplies, pending);
+				queue->pendingCount++;
 				g_mutex_unlock(queue->pendingLock);
 			}
 
@@ -807,6 +824,7 @@
 			pending = queue->pendingReplies->data;
 
 			queue->pendingReplies = osync_list_remove(queue->pendingReplies, pending);
+			queue->pendingCount--;
 
 			/** @todo Refcounting for OSyncPendingMessage */
 			if (pending->timeout_info)
@@ -815,6 +833,8 @@
 			osync_free(pending);
 		}
 
+		osync_assert(queue->pendingCount == 0);
+
 		if (queue->name)
 			osync_free(queue->name);
 		
@@ -1053,6 +1073,20 @@
 	osync_trace(TRACE_EXIT, "%s", __func__);
 }
 
+void osync_queue_set_pending_limit(OSyncQueue *queue, unsigned int limit)
+{
+	/* The pending limit is used on queues which receive commands and run timeouts.
+	   It is used to limit the number of pending transactions so timeouts don't occur
+	   just because there are a lot of commands waiting to complete.
+	   It is not necessary on other queues and, in fact, MUST NOT be set on
+	   queues which receive command replies as it could cause deadlocks. */
+	osync_trace(TRACE_ENTRY, "%s(%p, %u)", __func__, queue, limit);
+
+	queue->pendingLimit = limit;
+	
+	osync_trace(TRACE_EXIT, "%s", __func__);
+}
+
 void osync_queue_setup_with_gmainloop(OSyncQueue *queue, GMainContext *context)
 {
 	OSyncQueue **queueptr = NULL;
@@ -1189,6 +1223,7 @@
 		
 		g_mutex_lock(replyqueue->pendingLock);
 		replyqueue->pendingReplies = osync_list_append(replyqueue->pendingReplies, pending);
+		replyqueue->pendingCount++;
 		g_mutex_unlock(replyqueue->pendingLock);
 	}
 	

Modified: branches/timeout/opensync/ipc/opensync_queue_internals.h
==============================================================================
--- branches/timeout/opensync/ipc/opensync_queue_internals.h	Fri Jan 23 15:07:37 2009	(r5226)
+++ branches/timeout/opensync/ipc/opensync_queue_internals.h	Sat Jan 24 01:29:49 2009	(r5227)
@@ -99,6 +99,26 @@
 OSYNC_TEST_EXPORT void osync_queue_cross_link(OSyncQueue *cmd_queue, OSyncQueue *reply_queue);
 
 /**
+ * @brief Set pending limit on queue
+ * 
+ * This should be used on queues used to receive incoming commands to
+ * limit the number of outstanding commands waiting for replies.
+ * This avoids timing out commands which are just waiting for previous
+ * commands to finish.
+ *
+ * Note: the pending limit should not be set on other queues and
+ * MUST NOT be set on command reply queues as it could cause deadlocks.
+ *
+ * The recommended value to use is OSYNC_QUEUE_PENDING_LIMIT.
+ * 
+ * @param queue The command queue used to receive incoming commands
+ * @param limit The maximum number of waiting commands 
+ * 
+ */
+OSYNC_TEST_EXPORT void osync_queue_set_pending_limit(OSyncQueue *queue, unsigned int limit);
+#define OSYNC_QUEUE_PENDING_LIMIT 5
+
+/**
  * @brief Sends a Message to a Queue
  * @param queue Pointer to the queue
  * @param replyqueue

Modified: branches/timeout/opensync/ipc/opensync_queue_private.h
==============================================================================
--- branches/timeout/opensync/ipc/opensync_queue_private.h	Fri Jan 23 15:07:37 2009	(r5226)
+++ branches/timeout/opensync/ipc/opensync_queue_private.h	Sat Jan 24 01:29:49 2009	(r5227)
@@ -71,6 +71,7 @@
 	/** List of pending replies */
 	OSyncList *pendingReplies;
 	GMutex *pendingLock;
+	unsigned int pendingCount, pendingLimit;
 	
 	GSourceFuncs *write_functions;
 	GSource *write_source;

Modified: branches/timeout/tests/CMakeLists.txt
==============================================================================
--- branches/timeout/tests/CMakeLists.txt	Fri Jan 23 15:07:37 2009	(r5226)
+++ branches/timeout/tests/CMakeLists.txt	Sat Jan 24 01:29:49 2009	(r5227)
@@ -248,6 +248,7 @@
 OSYNC_TESTCASE(ipc ipc_timeout)
 OSYNC_TESTCASE(ipc ipc_loop_with_timeout)
 OSYNC_TESTCASE(ipc ipc_late_reply)
+OSYNC_TESTCASE(ipc ipc_loop_timeout_with_idle)
 ENDIF (NOT WIN32)
 
 BUILD_CHECK_TEST( mapping mapping-tests/check_mapping.c ${TEST_TARGET_LIBRARIES} )

Modified: branches/timeout/tests/ipc-tests/check_ipc.c
==============================================================================
--- branches/timeout/tests/ipc-tests/check_ipc.c	Fri Jan 23 15:07:37 2009	(r5226)
+++ branches/timeout/tests/ipc-tests/check_ipc.c	Sat Jan 24 01:29:49 2009	(r5227)
@@ -1362,7 +1362,7 @@
 }
 END_TEST
 
-void callback_handler(OSyncMessage *message, void *user_data)
+void callback_handler_check_reply(OSyncMessage *message, void *user_data)
 {
 	osync_trace(TRACE_ENTRY, "%s(%p, %p)", __func__, message, user_data);
 	OSyncError *error = NULL;
@@ -1380,7 +1380,7 @@
 	osync_trace(TRACE_EXIT, "%s", __func__);
 }
 
-void server_handler3(OSyncMessage *message, void *user_data)
+void server_handler_abort(OSyncMessage *message, void *user_data)
 {
 	abort();
 }
@@ -1484,7 +1484,7 @@
 		GMainContext *context = g_main_context_new();
 		OSyncThread *thread = osync_thread_new(context, &error);
 		
-		osync_queue_set_message_handler(server_queue, server_handler3, GINT_TO_POINTER(1));
+		osync_queue_set_message_handler(server_queue, server_handler_abort, GINT_TO_POINTER(1));
 		
 		osync_queue_setup_with_gmainloop(server_queue, context);
 		
@@ -1507,7 +1507,7 @@
 			osync_message_write_long_long_int(message, 400000000);
 			osync_message_write_data(message, data, strlen(data) + 1);
 			
-			osync_message_set_handler(message, callback_handler, GINT_TO_POINTER(1));
+			osync_message_set_handler(message, callback_handler_check_reply, GINT_TO_POINTER(1));
 			
 			fail_unless(osync_queue_send_message(client_queue, server_queue, message, &error), NULL);
 			fail_unless(!osync_error_is_set(&error), NULL);
@@ -2164,6 +2164,7 @@
 		OSyncThread *thread = osync_thread_new(context, &error);
 	
 		osync_queue_set_message_handler(client_queue, client_handler5, GINT_TO_POINTER(1));
+		osync_queue_set_pending_limit(client_queue, OSYNC_QUEUE_PENDING_LIMIT);
 		
 		osync_queue_setup_with_gmainloop(client_queue, context);
 		
@@ -2275,7 +2276,8 @@
 }
 END_TEST
 
-void client_handler6(OSyncMessage *message, void *user_data)
+int ch_sleep_time = 3; // Seconds
+void client_handler_sleep(OSyncMessage *message, void *user_data)
 {
 	osync_trace(TRACE_ENTRY, "%s(%p, %p)", __func__, message, user_data);
 	OSyncError *error = NULL;
@@ -2300,7 +2302,7 @@
 	osync_assert(!strcmp(databuf, data5));
 
 	// Do some time consuming processing
-	g_usleep(3*G_USEC_PER_SEC);
+	g_usleep(ch_sleep_time*G_USEC_PER_SEC);
 	
 	OSyncMessage *reply = osync_message_new_reply(message, &error);
 	
@@ -2330,7 +2332,8 @@
 
 	num_callback_timeout = 0;
 	num_callback = 0;
-	
+	ch_sleep_time = 3;
+
 	OSyncError *error = NULL;
 	server_queue = osync_queue_new("/tmp/testpipe-server", &error);
 	client_queue = osync_queue_new("/tmp/testpipe-client", &error);
@@ -2348,7 +2351,8 @@
 		GMainContext *context = g_main_context_new();
 		OSyncThread *thread = osync_thread_new(context, &error);
 	
-		osync_queue_set_message_handler(client_queue, client_handler6, GINT_TO_POINTER(1));
+		osync_queue_set_message_handler(client_queue, client_handler_sleep, GINT_TO_POINTER(1));
+		osync_queue_set_pending_limit(client_queue, OSYNC_QUEUE_PENDING_LIMIT);
 		
 		osync_queue_setup_with_gmainloop(client_queue, context);
 		
@@ -2460,72 +2464,211 @@
 }
 END_TEST
 
-void callback_handler7(OSyncMessage *message, void *user_data)
+START_TEST (ipc_loop_with_timeout)
 {
-	osync_trace(TRACE_ENTRY, "%s(%p, %p)", __func__, message, user_data);
+	
+	/* Even though each action takes 1 second, none of these messages should time out
+	   as they are being sent with a timeout of 3 seconds */
+	
+	num_msgs = 0;
+	req_msgs = 20;
+	ch_sleep_time = 1; // Second
+	
+	char *testbed = setup_testbed(NULL);
+	osync_testing_file_remove("/tmp/testpipe-server");
+	osync_testing_file_remove("/tmp/testpipe-client");
+	
 	OSyncError *error = NULL;
+	server_queue = osync_queue_new("/tmp/testpipe-server", &error);
+	client_queue = osync_queue_new("/tmp/testpipe-client", &error);
+	OSyncMessage *message = NULL;
 	
-	osync_assert(GPOINTER_TO_INT(user_data) == 1);
+	osync_queue_create(server_queue, &error);
+	fail_unless(error == NULL, NULL);
 	
-	num_msgs++;
-	osync_assert(osync_message_get_command(message) == OSYNC_MESSAGE_REPLY);
+	osync_queue_create(client_queue, &error);
+	fail_unless(error == NULL, NULL);
+	char *data = "this is another test string";
 	
-	if (num_msgs >= req_msgs) {
-		osync_queue_disconnect(server_queue, &error);
+	pid_t cpid = fork();
+	if (cpid == 0) { //Child
+		
+		GMainContext *context = g_main_context_new();
+		OSyncThread *thread = osync_thread_new(context, &error);
+	
+		osync_queue_set_message_handler(client_queue, client_handler_sleep, GINT_TO_POINTER(1));
+		osync_queue_set_pending_limit(client_queue, OSYNC_QUEUE_PENDING_LIMIT);
+		
+		osync_queue_setup_with_gmainloop(client_queue, context);
+		
+		osync_thread_start(thread);
+		
+		osync_assert(osync_queue_connect(client_queue, OSYNC_QUEUE_RECEIVER, &error));
 		osync_assert(error == NULL);
+
+		osync_assert(osync_queue_connect(server_queue, OSYNC_QUEUE_SENDER, &error));
+		osync_assert(error == NULL);
+
+		osync_queue_cross_link(client_queue, server_queue);
+		
+		message = osync_queue_get_message(server_queue);
+		
+		if (osync_message_get_command(message) != OSYNC_MESSAGE_QUEUE_HUP) {
+			exit (1);
+		}
+		
+		osync_message_unref(message);
+	
+		if (osync_queue_disconnect(server_queue, &error) != TRUE || error != NULL)
+			exit(1);
+		osync_queue_unref(server_queue);
+		
+		osync_assert(osync_queue_disconnect(client_queue, &error));
+		osync_assert(error == NULL);
+		
+		osync_thread_stop(thread);
+		osync_thread_unref(thread);
+		
+		osync_queue_unref(client_queue);
+		
+		g_free(testbed);
+		
+		exit(0);
+	} else {
+		GMainContext *context = g_main_context_new();
+		OSyncThread *thread = osync_thread_new(context, &error);
+		
+		osync_queue_set_message_handler(server_queue, server_handler_abort, GINT_TO_POINTER(1));
+		
+		osync_queue_setup_with_gmainloop(server_queue, context);
+		
+		osync_thread_start(thread);
+	
+		fail_unless(osync_queue_connect(client_queue, OSYNC_QUEUE_SENDER, &error), NULL);
+		fail_unless(error == NULL, NULL);
+		
+		fail_unless(osync_queue_connect(server_queue, OSYNC_QUEUE_RECEIVER, &error), NULL);
+		fail_unless(error == NULL, NULL);
+		
+		int i = 0;
+		for (i = 0; i < req_msgs; i++) {
+			message = osync_message_new(OSYNC_MESSAGE_INITIALIZE, 0, &error);
+			fail_unless(message != NULL, NULL);
+			fail_unless(!osync_error_is_set(&error), NULL);
+			
+			osync_message_write_int(message, 4000000);
+			osync_message_write_string(message, "this is a test string");
+			osync_message_write_long_long_int(message, 400000000);
+			osync_message_write_data(message, data, strlen(data) + 1);
+			
+			osync_message_set_handler(message, callback_handler_check_reply, GINT_TO_POINTER(1));
+			
+			fail_unless(osync_queue_send_message_with_timeout(client_queue, server_queue, message, 3, &error), NULL);
+			fail_unless(!osync_error_is_set(&error), NULL);
+			
+			osync_message_unref(message);
+		}
+		
+		message = osync_queue_get_message(client_queue);
+		
+		fail_unless(osync_message_get_command(message) == OSYNC_MESSAGE_QUEUE_HUP);
+		
+		osync_message_unref(message);
+		
+		osync_queue_disconnect(client_queue, &error);
+		fail_unless(error == NULL, NULL);
+		
+		osync_thread_stop(thread);
+		osync_thread_unref(thread);
+		
+		int status = 0;
+		wait(&status);
+		fail_unless(WEXITSTATUS(status) == 0, NULL);
 	}
 	
-	osync_trace(TRACE_EXIT, "%s", __func__);
-}
+	fail_unless(osync_testing_file_exists("/tmp/testpipe-client") == TRUE, NULL);
+	
+	fail_unless(osync_queue_remove(client_queue, &error), NULL);
+	fail_unless(osync_queue_remove(server_queue, &error), NULL);
+	fail_unless(!osync_error_is_set(&error), NULL);
+	
+	fail_unless(osync_testing_file_exists("/tmp/testpipe-client") == FALSE, NULL);
 
-void server_handler7(OSyncMessage *message, void *user_data)
-{
-	abort();
+	osync_queue_unref(client_queue);
+	osync_queue_unref(server_queue);
+	
+	destroy_testbed(testbed);
 }
+END_TEST
 
-void client_handler7(OSyncMessage *message, void *user_data)
+GSList *ch_pending = NULL;
+void client_handler_first_part(OSyncMessage *message, void *user_data)
 {
 	osync_trace(TRACE_ENTRY, "%s(%p, %p)", __func__, message, user_data);
 	OSyncError *error = NULL;
 	
 	osync_assert(GPOINTER_TO_INT(user_data) ==1);
 	osync_assert(osync_message_get_command(message) == OSYNC_MESSAGE_INITIALIZE);
-	
+		
 	int int1;
 	long long int longint1;
 	char *string;
-	void *databuf;
+	char databuf[strlen(data5) + 1];
+	
 	
 	osync_message_read_int(message, &int1);
-	osync_message_read_const_string(message, &string);
+	osync_message_read_string(message, &string);
 	osync_message_read_long_long_int(message, &longint1);
-	osync_message_read_const_data(message, &databuf, strlen("this is another test string") + 1);
+	osync_message_read_data(message, databuf, strlen(data5) + 1);
 	
 	osync_assert(int1 == 4000000);
 	osync_assert(!strcmp(string, "this is a test string"));
 	osync_assert(longint1 == 400000000);
-	osync_assert(!strcmp(databuf, "this is another test string"));
+	osync_assert(!strcmp(databuf, data5));
 
-	// Do some time consuming processing
-	g_usleep(1*G_USEC_PER_SEC);
-	
-	OSyncMessage *reply = osync_message_new_reply(message, &error);
-	
-	osync_queue_send_message(server_queue, NULL, reply, &error);
+	// Put message on pending queue and return
+	osync_message_ref(message);
+	ch_pending = g_slist_append(ch_pending, message);
+
+	osync_trace(TRACE_EXIT, "%s", __func__);
+}
+gboolean client_handler_second_part(gpointer userdata)
+{
+	OSyncError *error = NULL;
+
+	osync_trace(TRACE_ENTRY, "%s(%p)", __func__, userdata);
+
+	if (ch_pending) {
+		
+		OSyncMessage *message = ch_pending->data;
+
+		OSyncMessage *reply = osync_message_new_reply(message, &error);
 	
-	osync_message_unref(reply);
+		osync_queue_send_message(server_queue, NULL, reply, &error);
 	
-	osync_trace(TRACE_EXIT, "%s", __func__);
+		osync_message_unref(reply);
+
+		ch_pending = g_slist_remove(ch_pending, message);
+		osync_message_unref(message);
+
+		osync_trace(TRACE_EXIT, "%s", __func__);
+		return TRUE;
+	}
+		
+	osync_trace(TRACE_EXIT, "%s: no more entries", __func__);
+	return FALSE;
 }
 
-START_TEST (ipc_loop_with_timeout)
+START_TEST (ipc_loop_timeout_with_idle)
 {
 	
-	/* Even though each action takes 1 second, none of these messages should time out
+	/* Same as ipc_loop_with_timeout except that the client handler doesn't sleep,
+	   so the queue dispatchers can run while the operation is waiting.
+	   Even though each action takes 1 second, none of these messages should time out
 	   as they are being sent with a timeout of 3 seconds */
 	
 	num_msgs = 0;
-	req_msgs = 20;
+	req_msgs = 30;
 	
 	char *testbed = setup_testbed(NULL);
 	osync_testing_file_remove("/tmp/testpipe-server");
@@ -2549,7 +2692,8 @@
 		GMainContext *context = g_main_context_new();
 		OSyncThread *thread = osync_thread_new(context, &error);
 	
-		osync_queue_set_message_handler(client_queue, client_handler7, GINT_TO_POINTER(1));
+		osync_queue_set_message_handler(client_queue, client_handler_first_part, GINT_TO_POINTER(1));
+		osync_queue_set_pending_limit(client_queue, OSYNC_QUEUE_PENDING_LIMIT);
 		
 		osync_queue_setup_with_gmainloop(client_queue, context);
 		
@@ -2562,6 +2706,12 @@
 		osync_assert(error == NULL);
 
 		osync_queue_cross_link(client_queue, server_queue);
+
+		GSource *tsource = g_timeout_source_new(1000);
+		osync_assert(tsource);
+		g_source_set_callback(tsource, client_handler_second_part, NULL, NULL);
+		osync_assert(g_source_attach(tsource, context));
+		g_source_unref(tsource);
 		
 		message = osync_queue_get_message(server_queue);
 		
@@ -2590,7 +2740,7 @@
 		GMainContext *context = g_main_context_new();
 		OSyncThread *thread = osync_thread_new(context, &error);
 		
-		osync_queue_set_message_handler(server_queue, server_handler7, GINT_TO_POINTER(1));
+		osync_queue_set_message_handler(server_queue, server_handler_abort, GINT_TO_POINTER(1));
 		
 		osync_queue_setup_with_gmainloop(server_queue, context);
 		
@@ -2613,9 +2763,9 @@
 			osync_message_write_long_long_int(message, 400000000);
 			osync_message_write_data(message, data, strlen(data) + 1);
 			
-			osync_message_set_handler(message, callback_handler7, GINT_TO_POINTER(1));
+			osync_message_set_handler(message, callback_handler_check_reply, GINT_TO_POINTER(1));
 			
-			fail_unless(osync_queue_send_message_with_timeout(client_queue, server_queue, message, 3, &error), NULL);
+			fail_unless(osync_queue_send_message_with_timeout(client_queue, server_queue, message, 10, &error), NULL);
 			fail_unless(!osync_error_is_set(&error), NULL);
 			
 			osync_message_unref(message);
@@ -2681,5 +2831,6 @@
 OSYNC_TESTCASE_ADD(ipc_timeout)
 OSYNC_TESTCASE_ADD(ipc_late_reply)
 OSYNC_TESTCASE_ADD(ipc_loop_with_timeout)
+OSYNC_TESTCASE_ADD(ipc_loop_timeout_with_idle)
 OSYNC_TESTCASE_END
 

[Opensync-commits] r183 - wbxml2/trunk

[<dg...@su...>]

Author: bellmich
Date: Fri Jan 23 15:50:38 2009
New Revision: 183
URL: http://libwbxml.opensync.org/changeset/183

Log:
increment number of OMA DM DDF tests

Modified:
   wbxml2/trunk/CMakeLists.txt

Modified: wbxml2/trunk/CMakeLists.txt
==============================================================================
--- wbxml2/trunk/CMakeLists.txt	Fri Jan 23 15:50:09 2009	(r182)
+++ wbxml2/trunk/CMakeLists.txt	Fri Jan 23 15:50:38 2009	(r183)
@@ -229,7 +229,7 @@
 CONFIGURE_FILE( "tests/launchTests.sh" "${CMAKE_CURRENT_BINARY_DIR}/launchTests.sh")
 CONFIGURE_FILE( "tests/normalize_xml.pl" "${CMAKE_CURRENT_BINARY_DIR}/normalize_xml.pl" @ONLY)
 SET( airsync_tests 1 )
-SET( ddf_tests 5 )
+SET( ddf_tests 7 )
 # 2009-Jan-19 bellmich
 # All DRMREL 1.0 tests are switched off.
 # The background is the extremely bad specification.

[Opensync-commits] r182 - in wbxmlTestSuite/trunk: . ddf

[<dg...@su...>]

Author: bellmich
Date: Fri Jan 23 15:50:09 2009
New Revision: 182
URL: http://libwbxml.opensync.org/changeset/182

Log:
added tests for embedded OMA DM DDF management trees

Added:
   wbxmlTestSuite/trunk/ddf/ddf-001.xml
   wbxmlTestSuite/trunk/ddf/syncml_with_ddf-001.xml
Modified:
   wbxmlTestSuite/trunk/launchTests.sh

Added: wbxmlTestSuite/trunk/ddf/ddf-001.xml
==============================================================================
--- /dev/null	00:00:00 1970	(empty, because file is newly added)
+++ wbxmlTestSuite/trunk/ddf/ddf-001.xml	Fri Jan 23 15:50:09 2009	(r182)
@@ -0,0 +1,55 @@
+<?xml version="1.0"?>
+<!DOCTYPE MgmtTree PUBLIC "-//OMA//DTD-DM-DDF 1.2//EN" "http://www.openmobilealliance.org/tech/DTD/dm_ddf-v1_2.dtd">
+            <MgmtTree xmlns="syncml:dmddf1.2">
+              <VerDTD>1.2</VerDTD>
+              <Node>
+                <NodeName>E-Mail</NodeName>
+                <RTProperties>
+                  <Format>
+                    <node/>
+                  </Format>
+                  <Type>
+                    <DDFName>com.operatorX.dm/1.0/EMail</DDFName>
+                  </Type>
+                </RTProperties>
+                <Node>
+                  <NodeName>POP3</NodeName>
+                  <Node>
+                    <NodeName>POPServer</NodeName>
+                    <RTProperties>
+                      <Format>
+                        <chr/>
+                      </Format>
+                      <Type>
+                        <MIME>text/plain</MIME>
+                      </Type>
+                    </RTProperties>
+                    <Value>mail.Operatorx.com</Value>
+                  </Node>
+                  <Node>
+                    <NodeName>UserID</NodeName>
+                    <RTProperties>
+                      <Format>
+                        <chr/>
+                      </Format>
+                      <Type>
+                        <MIME>text/plain</MIME>
+                      </Type>
+                    </RTProperties>
+                    <Value>UserName</Value>
+                  </Node>
+                  <Node>
+                    <NodeName>PWD</NodeName>
+                    <RTProperties>
+                      <Format>
+                        <chr/>
+                      </Format>
+                      <Type>
+                        <MIME>text/plain</MIME>
+                      </Type>
+                    </RTProperties>
+                    <Value>4571F7C34A9876B3</Value>
+                  </Node>
+                </Node>
+              </Node>
+            </MgmtTree>

Added: wbxmlTestSuite/trunk/ddf/syncml_with_ddf-001.xml
==============================================================================
--- /dev/null	00:00:00 1970	(empty, because file is newly added)
+++ wbxmlTestSuite/trunk/ddf/syncml_with_ddf-001.xml	Fri Jan 23 15:50:09 2009	(r182)
@@ -0,0 +1,153 @@
+<?xml version="1.0"?>
+<!DOCTYPE SyncML PUBLIC "-//SYNCML//DTD SyncML 1.2//EN" "http://www.openmobilealliance.org/tech/DTD/OMA-TS-SyncML_RepPro_DTD-V1_2.dtd">
+<SyncML>
+  <SyncHdr>
+    <VerDTD>1.2</VerDTD>
+    <VerProto>SyncML/1.2</VerProto>
+    <SessionID>1</SessionID>
+    <MsgID>2</MsgID>
+    <Target>
+      <LocURI>IMEI:493005100592800</LocURI>
+    </Target>
+    <Source>
+      <LocURI>http://www.syncml.org/sync-server</LocURI>
+    </Source>
+  </SyncHdr>
+  <SyncBody>
+    <Status>
+      <CmdID>1</CmdID>
+      <MsgRef>2</MsgRef>
+      <CmdRef>0</CmdRef>
+      <Cmd>SyncHdr</Cmd>
+      <TargetRef>http://www.syncml.org/sync-server</TargetRef>
+      <SourceRef>IMEI:493005100592800</SourceRef>
+      <Data>200</Data>
+    </Status>
+    <Status>
+      <!--This is a status for the client modifications to the server.-->
+      <CmdID>2</CmdID>
+      <MsgRef>2</MsgRef>
+      <CmdRef>3</CmdRef>
+      <Cmd>Sync</Cmd>
+      <TargetRef>./contacts/james_bond</TargetRef>
+      <SourceRef>./dev-contacts</SourceRef>
+      <Data>200</Data>
+      <!--Statuscode for Success-->
+    </Status>
+    <Status>
+      <CmdID>3</CmdID>
+      <MsgRef>2</MsgRef>
+      <CmdRef>4</CmdRef>
+      <Cmd>Replace</Cmd>
+      <SourceRef>1012</SourceRef>
+      <Data>200</Data>
+      <!--Statuscode for Success-->
+    </Status>
+    <Sync>
+      <CmdID>4</CmdID>
+      <Target>
+        <LocURI>./dev-contacts</LocURI>
+      </Target>
+      <Source>
+        <LocURI>./contacts/james_bond</LocURI>
+      </Source>
+      <Replace>
+        <CmdID>5</CmdID>
+        <Meta>
+          <Type xmlns="syncml:metinf">text/x-vcard</Type>
+        </Meta>
+        <Item>
+          <Target>
+            <LocURI>1023</LocURI>
+          </Target>
+          <!--The vCard data would be placed here.-->
+          <Data></Data>
+        </Item>
+      </Replace>
+      <Add>
+        <CmdID>6</CmdID>
+        <Meta>
+          <Type xmlns="syncml:metinf">text/x-vcard</Type>
+        </Meta>
+        <Item>
+          <Source>
+            <LocURI>10536681</LocURI>
+          </Source>
+          <!--The vCard data would be placed here.-->
+          <Data></Data>
+        </Item>
+      </Add>
+      <!-- This example was directly copied from the OMA DM TNDS 1.2 specification -->
+      <Add>
+        <CmdID>4</CmdID>
+        <Item>
+          <Target>
+            <LocURI>/OperatorX</LocURI>
+          </Target>
+          <Meta>
+            <Format xmlns="syncml:metinf">xml</Format>
+            <Type xmlns="syncml:metinf">
+                              application/vnd.syncml.dmtnds+xml
+                        </Type>
+          </Meta>
+          <Data>
+            <MgmtTree xmlns="syncml:dmddf1.2">
+              <VerDTD>1.2</VerDTD>
+              <Node>
+                <NodeName>E-Mail</NodeName>
+                <RTProperties>
+                  <Format>
+                    <node/>
+                  </Format>
+                  <Type>
+                    <DDFName>com.operatorX.dm/1.0/EMail</DDFName>
+                  </Type>
+                </RTProperties>
+                <Node>
+                  <NodeName>POP3</NodeName>
+                  <Node>
+                    <NodeName>POPServer</NodeName>
+                    <RTProperties>
+                      <Format>
+                        <chr/>
+                      </Format>
+                      <Type>
+                        <MIME>text/plain</MIME>
+                      </Type>
+                    </RTProperties>
+                    <Value>mail.Operatorx.com</Value>
+                  </Node>
+                  <Node>
+                    <NodeName>UserID</NodeName>
+                    <RTProperties>
+                      <Format>
+                        <chr/>
+                      </Format>
+                      <Type>
+                        <MIME>text/plain</MIME>
+                      </Type>
+                    </RTProperties>
+                    <Value>UserName</Value>
+                  </Node>
+                  <Node>
+                    <NodeName>PWD</NodeName>
+                    <RTProperties>
+                      <Format>
+                        <chr/>
+                      </Format>
+                      <Type>
+                        <MIME>text/plain</MIME>
+                      </Type>
+                    </RTProperties>
+                    <Value>4571F7C34A9876B3</Value>
+                  </Node>
+                </Node>
+              </Node>
+            </MgmtTree>
+          </Data>
+        </Item>
+      </Add>
+    </Sync>
+    <Final/>
+  </SyncBody>
+</SyncML>

Modified: wbxmlTestSuite/trunk/launchTests.sh
==============================================================================
--- wbxmlTestSuite/trunk/launchTests.sh	Fri Jan 23 15:48:59 2009	(r181)
+++ wbxmlTestSuite/trunk/launchTests.sh	Fri Jan 23 15:50:09 2009	(r182)
@@ -98,7 +98,7 @@
   if [ `basename $i` != 'ddf' ]; then
   	TESTLIST=$(find $i -maxdepth 1 -type f -name "*.xml" -print | sort )
   else
-  	TESTLIST=$(find $i -maxdepth 1 -type f -name "*.ddf" -print | sort )
+  	TESTLIST=$(find $i -maxdepth 1 -type f \( -name "*.ddf" -or -name "*.xml" \) -print | sort )
   fi
   if [ " $3" != " " ]
   then
@@ -146,7 +146,7 @@
     else if [ "$TESTDIR" = "airsync" ];
     then
         PARAMS="-l AIRSYNC"
-    else if [ "$TESTDIR" = "ddf" ];
+    else if [ "$TESTDIR" != `basename $i ddf` ];
     then
         PARAMS="-l DMDDF12"
     else

[Opensync-commits] r181 - wbxml2/trunk/src

[<dg...@su...>]

Author: bellmich
Date: Fri Jan 23 15:48:59 2009
New Revision: 181
URL: http://libwbxml.opensync.org/changeset/181

Log:
added support for embedded OMA DM DDF management trees

Modified:
   wbxml2/trunk/src/wbxml_encoder.c
   wbxml2/trunk/src/wbxml_errors.c
   wbxml2/trunk/src/wbxml_tables.c
   wbxml2/trunk/src/wbxml_tree.c
   wbxml2/trunk/src/wbxml_tree_clb_xml.c

Modified: wbxml2/trunk/src/wbxml_encoder.c
==============================================================================
--- wbxml2/trunk/src/wbxml_encoder.c	Thu Jan 22 16:26:23 2009	(r180)
+++ wbxml2/trunk/src/wbxml_encoder.c	Fri Jan 23 15:48:59 2009	(r181)
@@ -2000,6 +2000,10 @@
             if (WBXML_STRCASECMP(buffer, "application/vnd.syncml-devinf+xml") == 0) {
                 the_buffer = (WB_UTINY*) "application/vnd.syncml-devinf+wbxml";
             }
+            /* Change text in <Type> from "application/vnd.syncml.dmtnds+xml" to "application/vnd.syncml.dmtnds+wbxml" */
+            if (WBXML_STRCASECMP(buffer, "application/vnd.syncml.dmtnds+xml") == 0) {
+                the_buffer = (WB_UTINY*) "application/vnd.syncml.dmtnds+wbxml";
+            }
         }
 #endif /* WBXML_SUPPORT_SYNCML */
     }
@@ -4304,6 +4308,19 @@
             if ((tmp = wbxml_buffer_create_from_cstr("application/vnd.syncml-devinf+xml")) == NULL)
                 return WBXML_ERROR_NOT_ENOUGH_MEMORY;
         }
+        /* Change text in <Type> from "application/vnd.syncml.dmtnds+wbxml" to "application/vnd.syncml.dmtnds+xml" */
+        if ((encoder->lang->langID == WBXML_LANG_SYNCML_SYNCML12) &&
+            (encoder->current_tag != NULL) &&
+            (encoder->current_tag->wbxmlCodePage == 0x01 ) &&
+            (encoder->current_tag->wbxmlToken == 0x13 ) &&
+            (wbxml_buffer_compare_cstr(tmp, "application/vnd.syncml.dmtnds+wbxml") == 0))
+        {
+            wbxml_buffer_destroy(tmp);
+
+            /* Change Content */
+            if ((tmp = wbxml_buffer_create_from_cstr("application/vnd.syncml.dmtnds+xml")) == NULL)
+                return WBXML_ERROR_NOT_ENOUGH_MEMORY;
+        }
 #endif /* WBXML_SUPPORT_SYNCML */
 
         /* Fix text */

Modified: wbxml2/trunk/src/wbxml_errors.c
==============================================================================
--- wbxml2/trunk/src/wbxml_errors.c	Thu Jan 22 16:26:23 2009	(r180)
+++ wbxml2/trunk/src/wbxml_errors.c	Fri Jan 23 15:48:59 2009	(r181)
@@ -94,6 +94,7 @@
     /* WBXML Encoder Errors */
     { WBXML_ERROR_ENCODER_APPEND_DATA,          "Can't append data to output buffer" },
     { WBXML_ERROR_STRTBL_DISABLED,              "String Table generation disabled: can't encode Literal" },
+    { WBXML_ERROR_UNKNOWN_XML_LANGUAGE,		"The XML language is unknown." },
     { WBXML_ERROR_XML_NODE_NOT_ALLOWED,         "XML Node Type not allowed" },
     { WBXML_ERROR_XML_NULL_ATTR_NAME,           "NULL XML Attribute Name" },
     { WBXML_ERROR_XML_PARSING_FAILED,           "Parsing of XML Document Failed" },

Modified: wbxml2/trunk/src/wbxml_tables.c
==============================================================================
--- wbxml2/trunk/src/wbxml_tables.c	Thu Jan 22 16:26:23 2009	(r180)
+++ wbxml2/trunk/src/wbxml_tables.c	Fri Jan 23 15:48:59 2009	(r181)
@@ -2018,8 +2018,8 @@
 };
 
 const WBXMLNameSpaceEntry sv_syncml_dmddf12_ns_table[] = {
-    { "MgmtTree",	0x02 },     /**< Code Page 2: DMDDF */
-    { NULL,             0x00 }
+    { "syncml:dmddf1.2", 0x02 },     /**< Code Page 2: OMA DM DDF */
+    { NULL,              0x00 }
 };
 
 #endif /* WBXML_SUPPORT_SYNCML */

Modified: wbxml2/trunk/src/wbxml_tree.c
==============================================================================
--- wbxml2/trunk/src/wbxml_tree.c	Thu Jan 22 16:26:23 2009	(r180)
+++ wbxml2/trunk/src/wbxml_tree.c	Fri Jan 23 15:48:59 2009	(r181)
@@ -825,6 +825,16 @@
                     return WBXML_SYNCML_DATA_TYPE_NORMAL;
                 }
 
+                /* application/vnd.syncml.dmtnds+wbxml */
+                if (wbxml_buffer_compare_cstr(tmp_node->children->content, "application/vnd.syncml.dmtnds+wbxml") == 0) {
+                    return WBXML_SYNCML_DATA_TYPE_WBXML;
+                }
+                
+                /* application/vnd.syncml.dmtnds+xml */
+                if (wbxml_buffer_compare_cstr(tmp_node->children->content, "application/vnd.syncml.dmtnds+xml") == 0) {
+                    return WBXML_SYNCML_DATA_TYPE_NORMAL;
+                }
+
                 /* text/clear */
                 if (wbxml_buffer_compare_cstr(tmp_node->children->content, "text/clear") == 0) {
                     return WBXML_SYNCML_DATA_TYPE_CLEAR;

Modified: wbxml2/trunk/src/wbxml_tree_clb_xml.c
==============================================================================
--- wbxml2/trunk/src/wbxml_tree_clb_xml.c	Thu Jan 22 16:26:23 2009	(r180)
+++ wbxml2/trunk/src/wbxml_tree_clb_xml.c	Fri Jan 23 15:48:59 2009	(r181)
@@ -143,8 +143,14 @@
 
 #if defined( WBXML_SUPPORT_SYNCML )
 
-    /* If this is an embedded (not root) "DevInf" document, skip it */
-    if ((WBXML_STRCMP(localName, "syncml:devinf:DevInf") == 0) &&
+    /* If this is an embedded (not root) document, skip it
+     * Actually SyncML DevInf and DM DDF are known as such
+     * potentially embedded documents.
+     */
+    if ((
+         (WBXML_STRCMP(localName, "syncml:devinf:DevInf") == 0) ||
+         (WBXML_STRCMP(localName, "syncml:dmddf1.2:MgmtTree") == 0)
+        )&&
         (tree_ctx->current != NULL))
     {
         tree_ctx->skip_start = XML_GetCurrentByteIndex(tree_ctx->xml_parser);
@@ -174,7 +180,7 @@
 {
     WBXMLTreeClbCtx *tree_ctx = (WBXMLTreeClbCtx *) ctx;
 #if defined( WBXML_SUPPORT_SYNCML )
-    WBXMLBuffer *devinf_doc = NULL;
+    WBXMLBuffer *embed_doc = NULL;
     WBXMLTree *tree = NULL;
     WBXMLError ret = WBXML_OK;
 #endif /* WBXML_SUPPORT_SYNCML */
@@ -194,26 +200,43 @@
             /* End of skipped node */
 
 #if defined( WBXML_SUPPORT_SYNCML )
-            if (WBXML_STRCMP(localName, "syncml:devinf:DevInf") == 0) {
-                /* Get embedded DevInf Document */
-                devinf_doc = wbxml_buffer_create(tree_ctx->input_buff + tree_ctx->skip_start, 
+            if (WBXML_STRCMP(localName, "syncml:devinf:DevInf") == 0 ||
+	        WBXML_STRCMP(localName, "syncml:dmddf1.2:MgmtTree") == 0) {
+                /* Get embedded DevInf or DM DDF Document */
+                embed_doc = wbxml_buffer_create(tree_ctx->input_buff + tree_ctx->skip_start, 
                                                  XML_GetCurrentByteIndex(tree_ctx->xml_parser) - tree_ctx->skip_start,
                                                  XML_GetCurrentByteIndex(tree_ctx->xml_parser) - tree_ctx->skip_start + 10);
+                if (embed_doc == NULL) {
+                    tree_ctx->error = WBXML_ERROR_NOT_ENOUGH_MEMORY;
+                    wbxml_buffer_destroy(embed_doc);
+                    return;
+                }
 
                 if (tree_ctx->expat_utf16) {
                     /** @todo Convert from UTF-16 to UTF-8 */
                 }
 
-                /* Check Buffer Creation and addd </DevInf> ending tag */
-                if ((devinf_doc == NULL) || (!wbxml_buffer_append_cstr(devinf_doc, "</DevInf>")))
+                /* Check Buffer Creation and add the closing tag */
+		if ((WBXML_STRCMP(localName, "syncml:devinf:DevInf") == 0 &&
+		     (!wbxml_buffer_append_cstr(embed_doc, "</DevInf>")))
+                    ||
+		    (WBXML_STRCMP(localName, "syncml:dmddf1.2:MgmtTree") == 0 &&
+		     (!wbxml_buffer_append_cstr(embed_doc, "</MgmtTree>"))))
                 {
                     tree_ctx->error = WBXML_ERROR_NOT_ENOUGH_MEMORY;
-                    wbxml_buffer_destroy(devinf_doc);
+                    wbxml_buffer_destroy(embed_doc);
                     return;
                 }
 
                 /* Add doctype to give the XML parser a chance */
 		const WBXMLLangEntry *lang;
+		if (WBXML_STRCMP(localName, "syncml:dmddf1.2:MgmtTree") == 0 &&
+		    tree_ctx->tree->lang->langID != WBXML_LANG_SYNCML_SYNCML12)
+		{
+                    tree_ctx->error = WBXML_ERROR_UNKNOWN_XML_LANGUAGE;
+                    wbxml_buffer_destroy(embed_doc);
+                    return;
+		}
 		switch(tree_ctx->tree->lang->langID)
 		{
 			case WBXML_LANG_SYNCML_SYNCML10:
@@ -223,7 +246,11 @@
 				lang = wbxml_tables_get_table(WBXML_LANG_SYNCML_DEVINF11);
 				break;
 			case WBXML_LANG_SYNCML_SYNCML12:
-				lang = wbxml_tables_get_table(WBXML_LANG_SYNCML_DEVINF12);
+				if (WBXML_STRCMP(localName, "syncml:dmddf1.2:MgmtTree") == 0) {
+					lang = wbxml_tables_get_table(WBXML_LANG_SYNCML_DMDDF12);
+				} else {
+					lang = wbxml_tables_get_table(WBXML_LANG_SYNCML_DEVINF12);
+				}
 				break;
 			default:
 				tree_ctx->error = WBXML_ERROR_UNKNOWN_XML_LANGUAGE;
@@ -231,28 +258,28 @@
 		}
 
 		/* DOCTYPE in reverse order */
-		if (!wbxml_buffer_insert_cstr(devinf_doc,(WB_UTINY *) "\">\n", 0) ||                     /* > */
-		    !wbxml_buffer_insert_cstr(devinf_doc, (WB_UTINY *) lang->publicID->xmlDTD, 0) ||      /* DTD */
-		    !wbxml_buffer_insert_cstr(devinf_doc, (WB_UTINY *) "\" \"", 0) ||                     /* DTD */
-		    !wbxml_buffer_insert_cstr(devinf_doc, (WB_UTINY *) lang->publicID->xmlPublicID, 0) || /* Public ID */
-		    !wbxml_buffer_insert_cstr(devinf_doc, (WB_UTINY *) " PUBLIC \"", 0) ||                /*  PUBLIC " */
-		    !wbxml_buffer_insert_cstr(devinf_doc, (WB_UTINY *) lang->publicID->xmlRootElt, 0) ||  /* Root Element */
-		    !wbxml_buffer_insert_cstr(devinf_doc, (WB_UTINY *) "<!DOCTYPE ", 0))                  /* <!DOCTYPE */
+		if (!wbxml_buffer_insert_cstr(embed_doc,(WB_UTINY *) "\">\n", 0) ||                     /* > */
+		    !wbxml_buffer_insert_cstr(embed_doc, (WB_UTINY *) lang->publicID->xmlDTD, 0) ||      /* DTD */
+		    !wbxml_buffer_insert_cstr(embed_doc, (WB_UTINY *) "\" \"", 0) ||                     /* DTD */
+		    !wbxml_buffer_insert_cstr(embed_doc, (WB_UTINY *) lang->publicID->xmlPublicID, 0) || /* Public ID */
+		    !wbxml_buffer_insert_cstr(embed_doc, (WB_UTINY *) " PUBLIC \"", 0) ||                /*  PUBLIC " */
+		    !wbxml_buffer_insert_cstr(embed_doc, (WB_UTINY *) lang->publicID->xmlRootElt, 0) ||  /* Root Element */
+		    !wbxml_buffer_insert_cstr(embed_doc, (WB_UTINY *) "<!DOCTYPE ", 0))                  /* <!DOCTYPE */
 		{
 			tree_ctx->error = WBXML_ERROR_ENCODER_APPEND_DATA;
-                	wbxml_buffer_destroy(devinf_doc);
+                	wbxml_buffer_destroy(embed_doc);
 			return;
 		}
 
-                WBXML_DEBUG((WBXML_PARSER, "\t DevInf Doc : '%s'", wbxml_buffer_get_cstr(devinf_doc)));
+                WBXML_DEBUG((WBXML_PARSER, "\t Embedded Doc : '%s'", wbxml_buffer_get_cstr(embed_doc)));
 
                 /* Parse 'DevInf' Document */
-                if ((ret = wbxml_tree_from_xml(wbxml_buffer_get_cstr(devinf_doc),
-                                               wbxml_buffer_len(devinf_doc),
+                if ((ret = wbxml_tree_from_xml(wbxml_buffer_get_cstr(embed_doc),
+                                               wbxml_buffer_len(embed_doc),
                                                &tree)) != WBXML_OK)
                 {
                     tree_ctx->error = ret;
-                    wbxml_buffer_destroy(devinf_doc);
+                    wbxml_buffer_destroy(embed_doc);
                     return;
                 }
 
@@ -264,12 +291,12 @@
                 {
                     tree_ctx->error = WBXML_ERROR_INTERNAL;
                     wbxml_tree_destroy(tree);
-                    wbxml_buffer_destroy(devinf_doc);
+                    wbxml_buffer_destroy(embed_doc);
                     return;
                 }
 
                 /* Clean-up */
-                wbxml_buffer_destroy(devinf_doc);
+                wbxml_buffer_destroy(embed_doc);
                 tree_ctx->skip_lvl = 0;
             }
 #endif /* WBXML_SUPPORT_SYNCML */

[Opensync-commits] r5226 - in trunk/opensync: engine group plugin

[<dg...@su...>]

Author: dgollub
Date: Fri Jan 23 15:07:37 2009
New Revision: 5226
URL: http://www.opensync.org/changeset/5226

Log:
Get rid of hardcoded "data" ObjTypeSink handling. Mixed-objtype should
be now possible not only for "data" objtype.

Modified:
   trunk/opensync/engine/opensync_obj_engine.c
   trunk/opensync/group/opensync_group.c
   trunk/opensync/group/opensync_group_private.h
   trunk/opensync/group/opensync_member.c
   trunk/opensync/plugin/opensync_plugin_info.c

Modified: trunk/opensync/engine/opensync_obj_engine.c
==============================================================================
--- trunk/opensync/engine/opensync_obj_engine.c	Fri Jan 23 14:26:20 2009	(r5225)
+++ trunk/opensync/engine/opensync_obj_engine.c	Fri Jan 23 15:07:37 2009	(r5226)
@@ -771,7 +771,6 @@
 
 osync_bool osync_obj_engine_initialize(OSyncObjEngine *engine, OSyncError **error)
 {
-	osync_bool dummy_sink = FALSE;
 	const char *objtype = NULL;
 	int num = 0;
 	int i = 0;
@@ -788,22 +787,12 @@
 		OSyncObjTypeSink *sink = osync_client_proxy_find_objtype_sink(proxy, objtype);
 		OSyncSinkEngine *sinkengine = NULL; 
 
-		dummy_sink = FALSE;
-
-		if (!sink) {
-			/* "data" sink engine counts also as valid. */
-			sink = osync_client_proxy_find_objtype_sink(proxy, "data"); 
-			if (!sink)
-				continue;
-
-			dummy_sink = TRUE;
-		}
 		
 		sinkengine = osync_sink_engine_new(i, proxy, engine, error);
 		if (!sinkengine)
 			goto error;
 
-		if (dummy_sink)
+		if (!sink) 
 			engine->dummy_sink_engines = osync_list_append(engine->dummy_sink_engines, sinkengine);
 		else
 			engine->active_sink_engines = osync_list_append(engine->active_sink_engines, sinkengine);
@@ -1296,10 +1285,6 @@
 		member = osync_client_proxy_get_member(sinkengine->proxy);
 		objtype_sink = osync_member_find_objtype_sink(member, engine->objtype);
 
-		/* If sink could not be found use "data" sink if available */
-		/* TODO: Get rid of hardcoded-"data". Make this indepdendent of "data" */
-		if (!objtype_sink)
-			objtype_sink = osync_member_find_objtype_sink(member, "data");
 		/* TODO: Review if objtype_sink = NULL is valid at all. */
 		osync_assert(objtype_sink);
 

Modified: trunk/opensync/group/opensync_group.c
==============================================================================
--- trunk/opensync/group/opensync_group.c	Fri Jan 23 14:26:20 2009	(r5225)
+++ trunk/opensync/group/opensync_group.c	Fri Jan 23 15:07:37 2009	(r5226)
@@ -157,21 +157,12 @@
 	}
 }
 
-static void osync_group_add_one(gpointer key, gpointer value, gpointer user_data)
-{
-	GHashTable *table = user_data;
-	
-	int num = GPOINTER_TO_INT(value);
-	g_hash_table_replace(table, key, GINT_TO_POINTER(num + 1));
-}
-
 static GList *osync_group_get_supported_objtypes(OSyncGroup *group)
 {
 	GList *m = NULL;
 	GList *ret = NULL;
 	GHashTable *table = g_hash_table_new(g_str_hash, g_str_equal);
 		
-	int num_data = 0;
 	int i;
 		
 	/* Loop over all members... */
@@ -183,24 +174,13 @@
 			const char *objtype = osync_member_nth_objtype(member, i);
 			if (objtype != NULL) {
 				int num = 0;
-				/* For each objtype, add 1 to the hashtable. If the objtype is
-				 * the special objtype "data", add 1 to all objtypes */
-				if(!strcmp(objtype, "data"))
-					num_data++;
+				/* For each objtype, add 1 to the hashtable. */
 				num = GPOINTER_TO_INT(g_hash_table_lookup(table, objtype));
 				g_hash_table_replace(table, (char *)objtype, GINT_TO_POINTER(num + 1));
 			}
 		}
 	}
 	
-	for (i = 0; i < num_data; i++)
-		g_hash_table_foreach(table, osync_group_add_one, table);
-	
-	if (g_hash_table_size(table) == 0 && num_data >= 2) {
-		osync_trace(TRACE_INTERNAL, "No objtype found yet, but data available");
-		g_hash_table_replace(table, "data", GINT_TO_POINTER(num_data));
-	}
-	
 	g_hash_table_foreach(table, osync_group_build_list, &ret);
 	g_hash_table_destroy(table);
 	return ret;

Modified: trunk/opensync/group/opensync_group_private.h
==============================================================================
--- trunk/opensync/group/opensync_group_private.h	Fri Jan 23 14:26:20 2009	(r5225)
+++ trunk/opensync/group/opensync_group_private.h	Fri Jan 23 15:07:37 2009	(r5226)
@@ -52,8 +52,6 @@
 
 static void osync_group_build_list(gpointer key, gpointer value, gpointer user_data);
 
-static void osync_group_add_one(gpointer key, gpointer value, gpointer user_data);
-
 /** @brief Get list of supported object types of the group
  *
  * @param group The group

Modified: trunk/opensync/group/opensync_member.c
==============================================================================
--- trunk/opensync/group/opensync_member.c	Fri Jan 23 14:26:20 2009	(r5225)
+++ trunk/opensync/group/opensync_member.c	Fri Jan 23 15:07:37 2009	(r5226)
@@ -658,15 +658,8 @@
 {
 	OSyncObjTypeSink *sink = osync_member_find_objtype_sink(member, objtype);
 	if (!sink) {
-		/* FIXME: Nonsnse?!
-			 sink = osync_member_find_objtype_sink(member, "data");
-			 if (!sink) {
-		*/
 		osync_error_set(error, OSYNC_ERROR_GENERIC, "Unable to find objtype %s", objtype);
 		return NULL;
-		/*
-			}
-		*/
 	}
 	
 	return osync_objtype_sink_get_objformat_sinks(sink);

Modified: trunk/opensync/plugin/opensync_plugin_info.c
==============================================================================
--- trunk/opensync/plugin/opensync_plugin_info.c	Fri Jan 23 14:26:20 2009	(r5225)
+++ trunk/opensync/plugin/opensync_plugin_info.c	Fri Jan 23 15:07:37 2009	(r5226)
@@ -155,14 +155,6 @@
 			goto done;
 	}
 	
-	/* If we couldnt find the requested objtype, look if we find a sink
-	 * which accepts any objtype ("data") */
-	for (p = info->objtypes; p; p = p->next) {
-		sink = p->data;
-		if (g_ascii_strcasecmp(osync_objtype_sink_get_name(sink), "data") == 0)
-			goto done;
-	}
-	
 	osync_trace(TRACE_EXIT, "%s: NULL", __func__);
 	return NULL;
 

[Opensync-commits] r5225 - osynctool/trunk/tools

[<dg...@su...>]

Author: dgollub
Date: Fri Jan 23 14:26:20 2009
New Revision: 5225
URL: http://www.opensync.org/changeset/5225

Log:
Port osynctool multiply summary to use
osync_obj_engine_get_mapping_entry_engines_of_member() interface,
instead of the OSyncSinkEngine which i plan to drop from public API.

Modified:
   osynctool/trunk/tools/osynctool.c

Modified: osynctool/trunk/tools/osynctool.c
==============================================================================
--- osynctool/trunk/tools/osynctool.c	Fri Jan 23 14:23:16 2009	(r5224)
+++ osynctool/trunk/tools/osynctool.c	Fri Jan 23 14:26:20 2009	(r5225)
@@ -262,7 +262,7 @@
 
 static void multiply_summary(OSyncEngine *engine, void *user_data)
 {
-	unsigned int o, i, num_objengines, num_sinkengines;
+	unsigned int o, i, num_objengines, num_members;
 	const OSyncList *mapping_entry_engines, *e;
 	osync_bool dirty = FALSE;
 	OSyncError *error = NULL;
@@ -281,12 +281,11 @@
 
 		printf("\nObjType: %s\n", osync_obj_engine_get_objtype(objengine));
 
-		num_sinkengines = osync_obj_engine_num_sinkengines(objengine);
-		for (i=0; i < num_sinkengines; i++) {
-			OSyncSinkEngine *sink_engine = osync_obj_engine_nth_sinkengine(objengine, i);
-			OSyncMember *member = osync_sink_engine_get_member(sink_engine);
+		num_members = osync_obj_engine_num_members(objengine);
+		for (i=0; i < num_members; i++) {
+			OSyncMember *member = osync_obj_engine_nth_member(objengine, i);
 
-			mapping_entry_engines = osync_sink_engine_get_mapping_entry_engines(sink_engine);
+			mapping_entry_engines = osync_obj_engine_get_mapping_entry_engines_of_member(objengine, member);
 
 			added = modified = deleted = 0;
 

[Opensync-commits] r5224 - in trunk: . opensync/engine

[<dg...@su...>]

Author: dgollub
Date: Fri Jan 23 14:23:16 2009
New Revision: 5224
URL: http://www.opensync.org/changeset/5224

Log:
Introduce osync_obj_engine_get_mapping_entry_engines_of_member() to get
list of MappingEntryEngines of a specific member. This makes the
osync_sink_engine_* interface obsolate in the public API.

Example "summary" implenentation in opensnc_engine.c

This fixes #1037

Modified:
   trunk/opensync.sym
   trunk/opensync/engine/opensync_engine.c
   trunk/opensync/engine/opensync_obj_engine.c
   trunk/opensync/engine/opensync_obj_engine.h

Modified: trunk/opensync.sym
==============================================================================
--- trunk/opensync.sym	Fri Jan 23 12:03:19 2009	(r5223)
+++ trunk/opensync.sym	Fri Jan 23 14:23:16 2009	(r5224)
@@ -348,6 +348,7 @@
 osync_obj_engine_command
 osync_obj_engine_event
 osync_obj_engine_finalize
+osync_obj_engine_get_mapping_entry_engines_of_member
 osync_obj_engine_get_objtype
 osync_obj_engine_get_slowsync
 osync_obj_engine_initialize

Modified: trunk/opensync/engine/opensync_engine.c
==============================================================================
--- trunk/opensync/engine/opensync_engine.c	Fri Jan 23 12:03:19 2009	(r5223)
+++ trunk/opensync/engine/opensync_engine.c	Fri Jan 23 14:23:16 2009	(r5224)
@@ -1052,9 +1052,8 @@
 
 void osync_engine_trace_multiply_summary(OSyncEngine *engine)
 {
-	OSyncList *o, *s;
-	OSyncList *e;
-	unsigned int added, modified, deleted, unmodified, unknown;
+	OSyncList *o;
+	unsigned int added, modified, deleted, unmodified, unknown, total, n;
 	long long int memberid;
 
 	if (!osync_trace_is_enabled())
@@ -1068,13 +1067,18 @@
 
 		osync_trace(TRACE_INTERNAL, "ObjEngine: %s", objtype);
 
-		for (s = objengine->sink_engines; s; s = s->next) {
-			OSyncSinkEngine *sinkengine = s->data;
+
+		total = osync_obj_engine_num_members(objengine);
+		for (n = 0; n < total; n++) {
+			OSyncMember *member = osync_obj_engine_nth_member(objengine, n);
+			const OSyncList *entry_engines, *e;
+
+			entry_engines = osync_obj_engine_get_mapping_entry_engines_of_member(objengine, member);
 
 			added = modified = deleted = unmodified = unknown = 0;
-			memberid = osync_member_get_id(osync_client_proxy_get_member(sinkengine->proxy));
+			memberid = osync_member_get_id(member);
 
-			for (e = sinkengine->entries; e; e = e->next) {
+			for (e = entry_engines; e; e = e->next) {
 				OSyncMappingEntryEngine *mapping_entry_engine = e->data;
 
 				if (!mapping_entry_engine->dirty)

Modified: trunk/opensync/engine/opensync_obj_engine.c
==============================================================================
--- trunk/opensync/engine/opensync_obj_engine.c	Fri Jan 23 12:03:19 2009	(r5223)
+++ trunk/opensync/engine/opensync_obj_engine.c	Fri Jan 23 14:23:16 2009	(r5224)
@@ -1230,6 +1230,24 @@
 	return osync_sink_engine_get_member(sinkengine);
 }
 
+const OSyncList *osync_obj_engine_get_mapping_entry_engines_of_member(OSyncObjEngine *engine, OSyncMember *member)
+{
+	OSyncList *s;
+	osync_return_val_if_fail(engine, NULL);
+	osync_return_val_if_fail(member, NULL);
+
+	for (s = engine->active_sink_engines; s; s = s->next) {
+		OSyncSinkEngine *sinkengine = s->data;
+
+		if (member != osync_sink_engine_get_member(sinkengine))
+			continue;
+
+		return osync_sink_engine_get_mapping_entry_engines(sinkengine);
+	}
+
+	return NULL;
+}
+
 osync_bool osync_obj_engine_prepare_write(OSyncObjEngine *engine, OSyncError **error)
 {
 	OSyncList *p;

Modified: trunk/opensync/engine/opensync_obj_engine.h
==============================================================================
--- trunk/opensync/engine/opensync_obj_engine.h	Fri Jan 23 12:03:19 2009	(r5223)
+++ trunk/opensync/engine/opensync_obj_engine.h	Fri Jan 23 14:23:16 2009	(r5224)
@@ -74,5 +74,14 @@
  */
 OSYNC_EXPORT OSyncMember *osync_obj_engine_nth_member(OSyncObjEngine *engine, unsigned int nth);
 
+/*! @brief Get list of OSyncMappingEntryEngines of the OSyncObjEngine 
+ * for a specific member
+ *
+ * @param engine Pointer to an OSyncObjEngine
+ * @param member Pointer to OSyncMember to get OSyncMappingEntryEngine list from
+ * @returns List of OSyncMappingEntryEngines-elements or NULL if there are no Mapping Entry Engines. 
+ */
+OSYNC_EXPORT const OSyncList *osync_obj_engine_get_mapping_entry_engines_of_member(OSyncObjEngine *engine, OSyncMember *member);
+
 #endif /* OPENSYNC_OBJ_ENGINE_H_ */
 

[Opensync-commits] r5223 - in trunk: . opensync/engine

[<dg...@su...>]

Author: dgollub
Date: Fri Jan 23 12:03:19 2009
New Revision: 5223
URL: http://www.opensync.org/changeset/5223

Log:
Introduce osync_obj_engine_{nth,num}_members() interfaces as requested
in #1037

This will make the OSyncSinkEngine interface obsolete

Modified:
   trunk/opensync.sym
   trunk/opensync/engine/opensync_obj_engine.c
   trunk/opensync/engine/opensync_obj_engine.h

Modified: trunk/opensync.sym
==============================================================================
--- trunk/opensync.sym	Fri Jan 23 11:28:04 2009	(r5222)
+++ trunk/opensync.sym	Fri Jan 23 12:03:19 2009	(r5223)
@@ -352,7 +352,9 @@
 osync_obj_engine_get_slowsync
 osync_obj_engine_initialize
 osync_obj_engine_new
+osync_obj_engine_nth_member
 osync_obj_engine_nth_sinkengine
+osync_obj_engine_num_members
 osync_obj_engine_num_sinkengines
 osync_obj_engine_receive_change
 osync_obj_engine_ref

Modified: trunk/opensync/engine/opensync_obj_engine.c
==============================================================================
--- trunk/opensync/engine/opensync_obj_engine.c	Fri Jan 23 11:28:04 2009	(r5222)
+++ trunk/opensync/engine/opensync_obj_engine.c	Fri Jan 23 12:03:19 2009	(r5223)
@@ -1215,6 +1215,21 @@
 	return osync_list_length(engine->mapping_engines);
 }
 
+unsigned int osync_obj_engine_num_members(OSyncObjEngine *engine)
+{
+	osync_assert(engine);
+	return osync_obj_engine_num_sinkengines(engine);
+}
+
+OSyncMember *osync_obj_engine_nth_member(OSyncObjEngine *engine, unsigned int nth)
+{
+	OSyncSinkEngine *sinkengine;
+	osync_return_val_if_fail(engine, NULL);
+	sinkengine = osync_list_nth_data(engine->active_sink_engines, nth);
+	osync_assert(sinkengine);
+	return osync_sink_engine_get_member(sinkengine);
+}
+
 osync_bool osync_obj_engine_prepare_write(OSyncObjEngine *engine, OSyncError **error)
 {
 	OSyncList *p;

Modified: trunk/opensync/engine/opensync_obj_engine.h
==============================================================================
--- trunk/opensync/engine/opensync_obj_engine.h	Fri Jan 23 11:28:04 2009	(r5222)
+++ trunk/opensync/engine/opensync_obj_engine.h	Fri Jan 23 12:03:19 2009	(r5223)
@@ -58,5 +58,21 @@
  */
 OSYNC_EXPORT unsigned int osync_obj_engine_num_sinkengines(OSyncObjEngine *engine);
 
+/*! @brief Get total number of Members
+ *
+ * @param engine Pointer to OSyncObjEngine
+ * @returns Total number of OSyncMember elements 
+ */
+OSYNC_EXPORT unsigned int osync_obj_engine_num_members(OSyncObjEngine *engine);
+
+/** @brief Get the nth OSyncMember of the OSyncObjEngine
+ *
+ * @param engine A pointer to the engine
+ * @param nth The position of the OSyncMember to request
+ * @returns Pointer of the nth OSyncMember, or NULL if not available
+ *
+ */
+OSYNC_EXPORT OSyncMember *osync_obj_engine_nth_member(OSyncObjEngine *engine, unsigned int nth);
+
 #endif /* OPENSYNC_OBJ_ENGINE_H_ */
 

[Opensync-commits] r5222 - in trunk: opensync/archive opensync/engine tests/archive-tests

[<dg...@su...>]

Author: dgollub
Date: Fri Jan 23 11:28:04 2009
New Revision: 5222
URL: http://www.opensync.org/changeset/5222

Log:
Change mapping-table layout, by adding a column "objengine", which
stores the name of the objengine which wrote the change.

Note the difference between "objtype" and "objengine". "objtype" stores
the objtype-name of the change which got written. "objengine" stores the
objtype-name of the Object Engine which wrote the change. This is
required due to mixed-objtype syncing, which is limited to encapuslated
formats.

This chance requires to drop the mapping-table.
(TODO: implement table-drop function)
To reuse existing groups:

osyncdump $gorup --reset
rm ~/.opensync/groupX/archive.db

Introduced osync_engine_slowsync_for_mixed_objengines() which propagtes
slow-sync to all mixed-synced object engines if required. This is done
after all plugins completed connect_done().

Modified:
   trunk/opensync/archive/opensync_archive.c
   trunk/opensync/archive/opensync_archive_internals.h
   trunk/opensync/engine/opensync_engine.c
   trunk/opensync/engine/opensync_engine_internals.h
   trunk/opensync/engine/opensync_obj_engine.c
   trunk/opensync/engine/opensync_sink_engine.c
   trunk/tests/archive-tests/check_archive.c

Modified: trunk/opensync/archive/opensync_archive.c
==============================================================================
--- trunk/opensync/archive/opensync_archive.c	Fri Jan 23 00:51:13 2009	(r5221)
+++ trunk/opensync/archive/opensync_archive.c	Fri Jan 23 11:28:04 2009	(r5222)
@@ -59,7 +59,7 @@
 	 * enforce the entity-relationship model all queries for a single item should be:
 	 * WHERE objtype='%s' AND id=%lli
 	 */ 
-	query = "CREATE TABLE tbl_changes (objtype VARCHAR(64) NOT NULL, id INTEGER PRIMARY KEY AUTOINCREMENT, uid VARCHAR NOT NULL, memberid INTEGER NOT NULL, mappingid INTEGER NOT NULL )";
+	query = "CREATE TABLE tbl_changes (objtype VARCHAR(64) NOT NULL, id INTEGER PRIMARY KEY AUTOINCREMENT, uid VARCHAR NOT NULL, memberid INTEGER NOT NULL, mappingid INTEGER NOT NULL, objengine VARCHAR(64) NOT NULL )";
 	if (!osync_db_query(db, query, error)) {
 		goto error;
 	}
@@ -282,30 +282,35 @@
 	return -1;
 }
 
-long long int osync_archive_save_change(OSyncArchive *archive, long long int id, const char *uid, const char *objtype, long long int mappingid, long long int memberid, OSyncError **error)
+long long int osync_archive_save_change(OSyncArchive *archive, long long int id, const char *uid, const char *objtype, long long int mappingid, long long int memberid, const char *objengine, OSyncError **error)
 {
 	char *query = NULL;
 	char *escaped_uid = NULL;
 	char *escaped_objtype = NULL;
+	char *escaped_objengine = NULL;
 
-	osync_trace(TRACE_ENTRY, "%s(%p, %lli, %s, %s, %lli, %lli, %p)", __func__, archive, id, uid, objtype, mappingid, memberid, error);
+	osync_trace(TRACE_ENTRY, "%s(%p, %lli, %s, %s, %lli, %lli, %p, %s)", __func__, archive, id, uid, objtype, mappingid, memberid, __NULLSTR(objengine), error);
 	osync_assert(archive);
 	osync_assert(uid);
 	osync_assert(objtype);
+	osync_assert(objengine);
 
 	if (!osync_archive_create_changes(archive->db, objtype, error))
 		goto error;
 
 	escaped_uid = osync_db_sql_escape(uid);
 	escaped_objtype = osync_db_sql_escape(objtype);
+	escaped_objengine = osync_db_sql_escape(objengine);
 
 	if (!id) {
-		query = osync_strdup_printf("INSERT INTO tbl_changes (objtype, uid, mappingid, memberid) VALUES('%s', '%s', '%lli', '%lli')", escaped_objtype, escaped_uid, mappingid, memberid);
+		query = osync_strdup_printf("INSERT INTO tbl_changes (objtype, uid, mappingid, memberid, objengine) VALUES('%s', '%s', '%lli', '%lli', '%s')", escaped_objtype, escaped_uid, mappingid, memberid, objengine);
 	} else {
-		query = osync_strdup_printf("UPDATE tbl_changes SET uid='%s', mappingid='%lli', memberid='%lli' WHERE objtype='%s' AND id=%lli", escaped_uid, mappingid, memberid, escaped_objtype, id);
+		query = osync_strdup_printf("UPDATE tbl_changes SET uid='%s', mappingid='%lli', memberid='%lli', objengine='%s' WHERE objtype='%s' AND id=%lli", escaped_uid, mappingid, memberid, escaped_objengine, escaped_objtype, id);
 	}
+	osync_free(escaped_objengine);
 	osync_free(escaped_objtype);
 	osync_free(escaped_uid);
+	escaped_objengine = NULL;
 	escaped_objtype = NULL;
 	escaped_uid = NULL;
 	
@@ -583,3 +588,43 @@
 	return FALSE;
 }
 
+osync_bool osync_archive_get_mixed_objengines(OSyncArchive *archive, const char *objengine, OSyncList **objengines, OSyncError **error)
+{
+	char *query = NULL, *escaped_objengine, *objengine_name;
+	OSyncList *result, *row, *column;
+
+	osync_assert(archive);
+	osync_assert(objengine);
+	osync_assert(objengines);
+
+	escaped_objengine = osync_db_sql_escape(objengine);
+
+	query = osync_strdup_printf("SELECT DISTINCT(b.objengine) FROM tbl_changes, tbl_changes as a, tbl_changes as b "
+			"WHERE a.mappingid == b.mappingid AND a.objengine == '%s';", escaped_objengine);
+	result = osync_db_query_table(archive->db, query, error);
+	osync_free(query);
+	
+	/* Check for error of osync_db_query_table() call. */
+	if (osync_error_is_set(error))
+		goto error;
+
+	for (row = result; row; row = row->next) { 
+		column = row->data;
+
+		objengine_name = osync_list_nth_data(column, 0);
+		if (!objengine_name) {
+			osync_error_set(error, OSYNC_ERROR_GENERIC, "Database table tbl_changes corrupt. Couldn't query for mixed object engines.");
+			goto error;
+		}
+		
+		*objengines = osync_list_append((*objengines), osync_strdup(objengine_name));
+	}
+
+	osync_db_free_list(result);	
+
+
+	return TRUE;
+error:
+	return FALSE;
+}
+

Modified: trunk/opensync/archive/opensync_archive_internals.h
==============================================================================
--- trunk/opensync/archive/opensync_archive_internals.h	Fri Jan 23 00:51:13 2009	(r5221)
+++ trunk/opensync/archive/opensync_archive_internals.h	Fri Jan 23 11:28:04 2009	(r5222)
@@ -71,10 +71,11 @@
  * @param objtype Reported object type of entry
  * @param mappingid Mapped ID of entry 
  * @param memberid ID of member which reported entry 
+ * @param objengine Object Engine which handles the change
  * @param error Pointer to an error struct
  * @return Returns number of entries in archive group database. 0 on error. 
  */
-OSYNC_TEST_EXPORT long long int osync_archive_save_change(OSyncArchive *archive, long long int id, const char *uid, const char *objtype, long long int mappingid, long long int memberid, OSyncError **error);
+OSYNC_TEST_EXPORT long long int osync_archive_save_change(OSyncArchive *archive, long long int id, const char *uid, const char *objtype, long long int mappingid, long long int memberid, const char *objengine, OSyncError **error);
 
 /**
  * @brief Deletes an entry from a group archive.
@@ -130,6 +131,8 @@
  * @return Returns TRUE on success, FALSE otherwise 
  */
 osync_bool osync_archive_flush_ignored_conflict(OSyncArchive *archive, const char *objtype, OSyncError **error);
+
+osync_bool osync_archive_get_mixed_objengines(OSyncArchive *archive, const char *objengine, OSyncList **objengines, OSyncError **error);
 /*@}*/
 
 #endif /*OPENSYNC_ARCHIVE_INTERNALS_H_*/

Modified: trunk/opensync/engine/opensync_engine.c
==============================================================================
--- trunk/opensync/engine/opensync_engine.c	Fri Jan 23 00:51:13 2009	(r5221)
+++ trunk/opensync/engine/opensync_engine.c	Fri Jan 23 11:28:04 2009	(r5222)
@@ -828,6 +828,41 @@
 	return FALSE;
 }
 
+osync_bool osync_engine_slowsync_for_mixed_objengines(OSyncEngine *engine, OSyncError **error)
+{
+	OSyncList *mixedengines = NULL, *o, *m;
+	const char *objengine_objtype;
+
+	for (o = engine->object_engines; o; o = o->next) {
+		OSyncObjEngine *objengine = o->data;
+
+		if (!osync_obj_engine_get_slowsync(objengine))
+			continue;
+
+		objengine_objtype = osync_obj_engine_get_objtype(objengine);
+		if (!osync_archive_get_mixed_objengines(engine->archive, objengine_objtype, &mixedengines, error))
+			goto error;
+	}
+
+	for (o = engine->object_engines; o; o = o->next) {
+		OSyncObjEngine *objengine = o->data;
+
+		for (m = mixedengines; m; m = m->next) {
+
+			if (strcmp(osync_obj_engine_get_objtype(objengine), (char*)m->data)) 
+				continue;
+
+			osync_obj_engine_set_slowsync(objengine, TRUE);
+			break;
+		}
+		
+	}
+	
+	return TRUE;
+error:
+	return FALSE;
+}
+
 static void _osync_engine_generate_connect_done_event(OSyncEngine *engine)
 {
 	if (osync_bitcount(engine->proxy_errors | engine->proxy_connect_done) != osync_list_length(engine->proxies))
@@ -841,6 +876,8 @@
 			osync_status_update_engine(engine, OSYNC_ENGINE_EVENT_ERROR, locerror);
 			osync_engine_event(engine, OSYNC_ENGINE_EVENT_ERROR);
 		} else {
+			/* TODO error handling */
+			osync_engine_slowsync_for_mixed_objengines(engine, NULL);
 			osync_status_update_engine(engine, OSYNC_ENGINE_EVENT_CONNECT_DONE, NULL);
 			osync_engine_event(engine, OSYNC_ENGINE_EVENT_CONNECT_DONE);
 		}

Modified: trunk/opensync/engine/opensync_engine_internals.h
==============================================================================
--- trunk/opensync/engine/opensync_engine_internals.h	Fri Jan 23 00:51:13 2009	(r5221)
+++ trunk/opensync/engine/opensync_engine_internals.h	Fri Jan 23 11:28:04 2009	(r5222)
@@ -130,6 +130,20 @@
  */
 osync_bool osync_engine_handle_mixed_objtypes(OSyncEngine *engine, OSyncError **error);
 
+/** @brief Trigger slow-sync for object engines which have mixed object types synced
+ *
+ * This function checks if an object engine has requested a slow-sync and if
+ * further function need to run a slow-sync. Due to mixed object type syncing.
+ * Mixed object type syncing get detected by checking the mapping table if different
+ * object engine have written non-native object types.
+ *  
+ * @param engine Pointer to engine
+ * @param error Pointer to error-struct which get set on any error
+ * @returns TRUE on success, or FALSE on any error
+ *
+ */
+osync_bool osync_engine_slowsync_for_mixed_objengines(OSyncEngine *engine, OSyncError **error);
+
 /*@}*/
 
 #endif /*OPENSYNC_ENGINE_INTERNALS_H_*/

Modified: trunk/opensync/engine/opensync_obj_engine.c
==============================================================================
--- trunk/opensync/engine/opensync_obj_engine.c	Fri Jan 23 00:51:13 2009	(r5221)
+++ trunk/opensync/engine/opensync_obj_engine.c	Fri Jan 23 11:28:04 2009	(r5222)
@@ -481,8 +481,10 @@
 	OSyncMember *member = NULL;
 	OSyncMappingEntry *entry = NULL;
 	const char *objtype = NULL;
+	const char *objengine_objtype = NULL;
 	long long int id = 0;
 
+	objengine_objtype = osync_obj_engine_get_objtype(engine);
 	
 	osync_trace(TRACE_ENTRY, "%s(%p, %p, %s, %p)", __func__, proxy, userdata, uid, error);
 	
@@ -516,7 +518,7 @@
 		} else {
 
 			/* TODO error handling */
-			osync_archive_save_change(engine->archive, id, osync_change_get_uid(entry_engine->change), objtype, osync_mapping_get_id(mapping), osync_member_get_id(member), &locerror);
+			osync_archive_save_change(engine->archive, id, osync_change_get_uid(entry_engine->change), objtype, osync_mapping_get_id(mapping), osync_member_get_id(member), objengine_objtype, &locerror);
 		}
 	}
 

Modified: trunk/opensync/engine/opensync_sink_engine.c
==============================================================================
--- trunk/opensync/engine/opensync_sink_engine.c	Fri Jan 23 00:51:13 2009	(r5221)
+++ trunk/opensync/engine/opensync_sink_engine.c	Fri Jan 23 11:28:04 2009	(r5222)
@@ -255,7 +255,7 @@
 							osync_change_get_uid(entry_engine->change),
 							osync_change_get_objtype(entry_engine->change),
 							osync_mapping_get_id(mapping),
-							osync_member_get_id(member), error))
+							osync_member_get_id(member), objtype, error))
 					goto error;
 			}
 		}

Modified: trunk/tests/archive-tests/check_archive.c
==============================================================================
--- trunk/tests/archive-tests/check_archive.c	Fri Jan 23 00:51:13 2009	(r5221)
+++ trunk/tests/archive-tests/check_archive.c	Fri Jan 23 11:28:04 2009	(r5222)
@@ -59,7 +59,7 @@
 	OSyncList *memberids;
 	osync_archive_load_changes(archive, "contact", &ids, &uids, &mappingids, &memberids, &error);
 	
-	long long int id = osync_archive_save_change(archive, 0, "uid", "contact", 1, 1, &error);
+	long long int id = osync_archive_save_change(archive, 0, "uid", "contact", 1, 1, "contact", &error);
 	fail_unless(id != 0, NULL);
 	fail_unless(error == NULL, NULL);
 		
@@ -84,7 +84,7 @@
 	OSyncList *memberids;
 	osync_archive_load_changes(archive, "contact", &ids, &uids, &mappingids, &memberids, &error);
 	
-	long long int id = osync_archive_save_change(archive, 0, "uid", "contact", 1, 1, &error);
+	long long int id = osync_archive_save_change(archive, 0, "uid", "contact", 1, 1, "contact", &error);
 	fail_unless(id != 0, NULL);
 	fail_unless(error == NULL, NULL);
 	
@@ -114,7 +114,7 @@
 	OSyncList *memberids;
 	osync_archive_load_changes(archive, "contact", &ids, &uids, &mappingids, &memberids, &error);
 	
-	long long int id = osync_archive_save_change(archive, 0, "uid", "contact", 1, 1, &error);
+	long long int id = osync_archive_save_change(archive, 0, "uid", "contact", 1, 1, "contact", &error);
 	fail_unless(id != 0, NULL);
 	fail_unless(error == NULL, NULL);
 	
@@ -153,7 +153,7 @@
 	OSyncList *memberids;
 	osync_archive_load_changes(archive, "contact", &ids, &uids, &mappingids, &memberids, &error);
 	
-	long long int id = osync_archive_save_change(archive, 0, "uid", "contact", 1, 1, &error);
+	long long int id = osync_archive_save_change(archive, 0, "uid", "contact", 1, 1, "contact", &error);
 	fail_unless(id != 0, NULL);
 	fail_unless(error == NULL, NULL);
 	

[Opensync-commits] r5221 - in branches/timeout: . tests tests/ipc-tests

[<dg...@su...>]

Author: Graham Cobb
Date: Fri Jan 23 00:51:13 2009
New Revision: 5221
URL: http://www.opensync.org/changeset/5221

Log:
Unit tests for the new timeout code.

Tasks completed:

1) Created unit tests.  All tests pass.
5) Timeouts that occur while a plugin thread is sleeping work fine, but 
thread is not interrupted.  Nothing more planned.

This leaves (at least) the following tasks:

2) Limit the number of pending entries so the timeout is not
dependent on the number of outstanding requests
3) Add references for cross-linked pointers between queues and
remove linkages (and unref) when ??? (queue disconnected?)
4) Handle timeout before reply queue set up by generating an
error message on the command queue
6) Implement a mechanism to protect the queue itself: i.e. is the
receiver still reading the queue?

Modified:
   branches/timeout/ChangeLog
   branches/timeout/tests/CMakeLists.txt
   branches/timeout/tests/ipc-tests/check_ipc.c

Modified: branches/timeout/ChangeLog
==============================================================================
--- branches/timeout/ChangeLog	Thu Jan 22 23:01:49 2009	(r5220)
+++ branches/timeout/ChangeLog	Fri Jan 23 00:51:13 2009	(r5221)
@@ -1,5 +1,16 @@
 2009-01-22  Graham Cobb  <g+...@co...>
 
+	* tests/CMakeLists.txt: Add ipc_loop_with_timeout
+	Add ipc_late_reply
+
+	* tests/ipc-tests/check_ipc.c: Call osync_queue_cross_link every time
+	client mainloop is setup.
+	(ipc_timeout): Rewrite client to use a mainloop and a callback so new 
+	timeout processing can run
+	Rewrite server to use a mainloop so message callback gets called
+	(ipc_loop_with_timeout): Add ipc_loop_with_timeout
+	(ipc_late_reply): Add ipc_late_reply
+
 	* opensync/ipc/opensync_queue.c (_timeout_dispatch): Restart timeout search at start of
 	pending list as list may have been modified while unlocked.
 	(_timeout_dispatch): Set message id in error message

Modified: branches/timeout/tests/CMakeLists.txt
==============================================================================
--- branches/timeout/tests/CMakeLists.txt	Thu Jan 22 23:01:49 2009	(r5220)
+++ branches/timeout/tests/CMakeLists.txt	Fri Jan 23 00:51:13 2009	(r5221)
@@ -246,6 +246,8 @@
 OSYNC_TESTCASE(ipc ipc_pipes_stress)
 OSYNC_TESTCASE_DISABLED(ipc ipc_callback_break_pipes "1039")
 OSYNC_TESTCASE(ipc ipc_timeout)
+OSYNC_TESTCASE(ipc ipc_loop_with_timeout)
+OSYNC_TESTCASE(ipc ipc_late_reply)
 ENDIF (NOT WIN32)
 
 BUILD_CHECK_TEST( mapping mapping-tests/check_mapping.c ${TEST_TARGET_LIBRARIES} )

Modified: branches/timeout/tests/ipc-tests/check_ipc.c
==============================================================================
--- branches/timeout/tests/ipc-tests/check_ipc.c	Thu Jan 22 23:01:49 2009	(r5220)
+++ branches/timeout/tests/ipc-tests/check_ipc.c	Fri Jan 23 00:51:13 2009	(r5221)
@@ -1087,6 +1087,8 @@
 		osync_assert(error == NULL);
 		osync_assert(osync_queue_connect(server_queue, OSYNC_QUEUE_SENDER, &error));
 		osync_assert(error == NULL);
+
+		osync_queue_cross_link(client_queue, server_queue);
 		
 		message = osync_queue_get_message(server_queue);
 		
@@ -1272,6 +1274,8 @@
 		osync_assert(error == NULL);
 		osync_assert(osync_queue_connect(server_queue, OSYNC_QUEUE_SENDER, &error));
 		osync_assert(error == NULL);
+
+		osync_queue_cross_link(client_queue, server_queue);
 		
 		message = osync_queue_get_message(server_queue);
 		
@@ -1450,6 +1454,8 @@
 
 		osync_assert(osync_queue_connect(server_queue, OSYNC_QUEUE_SENDER, &error));
 		osync_assert(error == NULL);
+
+		osync_queue_cross_link(client_queue, server_queue);
 		
 		message = osync_queue_get_message(server_queue);
 		
@@ -1644,6 +1650,8 @@
 
 		osync_assert(osync_queue_connect(server_queue, OSYNC_QUEUE_SENDER, &error));
 		osync_assert(error == NULL);
+
+		osync_queue_cross_link(client_queue, server_queue);
 		
 		while (osync_queue_is_connected(client_queue)) { g_usleep(100); }
 		
@@ -1849,6 +1857,8 @@
 
 		osync_assert(osync_queue_connect(server_queue, OSYNC_QUEUE_SENDER, &error));
 		osync_assert(error == NULL);
+
+		osync_queue_cross_link(client_queue, server_queue);
 		
 		message = osync_queue_get_message(server_queue);
 		if (osync_message_get_command(message) != OSYNC_MESSAGE_QUEUE_HUP) {
@@ -1984,6 +1994,8 @@
 
 		osync_assert(osync_queue_connect(server_queue, OSYNC_QUEUE_SENDER, &error));
 		osync_assert(error == NULL);
+
+		osync_queue_cross_link(client_queue, server_queue);
 		
 		while (osync_queue_is_connected(client_queue)) { g_usleep(100); }
 		
@@ -2081,11 +2093,39 @@
 		num_callback_timeout++;
 	else
 		num_callback++;
+}
 
+char *data5 = "this is another test string";
+void client_handler5(OSyncMessage *message, void *user_data)
+{
+	osync_trace(TRACE_ENTRY, "%s(%p, %p)", __func__, message, user_data);
 	OSyncError *error = NULL;
-	OSyncMessage *reply = osync_message_new_reply(message, &error);
-	osync_queue_send_message(client_queue, NULL, reply, &error);
-	osync_message_unref(reply);
+	
+	osync_assert(GPOINTER_TO_INT(user_data) ==1);
+	osync_assert(osync_message_get_command(message) == OSYNC_MESSAGE_INITIALIZE);
+		
+	int int1;
+	long long int longint1;
+	char *string;
+	char databuf[strlen(data5) + 1];
+	
+	
+	osync_message_read_int(message, &int1);
+	osync_message_read_string(message, &string);
+	osync_message_read_long_long_int(message, &longint1);
+	osync_message_read_data(message, databuf, strlen(data5) + 1);
+	
+	osync_assert(int1 == 4000000);
+	osync_assert(!strcmp(string, "this is a test string"));
+	osync_assert(longint1 == 400000000);
+	osync_assert(!strcmp(databuf, data5));
+	
+	/* TIMEOUT TIMEOUT TIMEOUT (no reply...) */
+	
+	/* Proper code would reply to this message, but for testing
+	   purposes we don't reply and simulate a "timeout" situation */
+		
+	osync_trace(TRACE_EXIT, "%s", __func__);
 }
 
 START_TEST (ipc_timeout)
@@ -2116,78 +2156,245 @@
 	
 	osync_queue_create(client_queue, &error);
 	fail_unless(error == NULL, NULL);
-	char *data = "this is another test string";
 	
 	pid_t cpid = fork();
 	if (cpid == 0) { //Child
-		g_usleep(1*G_USEC_PER_SEC);
+		
+		GMainContext *context = g_main_context_new();
+		OSyncThread *thread = osync_thread_new(context, &error);
+	
+		osync_queue_set_message_handler(client_queue, client_handler5, GINT_TO_POINTER(1));
+		
+		osync_queue_setup_with_gmainloop(client_queue, context);
+		
+		osync_thread_start(thread);
+
 		osync_assert(osync_queue_connect(client_queue, OSYNC_QUEUE_RECEIVER, &error));
 		osync_assert(error == NULL);
 
 		osync_assert(osync_queue_connect(server_queue, OSYNC_QUEUE_SENDER, &error));
 		osync_assert(error == NULL);
+
+		osync_queue_cross_link(client_queue, server_queue);
 		
-		while (!(message = osync_queue_get_message(client_queue))) {
-			g_usleep(10000);
-		}
+		message = osync_queue_get_message(server_queue);
 		
-		if (osync_message_get_command(message) != OSYNC_MESSAGE_INITIALIZE) {
+		if (osync_message_get_command(message) != OSYNC_MESSAGE_QUEUE_HUP) {
 			exit (1);
 		}
 		
-		int int1;
-		long long int longint1;
-		char *string;
-		char databuf[strlen(data) + 1];
-			
-		osync_message_read_int(message, &int1);
-		osync_message_read_string(message, &string);
-		osync_message_read_long_long_int(message, &longint1);
-		osync_message_read_data(message, databuf, strlen(data) + 1);
+		osync_message_unref(message);
+	
+		if (osync_queue_disconnect(server_queue, &error) != TRUE || error != NULL)
+			exit(1);
+		osync_queue_unref(server_queue);
 		
-		osync_assert(int1 == 4000000);
-		osync_assert(!strcmp(string, "this is a test string"));
+		osync_assert(osync_queue_disconnect(client_queue, &error));
+		osync_assert(error == NULL);
+		
+		osync_thread_stop(thread);
+		osync_thread_unref(thread);
+		
+		osync_queue_unref(client_queue);
+		
+		g_free(testbed);
+		
+		exit(0);
+	} else {
+		GMainContext *context = g_main_context_new();
+		OSyncThread *thread = osync_thread_new(context, &error);
+		
+		osync_queue_set_message_handler(server_queue, server_handler4, GINT_TO_POINTER(1));
+		
+		osync_queue_setup_with_gmainloop(server_queue, context);
+		
+		osync_thread_start(thread);
 
-		osync_assert(longint1 == 400000000);
-		osync_assert(!strcmp(databuf, "this is another test string"));
+		fail_unless(osync_queue_connect(client_queue, OSYNC_QUEUE_SENDER, &error), NULL);
+		fail_unless(error == NULL, NULL);
+		
+		fail_unless(osync_queue_connect(server_queue, OSYNC_QUEUE_RECEIVER, &error), NULL);
+		fail_unless(error == NULL, NULL);
 		
-		/* TIMEOUT TIMEOUT TIMEOUT (no reply...) */
+		message = osync_message_new(OSYNC_MESSAGE_INITIALIZE, 0, &error);
+		fail_unless(message != NULL, NULL);
+		fail_unless(!osync_error_is_set(&error), NULL);
 
-		/* Proper code would reply to this message, but for testing
-		   purposes we don't reply and simulate a "timeout" situation */
-			
+		osync_message_set_handler(message, _message_handler, NULL);
+		
+		osync_message_write_int(message, 4000000);
+		osync_message_write_string(message, "this is a test string");
+		osync_message_write_long_long_int(message, 400000000);
+		osync_message_write_data(message, data5, strlen(data5) + 1);
+		
+		// Send with timeout of one second
+		fail_unless(osync_queue_send_message_with_timeout(client_queue, server_queue, message, 1, &error), NULL);
+		fail_unless(!osync_error_is_set(&error), NULL);
+
+		osync_message_unref(message);
+
+		// Block
+		g_usleep(5*G_USEC_PER_SEC);
+		
+		osync_queue_disconnect(server_queue, &error);
+		fail_unless(error == NULL, NULL);
+		
 		while (!(message = osync_queue_get_message(client_queue))) {
 			g_usleep(10000);
 		}
+		
+		if (osync_message_get_command(message) != OSYNC_MESSAGE_QUEUE_HUP) {
+			exit (1);
+		}
+		osync_message_unref(message);
+		
+		osync_queue_disconnect(client_queue, &error);
+		fail_unless(error == NULL, NULL);
+		
+		int status = 0;
+		wait(&status);
+		fail_unless(WEXITSTATUS(status) == 0, NULL);
+	}
+	
+	fail_unless(osync_testing_file_exists("/tmp/testpipe-client") == TRUE, NULL);
+	
+	fail_unless(osync_queue_remove(client_queue, &error), NULL);
+	fail_unless(osync_queue_remove(server_queue, &error), NULL);
+	fail_unless(!osync_error_is_set(&error), NULL);
 
-		OSyncMessage *reply = osync_message_new_reply(message, &error);
+	/* Check if the timeout handler replied with an error */
+	fail_unless(num_callback_timeout == 1, NULL);
+	fail_unless(num_callback == 0, NULL);
 	
-		osync_queue_send_message(server_queue, NULL, reply, &error);
+	fail_unless(osync_testing_file_exists("/tmp/testpipe-client") == FALSE, NULL);
+
+	osync_queue_unref(client_queue);
+	osync_queue_unref(server_queue);
+	
+	destroy_testbed(testbed);
+}
+END_TEST
 
+void client_handler6(OSyncMessage *message, void *user_data)
+{
+	osync_trace(TRACE_ENTRY, "%s(%p, %p)", __func__, message, user_data);
+	OSyncError *error = NULL;
+	
+	osync_assert(GPOINTER_TO_INT(user_data) ==1);
+	osync_assert(osync_message_get_command(message) == OSYNC_MESSAGE_INITIALIZE);
 		
-		if (osync_queue_disconnect(client_queue, &error) != TRUE || error != NULL)
-			exit(1);
-		osync_queue_unref(client_queue);
+	int int1;
+	long long int longint1;
+	char *string;
+	char databuf[strlen(data5) + 1];
 	
-		while (!(message = osync_queue_get_message(server_queue))) {
-			g_usleep(10000);
-		}
+	
+	osync_message_read_int(message, &int1);
+	osync_message_read_string(message, &string);
+	osync_message_read_long_long_int(message, &longint1);
+	osync_message_read_data(message, databuf, strlen(data5) + 1);
+	
+	osync_assert(int1 == 4000000);
+	osync_assert(!strcmp(string, "this is a test string"));
+	osync_assert(longint1 == 400000000);
+	osync_assert(!strcmp(databuf, data5));
+
+	// Do some time consuming processing
+	g_usleep(3*G_USEC_PER_SEC);
+	
+	OSyncMessage *reply = osync_message_new_reply(message, &error);
+	
+	osync_queue_send_message(server_queue, NULL, reply, &error);
+	
+	osync_message_unref(reply);
+		
+	osync_trace(TRACE_EXIT, "%s", __func__);
+}
+
+START_TEST (ipc_late_reply)
+{	
+	/* This testcase is inteded to test osync_queue_send_message_with_timeout().
+	   Client got forked and listens for messages from Server and replies.
+
+	   To simulate a "timeout" situation the Client delays 3 seconds and then replies
+
+	   The timeout handler will call the _message_handler() with an error and the late reply
+	   will be discarded.
+
+	   JFYI, every timed out message calls the callback/message_handler with an (timeout) error.
+	   */
+
+	char *testbed = setup_testbed(NULL);
+	osync_testing_file_remove("/tmp/testpipe-server");
+	osync_testing_file_remove("/tmp/testpipe-client");
+
+	num_callback_timeout = 0;
+	num_callback = 0;
+	
+	OSyncError *error = NULL;
+	server_queue = osync_queue_new("/tmp/testpipe-server", &error);
+	client_queue = osync_queue_new("/tmp/testpipe-client", &error);
+	OSyncMessage *message = NULL;
+	
+	osync_queue_create(server_queue, &error);
+	fail_unless(error == NULL, NULL);
+	
+	osync_queue_create(client_queue, &error);
+	fail_unless(error == NULL, NULL);
+	
+	pid_t cpid = fork();
+	if (cpid == 0) { //Child
+		
+		GMainContext *context = g_main_context_new();
+		OSyncThread *thread = osync_thread_new(context, &error);
+	
+		osync_queue_set_message_handler(client_queue, client_handler6, GINT_TO_POINTER(1));
+		
+		osync_queue_setup_with_gmainloop(client_queue, context);
+		
+		osync_thread_start(thread);
+
+		osync_assert(osync_queue_connect(client_queue, OSYNC_QUEUE_RECEIVER, &error));
+		osync_assert(error == NULL);
+
+		osync_assert(osync_queue_connect(server_queue, OSYNC_QUEUE_SENDER, &error));
+		osync_assert(error == NULL);
+
+		osync_queue_cross_link(client_queue, server_queue);
+		
+		message = osync_queue_get_message(server_queue);
 		
 		if (osync_message_get_command(message) != OSYNC_MESSAGE_QUEUE_HUP) {
 			exit (1);
 		}
-	
-		osync_message_unref(message);
-		g_usleep(1*G_USEC_PER_SEC);
 		
+		osync_message_unref(message);
+	
 		if (osync_queue_disconnect(server_queue, &error) != TRUE || error != NULL)
 			exit(1);
 		osync_queue_unref(server_queue);
 		
+		osync_assert(osync_queue_disconnect(client_queue, &error));
+		osync_assert(error == NULL);
+		
+		osync_thread_stop(thread);
+		osync_thread_unref(thread);
+		
+		osync_queue_unref(client_queue);
+		
 		g_free(testbed);
 		
 		exit(0);
 	} else {
+		GMainContext *context = g_main_context_new();
+		OSyncThread *thread = osync_thread_new(context, &error);
+		
+		osync_queue_set_message_handler(server_queue, server_handler4, GINT_TO_POINTER(1));
+		
+		osync_queue_setup_with_gmainloop(server_queue, context);
+		
+		osync_thread_start(thread);
+
 		fail_unless(osync_queue_connect(client_queue, OSYNC_QUEUE_SENDER, &error), NULL);
 		fail_unless(error == NULL, NULL);
 		
@@ -2203,7 +2410,7 @@
 		osync_message_write_int(message, 4000000);
 		osync_message_write_string(message, "this is a test string");
 		osync_message_write_long_long_int(message, 400000000);
-		osync_message_write_data(message, data, strlen(data) + 1);
+		osync_message_write_data(message, data5, strlen(data5) + 1);
 		
 		// Send with timeout of one second
 		fail_unless(osync_queue_send_message_with_timeout(client_queue, server_queue, message, 1, &error), NULL);
@@ -2212,13 +2419,7 @@
 		osync_message_unref(message);
 
 		// Block
-		while (!(message = osync_queue_get_message(server_queue))) {
-			g_usleep(100000);
-		}
-		
-		fail_unless(osync_message_get_command(message) == OSYNC_MESSAGE_REPLY);
-		
-		osync_message_unref(message);
+		g_usleep(5*G_USEC_PER_SEC);
 		
 		osync_queue_disconnect(server_queue, &error);
 		fail_unless(error == NULL, NULL);
@@ -2259,6 +2460,199 @@
 }
 END_TEST
 
+void callback_handler7(OSyncMessage *message, void *user_data)
+{
+	osync_trace(TRACE_ENTRY, "%s(%p, %p)", __func__, message, user_data);
+	OSyncError *error = NULL;
+	
+	osync_assert(GPOINTER_TO_INT(user_data) == 1);
+	
+	num_msgs++;
+	osync_assert(osync_message_get_command(message) == OSYNC_MESSAGE_REPLY);
+	
+	if (num_msgs >= req_msgs) {
+		osync_queue_disconnect(server_queue, &error);
+		osync_assert(error == NULL);
+	}
+	
+	osync_trace(TRACE_EXIT, "%s", __func__);
+}
+
+void server_handler7(OSyncMessage *message, void *user_data)
+{
+	abort();
+}
+
+void client_handler7(OSyncMessage *message, void *user_data)
+{
+	osync_trace(TRACE_ENTRY, "%s(%p, %p)", __func__, message, user_data);
+	OSyncError *error = NULL;
+	
+	osync_assert(GPOINTER_TO_INT(user_data) ==1);
+	osync_assert(osync_message_get_command(message) == OSYNC_MESSAGE_INITIALIZE);
+	
+	int int1;
+	long long int longint1;
+	char *string;
+	void *databuf;
+	
+	osync_message_read_int(message, &int1);
+	osync_message_read_const_string(message, &string);
+	osync_message_read_long_long_int(message, &longint1);
+	osync_message_read_const_data(message, &databuf, strlen("this is another test string") + 1);
+	
+	osync_assert(int1 == 4000000);
+	osync_assert(!strcmp(string, "this is a test string"));
+	osync_assert(longint1 == 400000000);
+	osync_assert(!strcmp(databuf, "this is another test string"));
+
+	// Do some time consuming processing
+	g_usleep(1*G_USEC_PER_SEC);
+	
+	OSyncMessage *reply = osync_message_new_reply(message, &error);
+	
+	osync_queue_send_message(server_queue, NULL, reply, &error);
+	
+	osync_message_unref(reply);
+	
+	osync_trace(TRACE_EXIT, "%s", __func__);
+}
+
+START_TEST (ipc_loop_with_timeout)
+{
+	
+	/* Even though each action takes 1 second, none of these messages should time out
+	   as they are being sent with a timeout of 3 seconds */
+	
+	num_msgs = 0;
+	req_msgs = 20;
+	
+	char *testbed = setup_testbed(NULL);
+	osync_testing_file_remove("/tmp/testpipe-server");
+	osync_testing_file_remove("/tmp/testpipe-client");
+	
+	OSyncError *error = NULL;
+	server_queue = osync_queue_new("/tmp/testpipe-server", &error);
+	client_queue = osync_queue_new("/tmp/testpipe-client", &error);
+	OSyncMessage *message = NULL;
+	
+	osync_queue_create(server_queue, &error);
+	fail_unless(error == NULL, NULL);
+	
+	osync_queue_create(client_queue, &error);
+	fail_unless(error == NULL, NULL);
+	char *data = "this is another test string";
+	
+	pid_t cpid = fork();
+	if (cpid == 0) { //Child
+		
+		GMainContext *context = g_main_context_new();
+		OSyncThread *thread = osync_thread_new(context, &error);
+	
+		osync_queue_set_message_handler(client_queue, client_handler7, GINT_TO_POINTER(1));
+		
+		osync_queue_setup_with_gmainloop(client_queue, context);
+		
+		osync_thread_start(thread);
+		
+		osync_assert(osync_queue_connect(client_queue, OSYNC_QUEUE_RECEIVER, &error));
+		osync_assert(error == NULL);
+
+		osync_assert(osync_queue_connect(server_queue, OSYNC_QUEUE_SENDER, &error));
+		osync_assert(error == NULL);
+
+		osync_queue_cross_link(client_queue, server_queue);
+		
+		message = osync_queue_get_message(server_queue);
+		
+		if (osync_message_get_command(message) != OSYNC_MESSAGE_QUEUE_HUP) {
+			exit (1);
+		}
+		
+		osync_message_unref(message);
+	
+		if (osync_queue_disconnect(server_queue, &error) != TRUE || error != NULL)
+			exit(1);
+		osync_queue_unref(server_queue);
+		
+		osync_assert(osync_queue_disconnect(client_queue, &error));
+		osync_assert(error == NULL);
+		
+		osync_thread_stop(thread);
+		osync_thread_unref(thread);
+		
+		osync_queue_unref(client_queue);
+		
+		g_free(testbed);
+		
+		exit(0);
+	} else {
+		GMainContext *context = g_main_context_new();
+		OSyncThread *thread = osync_thread_new(context, &error);
+		
+		osync_queue_set_message_handler(server_queue, server_handler7, GINT_TO_POINTER(1));
+		
+		osync_queue_setup_with_gmainloop(server_queue, context);
+		
+		osync_thread_start(thread);
+	
+		fail_unless(osync_queue_connect(client_queue, OSYNC_QUEUE_SENDER, &error), NULL);
+		fail_unless(error == NULL, NULL);
+		
+		fail_unless(osync_queue_connect(server_queue, OSYNC_QUEUE_RECEIVER, &error), NULL);
+		fail_unless(error == NULL, NULL);
+		
+		int i = 0;
+		for (i = 0; i < req_msgs; i++) {
+			message = osync_message_new(OSYNC_MESSAGE_INITIALIZE, 0, &error);
+			fail_unless(message != NULL, NULL);
+			fail_unless(!osync_error_is_set(&error), NULL);
+			
+			osync_message_write_int(message, 4000000);
+			osync_message_write_string(message, "this is a test string");
+			osync_message_write_long_long_int(message, 400000000);
+			osync_message_write_data(message, data, strlen(data) + 1);
+			
+			osync_message_set_handler(message, callback_handler7, GINT_TO_POINTER(1));
+			
+			fail_unless(osync_queue_send_message_with_timeout(client_queue, server_queue, message, 3, &error), NULL);
+			fail_unless(!osync_error_is_set(&error), NULL);
+			
+			osync_message_unref(message);
+		}
+		
+		message = osync_queue_get_message(client_queue);
+		
+		fail_unless(osync_message_get_command(message) == OSYNC_MESSAGE_QUEUE_HUP);
+		
+		osync_message_unref(message);
+		
+		osync_queue_disconnect(client_queue, &error);
+		fail_unless(error == NULL, NULL);
+		
+		osync_thread_stop(thread);
+		osync_thread_unref(thread);
+		
+		int status = 0;
+		wait(&status);
+		fail_unless(WEXITSTATUS(status) == 0, NULL);
+	}
+	
+	fail_unless(osync_testing_file_exists("/tmp/testpipe-client") == TRUE, NULL);
+	
+	fail_unless(osync_queue_remove(client_queue, &error), NULL);
+	fail_unless(osync_queue_remove(server_queue, &error), NULL);
+	fail_unless(!osync_error_is_set(&error), NULL);
+	
+	fail_unless(osync_testing_file_exists("/tmp/testpipe-client") == FALSE, NULL);
+
+	osync_queue_unref(client_queue);
+	osync_queue_unref(server_queue);
+	
+	destroy_testbed(testbed);
+}
+END_TEST
+
 OSYNC_TESTCASE_START("ipc")
 OSYNC_TESTCASE_ADD(ipc_new)
 OSYNC_TESTCASE_ADD(ipc_ref)
@@ -2285,5 +2679,7 @@
 OSYNC_TESTCASE_ADD(ipc_callback_break_pipes)
 
 OSYNC_TESTCASE_ADD(ipc_timeout)
+OSYNC_TESTCASE_ADD(ipc_late_reply)
+OSYNC_TESTCASE_ADD(ipc_loop_with_timeout)
 OSYNC_TESTCASE_END
 

[Opensync-commits] r5220 - in branches/timeout: . opensync/ipc

[<dg...@su...>]

Author: Graham Cobb
Date: Thu Jan 22 23:01:49 2009
New Revision: 5220
URL: http://www.opensync.org/changeset/5220

Log:
timeout: fix crash found running engine unit tests

Modified:
   branches/timeout/ChangeLog
   branches/timeout/opensync/ipc/opensync_queue.c

Modified: branches/timeout/ChangeLog
==============================================================================
--- branches/timeout/ChangeLog	Thu Jan 22 22:09:35 2009	(r5219)
+++ branches/timeout/ChangeLog	Thu Jan 22 23:01:49 2009	(r5220)
@@ -3,6 +3,7 @@
 	* opensync/ipc/opensync_queue.c (_timeout_dispatch): Restart timeout search at start of
 	pending list as list may have been modified while unlocked.
 	(_timeout_dispatch): Set message id in error message
+	(_timeout_dispatch): Undo change to restart timeout search (above)
 
 2009-01-21  Graham Cobb  <g+...@co...>
 

Modified: branches/timeout/opensync/ipc/opensync_queue.c
==============================================================================
--- branches/timeout/opensync/ipc/opensync_queue.c	Thu Jan 22 22:09:35 2009	(r5219)
+++ branches/timeout/opensync/ipc/opensync_queue.c	Thu Jan 22 23:01:49 2009	(r5220)
@@ -142,8 +142,10 @@
 			/* Lock again, to keep the iteration of the pendingReplies list atomic. */
 			g_mutex_lock(queue->pendingLock);
 
-			/* Restart search as list may have been modified while it was unlocked */
-			p = queue->pendingReplies;
+			/* The queue may have been modified while it was unlocked so don't go
+			   looking for any more entries.  If there are more, they will be found
+			   on the next call */
+			break;
 		}
 	}
 	

[Opensync-commits] r5219 - trunk/opensync/engine

[<dg...@su...>]

Author: dgollub
Date: Thu Jan 22 22:09:35 2009
New Revision: 5219
URL: http://www.opensync.org/changeset/5219

Log:
No re-assiging of mapping-entry-engines without OSyncChange required.

Modified:
   trunk/opensync/engine/opensync_engine.c

Modified: trunk/opensync/engine/opensync_engine.c
==============================================================================
--- trunk/opensync/engine/opensync_engine.c	Thu Jan 22 22:08:19 2009	(r5218)
+++ trunk/opensync/engine/opensync_engine.c	Thu Jan 22 22:09:35 2009	(r5219)
@@ -949,6 +949,10 @@
 				OSyncSinkEngine *new_sinkengine;
 
 				change = osync_entry_engine_get_change(mapping_entry_engine);
+				
+				if (!change)
+					continue;
+
 				objformat = osync_change_get_objformat(change);
 				osync_assert(objformat);
 				current_objtype = osync_objformat_get_objtype(objformat);

[Opensync-commits] r5218 - trunk/opensync/engine

[<dg...@su...>]

Author: dgollub
Date: Thu Jan 22 22:08:19 2009
New Revision: 5218
URL: http://www.opensync.org/changeset/5218

Log:
Store the original objtype in the archive/mappingtable instead of the
encapsulated objtype. This is required to have mapping-tables per
objtype. Reloading a mapping-table with mixed-objtype entries would end
up in an error about inconsistent mapping-table. Due to the limitation
of mixed-objtype syncing of encapsualted formats-only there is no need
to have an objtype-independent mapping-table (this would also make
slow-syncing per objtype make more difficult)

Modified:
   trunk/opensync/engine/opensync_sink_engine.c

Modified: trunk/opensync/engine/opensync_sink_engine.c
==============================================================================
--- trunk/opensync/engine/opensync_sink_engine.c	Thu Jan 22 21:54:12 2009	(r5217)
+++ trunk/opensync/engine/opensync_sink_engine.c	Thu Jan 22 22:08:19 2009	(r5218)
@@ -246,13 +246,15 @@
 			/* osync_assert_msg(!strcmp(objtype, osync_change_get_objtype(entry_engine->change), "Mixed-objtype in final write!")); */
 				
 			if (osync_change_get_changetype(entry_engine->change) == OSYNC_CHANGE_TYPE_DELETED) {
-				if (!osync_archive_delete_change(archive, osync_mapping_entry_get_id(entry), objtype, error))
+				if (!osync_archive_delete_change(archive, osync_mapping_entry_get_id(entry),
+							 osync_change_get_objtype(entry_engine->change), error))
 					goto error;
 			} else {
 				if (!osync_archive_save_change(archive, 
 							osync_mapping_entry_get_id(entry),
 							osync_change_get_uid(entry_engine->change),
-							objtype, osync_mapping_get_id(mapping),
+							osync_change_get_objtype(entry_engine->change),
+							osync_mapping_get_id(mapping),
 							osync_member_get_id(member), error))
 					goto error;
 			}

[Opensync-commits] r5217 - branches/timeout

[<dg...@su...>]

Author: Graham Cobb
Date: Thu Jan 22 21:54:12 2009
New Revision: 5217
URL: http://www.opensync.org/changeset/5217

Log:
Oops: undo local change to CMakeLists.txt in timeout branch

Modified:
   branches/timeout/CMakeLists.txt

Modified: branches/timeout/CMakeLists.txt
==============================================================================
--- branches/timeout/CMakeLists.txt	Thu Jan 22 21:51:07 2009	(r5216)
+++ branches/timeout/CMakeLists.txt	Thu Jan 22 21:54:12 2009	(r5217)
@@ -69,7 +69,6 @@
 FIND_PACKAGE( SWIG )	
 FIND_PACKAGE( PythonLibs )
 FIND_PACKAGE( Check )
-SET( CHECK_LIBRARIES "-lcheck_pic" )
 
 # test configuration
 

[Opensync-commits] r5216 - in branches/timeout: . opensync/client opensync/ipc

[<dg...@su...>]

Author: Graham Cobb
Date: Thu Jan 22 21:51:07 2009
New Revision: 5216
URL: http://www.opensync.org/changeset/5216

Log:
Initial work for new timeout handling.

OSyncQueue handling is reworked so that timeouts are not run on command 
sending queues but are run on command receiving queues.  This initial work
seems to work for simple cases both with and without timeouts.

However, the following tasks (at least) are still required:

1) Create unit tests
2) Limit the number of pending entries so the timeout is not
dependent on the number of outstanding requests
3) Add references for cross-linked pointers between queues and
remove linkages (and unref) when ??? (queue disconnected?)
4) Handle timeout before reply queue set up by generating an
error message on the command queue
5) Decide whether to handle timeouts that occur while a
plugin thread is sleeping.
6) Implement a mechanism to protect the queue itself: i.e. is the
receiver still reading the queue?

Modified:
   branches/timeout/   (props changed)
   branches/timeout/CMakeLists.txt
   branches/timeout/ChangeLog
   branches/timeout/opensync/client/opensync_client.c
   branches/timeout/opensync/client/opensync_client_proxy.c
   branches/timeout/opensync/ipc/opensync_message.c
   branches/timeout/opensync/ipc/opensync_message_internals.h
   branches/timeout/opensync/ipc/opensync_message_private.h
   branches/timeout/opensync/ipc/opensync_queue.c
   branches/timeout/opensync/ipc/opensync_queue_internals.h
   branches/timeout/opensync/ipc/opensync_queue_private.h

Modified: branches/timeout/CMakeLists.txt
==============================================================================
--- branches/timeout/CMakeLists.txt	Thu Jan 22 19:12:55 2009	(r5215)
+++ branches/timeout/CMakeLists.txt	Thu Jan 22 21:51:07 2009	(r5216)
@@ -69,6 +69,7 @@
 FIND_PACKAGE( SWIG )	
 FIND_PACKAGE( PythonLibs )
 FIND_PACKAGE( Check )
+SET( CHECK_LIBRARIES "-lcheck_pic" )
 
 # test configuration
 

Modified: branches/timeout/ChangeLog
==============================================================================
--- branches/timeout/ChangeLog	Thu Jan 22 19:12:55 2009	(r5215)
+++ branches/timeout/ChangeLog	Thu Jan 22 21:51:07 2009	(r5216)
@@ -1,3 +1,45 @@
+2009-01-22  Graham Cobb  <g+...@co...>
+
+	* opensync/ipc/opensync_queue.c (_timeout_dispatch): Restart timeout search at start of
+	pending list as list may have been modified while unlocked.
+	(_timeout_dispatch): Set message id in error message
+
+2009-01-21  Graham Cobb  <g+...@co...>
+
+	* opensync/ipc/opensync_queue.c (osync_queue_send_message_with_timeout): If this is
+	a reply, remove pending command
+
+	* opensync/client/opensync_client_proxy.c (_osync_client_proxy_hup_handler): treat
+	a queue error as a disconnect
+	(osync_client_proxy_spawn): cross-link command and reply queues
+
+	* opensync/client/opensync_client.c (_osync_client_handle_initialize): cross-link
+	command and reply queues
+	(_osync_client_hup_handler): treat a queue error as a disconnect
+
+2009-01-20  Graham Cobb  <g+...@co...>
+
+	* opensync/ipc/opensync_queue_internals.h: Add osync_queue_cross_link
+
+	* opensync/ipc/opensync_queue.c (osync_queue_cross_link): Cross-link command and reply queues
+
+	* opensync/ipc/opensync_queue.c (_osync_send_timeout_response): 
+	Callback function for timeouts: send the timeout error reponse
+	(_incoming_dispatch): For incoming messages which have timeouts, 
+	create a pending list entry
+
+	* opensync/ipc/opensync_queue_private.h: Add reply_queue for pointer to 
+	queue used for replies to incoming messages, if any.
+	Add cmd_queue for pointer to queue used to store pending commands.
+
+2009-01-19  Graham Cobb  <g+...@co...>
+
+	* opensync/ipc/opensync_message.c: Add opensync_message_set/get_timeout
+
+	* opensync/ipc/opensync_message_internals.h: Add opensync_message_set/get_timeout
+
+	* opensync/ipc/opensync_message_private.h: Add timeout to message
+
 2008-05-04  Graham Cobb  <g+...@co...>
 
 	* opensync/helper/opensync_hashtable.c (osync_hashtable_get_deleted): 

Modified: branches/timeout/opensync/client/opensync_client.c
==============================================================================
--- branches/timeout/opensync/client/opensync_client.c	Thu Jan 22 19:12:55 2009	(r5215)
+++ branches/timeout/opensync/client/opensync_client.c	Thu Jan 22 21:51:07 2009	(r5216)
@@ -604,6 +604,7 @@
 		}
 		
 		osync_client_set_outgoing_queue(client, outgoing);
+		osync_queue_cross_link(client->incoming, client->outgoing);
 		osync_queue_unref(outgoing);
 		osync_trace(TRACE_INTERNAL, "done connecting to engine");
 	}
@@ -1519,9 +1520,8 @@
 
 	osync_trace(TRACE_INTERNAL, "plugin received command %i on sending queue", osync_message_get_command(message));
 
-	if (osync_message_get_command(message) == OSYNC_MESSAGE_QUEUE_ERROR) {
-		/* Houston, we have a problem */
-	} else if (osync_message_get_command(message) == OSYNC_MESSAGE_QUEUE_HUP) {
+	if ( (osync_message_get_command(message) == OSYNC_MESSAGE_QUEUE_ERROR)
+	     || (osync_message_get_command(message) == OSYNC_MESSAGE_QUEUE_HUP) ) {
 		/* The remote side disconnected. So we can now disconnect as well and then
 		 * shutdown */
 		if (!osync_queue_disconnect(client->outgoing, &error))

Modified: branches/timeout/opensync/client/opensync_client_proxy.c
==============================================================================
--- branches/timeout/opensync/client/opensync_client_proxy.c	Thu Jan 22 19:12:55 2009	(r5215)
+++ branches/timeout/opensync/client/opensync_client_proxy.c	Thu Jan 22 21:51:07 2009	(r5216)
@@ -227,9 +227,8 @@
 
 	osync_trace(TRACE_INTERNAL, "client received command %i on sending queue", osync_message_get_command(message));
 
-	if (osync_message_get_command(message) == OSYNC_MESSAGE_QUEUE_ERROR) {
-		/* Houston, we have a problem */
-	} else if (osync_message_get_command(message) == OSYNC_MESSAGE_QUEUE_HUP) {
+	if ( (osync_message_get_command(message) == OSYNC_MESSAGE_QUEUE_ERROR) /* Treat an error as a disconnect */
+	     || (osync_message_get_command(message) == OSYNC_MESSAGE_QUEUE_HUP) ) {
 		/* The remote side disconnected. So we can now disconnect as well and then
 		 * shutdown */
 		if (!osync_queue_disconnect(proxy->outgoing, &error))
@@ -916,6 +915,7 @@
 			
 			osync_client_set_incoming_queue(proxy->client, read1);
 			osync_client_set_outgoing_queue(proxy->client, write2);
+			osync_queue_cross_link(read1, write2);
 			
 			if (!osync_client_run(proxy->client, error))
 				goto error_free_pipe2;

Modified: branches/timeout/opensync/ipc/opensync_message.c
==============================================================================
--- branches/timeout/opensync/ipc/opensync_message.c	Thu Jan 22 19:12:55 2009	(r5215)
+++ branches/timeout/opensync/ipc/opensync_message.c	Thu Jan 22 21:51:07 2009	(r5216)
@@ -74,6 +74,18 @@
 	return message->cmd;
 }
 
+void osync_message_set_timeout(OSyncMessage *message, unsigned int timeout)
+{
+	osync_assert(message);
+	message->timeout = timeout;
+}
+
+unsigned int osync_message_get_timeout(OSyncMessage *message)
+{
+	osync_assert(message);
+	return message->timeout;
+}
+
 void osync_message_set_id(OSyncMessage *message, long long int id)
 {
 	osync_assert(message);

Modified: branches/timeout/opensync/ipc/opensync_message_internals.h
==============================================================================
--- branches/timeout/opensync/ipc/opensync_message_internals.h	Thu Jan 22 19:12:55 2009	(r5215)
+++ branches/timeout/opensync/ipc/opensync_message_internals.h	Thu Jan 22 21:51:07 2009	(r5216)
@@ -193,6 +193,21 @@
  */
 OSYNC_TEST_EXPORT void osync_message_get_buffer(OSyncMessage *message, char **data, unsigned int *size);
 
+/** @brief Set timeout (in seconds) for the message object
+ * 
+ * @param message The message to modify 
+ * @param timeout The new message timeout
+ * 
+ */
+OSYNC_TEST_EXPORT void osync_message_set_timeout(OSyncMessage *message, unsigned int timeout);
+
+/** @brief Get timeout (in seconds) for the message object
+ * 
+ * @param message The message
+ * 
+ */
+OSYNC_TEST_EXPORT unsigned int osync_message_get_timeout(OSyncMessage *message);
+
 /** @brief Sets the handler that will receive the reply
  * 
  * @param message The message to work on

Modified: branches/timeout/opensync/ipc/opensync_message_private.h
==============================================================================
--- branches/timeout/opensync/ipc/opensync_message_private.h	Thu Jan 22 19:12:55 2009	(r5215)
+++ branches/timeout/opensync/ipc/opensync_message_private.h	Thu Jan 22 21:51:07 2009	(r5216)
@@ -45,7 +45,7 @@
 	/** The user data */
 	gpointer user_data;
 	/** The timeout associated with this message */
-	//timeout_info *to_info;
+	unsigned int timeout;
 	/** If this message has already been answered */
 	osync_bool is_answered;
 	/** The internal OSyncMarshal object **/

Modified: branches/timeout/opensync/ipc/opensync_queue.c
==============================================================================
--- branches/timeout/opensync/ipc/opensync_queue.c	Thu Jan 22 19:12:55 2009	(r5215)
+++ branches/timeout/opensync/ipc/opensync_queue.c	Thu Jan 22 21:51:07 2009	(r5216)
@@ -2,6 +2,7 @@
  * libosengine - A synchronization engine for the opensync framework
  * Copyright (C) 2004-2005  Armin Bauer <arm...@op...>
  * Copyright (C) 2007  Daniel Friedrich <dan...@op...>
+ * Copyright (C) 2009 Graham R. Cobb <g+o...@co...>
  * 
  * This library is free software; you can redistribute it and/or
  * modify it under the terms of the GNU Lesser General Public
@@ -110,11 +111,15 @@
 			OSyncError *timeouterr = NULL;
 			OSyncMessage *errormsg = NULL;
 
+			osync_trace(TRACE_INTERNAL, "Timeout for message with id %lli", pending->id);
+
+
 			/* Call the callback of the pending message */
 			osync_assert(pending->callback);
 			osync_error_set(&timeouterr, OSYNC_ERROR_IO_ERROR, "Timeout.");
 			errormsg = osync_message_new_errorreply(NULL, timeouterr, &error);
 			osync_error_unref(&timeouterr);
+			osync_message_set_id(errormsg, pending->id);
 
 			/* Remove first the pending message!
 				 To avoid that _incoming_dispatch catchs this message
@@ -137,7 +142,8 @@
 			/* Lock again, to keep the iteration of the pendingReplies list atomic. */
 			g_mutex_lock(queue->pendingLock);
 
-			break;
+			/* Restart search as list may have been modified while it was unlocked */
+			p = queue->pendingReplies;
 		}
 	}
 	
@@ -161,67 +167,146 @@
 	return FALSE;
 }
 
+static void _osync_send_timeout_response(OSyncMessage *errormsg, void *user_data)
+{
+	OSyncQueue *queue = user_data;
+
+	osync_trace(TRACE_ENTRY, "%s(%p, %p)", __func__, errormsg, queue);
+
+	if (queue->reply_queue) {
+		osync_queue_send_message_with_timeout(queue->reply_queue, NULL, errormsg, 0, NULL);
+	} else {
+		osync_message_unref(errormsg);
+		/* TODO: as we can't send the timeout response, disconnect the queue to signal to
+		   the sender that they aren't going to get a response.  Note: we can't just
+		   call osync_queue_disconnect as that tries to kill the thread and hits a deadlock! */
+		osync_trace(TRACE_EXIT_ERROR, "%s: cannot find reply queue to send timeout error", __func__);
+		return;
+	}
+	osync_trace(TRACE_EXIT, "%s", __func__);
+}
+
+/* Find and remove a pending message that this message is a reply to */
+static void _osync_queue_remove_pending_reply(OSyncQueue *queue, OSyncMessage *reply, gboolean callback)
+{
+	OSyncPendingMessage *pending = NULL;
+	OSyncList *p = NULL;
+
+	osync_trace(TRACE_ENTRY, "%s(%p, %p, %d)", __func__, queue, reply, callback);
+	osync_trace(TRACE_INTERNAL, "Searching for pending message id=%lli", osync_message_get_id(reply));
+
+	/* Search for the pending reply. We have to lock the
+	 * list since another thread might be duing the updates */
+	g_mutex_lock(queue->pendingLock);
+			
+	for (p = queue->pendingReplies; p; p = p->next) {
+		pending = p->data;
+			
+		if (pending->id == osync_message_get_id(reply)) {
+			
+			osync_trace(TRACE_INTERNAL, "Found pending message id=%lli: %p", osync_message_get_id(reply), pending);
+			/* Remove first the pending message!
+			   To avoid that _timeout_dispatch catchs this message
+			   when we're releasing the lock. If _timeout_dispatch
+			   would catch this message, the pending callback
+			   gets called twice! */
+			
+			queue->pendingReplies = osync_list_remove(queue->pendingReplies, pending);
+
+			/* Unlock the pending lock since the messages might be sent during the callback */
+			g_mutex_unlock(queue->pendingLock);
+			
+			if (callback) {
+				/* Call the callback of the pending message */
+				osync_assert(pending->callback);
+				pending->callback(reply, pending->user_data);
+			}
+
+			// TODO: Refcounting for OSyncPendingMessage
+			if (pending->timeout_info)
+				g_free(pending->timeout_info);
+			osync_free(pending);
+
+			osync_trace(TRACE_EXIT, "%s", __func__);
+			return;
+		}
+	}
+	
+	g_mutex_unlock(queue->pendingLock);
+	osync_trace(TRACE_EXIT, "%s", __func__);
+}
+
 /* This function is called from the master thread. The function dispatched incoming data from
  * the remote end */
 static gboolean _incoming_dispatch(GSource *source, GSourceFunc callback, gpointer user_data)
 {
-	OSyncPendingMessage *pending = NULL;
 	OSyncQueue *queue = user_data;
-	OSyncList *p = NULL;
 	OSyncMessage *message = NULL;
+	OSyncError *error = NULL;
 	
 	osync_trace(TRACE_ENTRY, "%s(%p)", __func__, user_data);
 
 	while ((message = g_async_queue_try_pop(queue->incoming))) {
 		/* We check if the message is a reply to something */
-		osync_trace(TRACE_INTERNAL, "Dispatching %p:%i(%s)", message, osync_message_get_cmd(message), osync_message_get_commandstr(message));
+		osync_trace(TRACE_INTERNAL, "Dispatching %p:%i(%s), timeout=%d, id=%lli", message, 
+			    osync_message_get_cmd(message), osync_message_get_commandstr(message), 
+			    osync_message_get_timeout(message), osync_message_get_id(message));
 		
 		if (osync_message_get_cmd(message) == OSYNC_MESSAGE_REPLY || osync_message_get_cmd(message) == OSYNC_MESSAGE_ERRORREPLY) {
-			
-			/* Search for the pending reply. We have to lock the
-			 * list since another thread might be duing the updates */
-			g_mutex_lock(queue->pendingLock);
-			
-			for (p = queue->pendingReplies; p; p = p->next) {
-				pending = p->data;
-			
-				if (pending->id == osync_message_get_id(message)) {
+			/* Remove pending reply and call callback */
+			_osync_queue_remove_pending_reply(queue, message, TRUE);
+		} else {
+			unsigned int timeout = osync_message_get_timeout(message);
+			if (timeout) {
+				/* This message has a timeout. Put message on pending list and run timeout */
+				/* NOTE: We really want to put it on the pending list for the corresponding
+				   outgoing queue but there may not be one at this time (if this is an Initialise,
+				   for example).  So we put it on the pending list of the incoming queue and deal with
+				   finding the outgoing queue if the timeout fires.  This also complicates removing
+				   pending items when sending the responses. Oh well. */
+				
+				OSyncPendingMessage *pending = osync_try_malloc0(sizeof(OSyncPendingMessage), &error);
+				if (!pending)
+					goto error;
 
-					/* Remove first the pending message!
-						 To avoid that _timeout_dispatch catchs this message
-						 when we're releasing the lock. If _timeout_dispatch
-						 would catch this message, the pending callback
-						 gets called twice! */
+				pending->id = osync_message_get_id(message);
 
-					queue->pendingReplies = osync_list_remove(queue->pendingReplies, pending);
+				OSyncTimeoutInfo *toinfo = osync_try_malloc0(sizeof(OSyncTimeoutInfo), &error);
+				if (!toinfo)
+					goto error;
 
-					/* Unlock the pending lock since the messages might be sent during the callback */
-					g_mutex_unlock(queue->pendingLock);
-			
-					/* Call the callback of the pending message */
-					osync_assert(pending->callback);
-					pending->callback(message, pending->user_data);
-
-					// TODO: Refcounting for OSyncPendingMessage
-					if (pending->timeout_info)
-						g_free(pending->timeout_info);
-					osync_free(pending);
-
-					/* Lock again, to keep the iteration of the pendingReplies list atomic. */
-					g_mutex_lock(queue->pendingLock);
-					break;
-				}
+				GTimeVal current_time;
+				g_source_get_current_time(queue->timeout_source, &current_time);
+
+				toinfo->expiration = current_time;
+				toinfo->expiration.tv_sec += timeout;
+
+				pending->timeout_info = toinfo;
+
+				pending->callback = _osync_send_timeout_response;
+				pending->user_data = queue;
+		
+				g_mutex_lock(queue->pendingLock);
+				queue->pendingReplies = osync_list_append(queue->pendingReplies, pending);
+				g_mutex_unlock(queue->pendingLock);
 			}
-			
-			g_mutex_unlock(queue->pendingLock);
-		} else 
+
 			queue->message_handler(message, queue->user_data);
+		}
 		
 		osync_message_unref(message);
 	}
 	
 	osync_trace(TRACE_EXIT, "%s: Done dispatching", __func__);
 	return TRUE;
+
+ error:
+	/* TODO: as we can't send the timeout response, disconnect the queue to signal to
+	   the sender that they aren't going to get a response.  Note: we can't just
+	   call osync_queue_disconnect as that tries to kill the thread and hits a deadlock! */
+
+	osync_trace(TRACE_EXIT_ERROR, "%s: %s", __func__, osync_error_print(&error));
+	return TRUE;
 }
 
 static void _osync_queue_stop_incoming(OSyncQueue *queue)
@@ -328,6 +413,10 @@
 		if (!_osync_queue_write_long_long_int(queue, osync_message_get_id(message), &error))
 			goto error;
 		
+		/* The timeout (integer) */
+		if (!_osync_queue_write_int(queue, (int) osync_message_get_timeout(message), &error))
+			goto error;
+		
 		osync_message_get_buffer(message, &data, &length);
 		
 		if (length) {
@@ -498,7 +587,7 @@
 	
 	do {
 		int size = 0;
-		int cmd = 0;
+		int cmd = 0, timeout = 0;
 		long long int id = 0;
 		char *buffer = NULL;
 		
@@ -514,11 +603,16 @@
 		if (!_osync_queue_read_long_long_int(queue, &id, &error))
 			goto error;
 		
+		/* The timeout */
+		if (!_osync_queue_read_int(queue, &timeout, &error))
+			goto error;
+		
 		message = osync_message_new(cmd, size, &error);
 		if (!message)
 			goto error;
 	
 		osync_message_set_id(message, id);
+		osync_message_set_timeout(message, (unsigned int) timeout);
 		
 		/* We now get the buffer from the message which will already
 		 * have the correct size for the read */
@@ -933,7 +1027,7 @@
 
 void osync_queue_set_message_handler(OSyncQueue *queue, OSyncMessageHandler handler, gpointer user_data)
 {
-	osync_trace(TRACE_ENTRY, "%s(%p, %p, %p)", __func__, queue, handler, user_data);
+	osync_trace(TRACE_ENTRY, "%s(%p, %p, %p, %p)", __func__, queue, handler, user_data);
 	
 	queue->message_handler = handler;
 	queue->user_data = user_data;
@@ -941,6 +1035,22 @@
 	osync_trace(TRACE_EXIT, "%s", __func__);
 }
 
+void osync_queue_cross_link(OSyncQueue *cmd_queue, OSyncQueue *reply_queue)
+{
+	/* Cross-linking is needed to make timeouts work when commands are
+	   being received on one queue and replies sent on another.
+	   Note that cross-linking is not needed when commands are being sent. */
+	osync_trace(TRACE_ENTRY, "%s(%p, %p)", __func__, cmd_queue, reply_queue);
+
+	osync_assert(cmd_queue->type == OSYNC_QUEUE_RECEIVER);
+	osync_assert(reply_queue->type == OSYNC_QUEUE_SENDER);
+
+	cmd_queue->reply_queue = reply_queue;
+	reply_queue->cmd_queue = cmd_queue;
+	
+	osync_trace(TRACE_EXIT, "%s", __func__);
+}
+
 void osync_queue_setup_with_gmainloop(OSyncQueue *queue, GMainContext *context)
 {
 	OSyncQueue **queueptr = NULL;
@@ -1035,6 +1145,19 @@
 {
 	osync_trace(TRACE_ENTRY, "%s(%p, %p, %p, %u, %p)", __func__, queue, replyqueue, message, timeout, error);
 
+	if (queue->cmd_queue) {
+		/* This queue is a reply queue for some command receiving queue */
+
+		/* If this is actually a reply message, we check to see if there is
+		   message to remove from the command queue's pending list */
+
+		if (osync_message_get_cmd(message) == OSYNC_MESSAGE_REPLY || osync_message_get_cmd(message) == OSYNC_MESSAGE_ERRORREPLY) {
+			/* Remove pending reply but do not call callback
+			   (real callback will be called by the receiver of this reply later) */
+			_osync_queue_remove_pending_reply(queue->cmd_queue, message, FALSE);
+		}
+	}
+
 	if (osync_message_get_handler(message)) {
 		OSyncPendingMessage *pending = NULL;
 		GTimeVal current_time;
@@ -1053,14 +1176,8 @@
 		osync_trace(TRACE_INTERNAL, "Setting id %lli for pending reply", id);
 
 		if (timeout) {
-			OSyncTimeoutInfo *toinfo = osync_try_malloc0(sizeof(OSyncTimeoutInfo), error);
-			if (!toinfo)
-				goto error;
-
-			toinfo->expiration = current_time;
-			toinfo->expiration.tv_sec += timeout;
-
-			pending->timeout_info = toinfo;
+			/* Send timeout info to other end to handle */
+			osync_message_set_timeout(message, timeout);
 		} else {
 			osync_trace(TRACE_INTERNAL, "handler message got sent without timeout!: %s", osync_message_get_commandstr(message));
 		}
@@ -1118,6 +1235,7 @@
 {
 	osync_assert(queue);
 	return queue->name;
+
 }
 
 int osync_queue_get_fd(OSyncQueue *queue)

Modified: branches/timeout/opensync/ipc/opensync_queue_internals.h
==============================================================================
--- branches/timeout/opensync/ipc/opensync_queue_internals.h	Thu Jan 22 19:12:55 2009	(r5215)
+++ branches/timeout/opensync/ipc/opensync_queue_internals.h	Thu Jan 22 21:51:07 2009	(r5216)
@@ -85,6 +85,20 @@
 OSYNC_TEST_EXPORT void osync_queue_set_message_handler(OSyncQueue *queue, OSyncMessageHandler handler, gpointer user_data);
 
 /**
+ * @brief Cross links command queue and reply queue
+ * 
+ * Stores the queue used for replies in the command queue object so
+ * that timeout responses can be sent if necessary.
+ * And stores the command queue in the reply queue object so that
+ * replies can remove pending messages before they time out.
+ * 
+ * @param cmd_queue The command queue used to receive incoming commands
+ * @param reply_queue The queue used to send replies 
+ * 
+ */
+OSYNC_TEST_EXPORT void osync_queue_cross_link(OSyncQueue *cmd_queue, OSyncQueue *reply_queue);
+
+/**
  * @brief Sends a Message to a Queue
  * @param queue Pointer to the queue
  * @param replyqueue

Modified: branches/timeout/opensync/ipc/opensync_queue_private.h
==============================================================================
--- branches/timeout/opensync/ipc/opensync_queue_private.h	Thu Jan 22 19:12:55 2009	(r5215)
+++ branches/timeout/opensync/ipc/opensync_queue_private.h	Thu Jan 22 21:51:07 2009	(r5216)
@@ -89,6 +89,12 @@
 
 	/** Reference count **/
 	int ref_count;
+
+	/* Queue for replies to incoming messages */
+	OSyncQueue *reply_queue;
+
+	/* Queue for receiving commands we are replying to */
+	OSyncQueue *cmd_queue;
 };