2003-03-16 Havoc Pennington <hp@pobox.com>
+ * dbus/dbus-string.c (_dbus_string_validate_utf8): oops, unbreak
+ this. always run the test suite before commit...
+
+ * bus/*: adapt to DBusConnection API changes
+
+ * glib/dbus-gmain.c: adapt to DBusConnection API changes,
+ requires renaming stuff to avoid dbus_connection_dispatch name
+ conflict.
+
+ * dbus/dbus-transport.c (_dbus_transport_queue_messages): new
+ function
+
+ * dbus/dbus-message.c (_dbus_message_loader_queue_messages):
+ separate from _dbus_message_loader_return_buffer()
+
+ * dbus/dbus-connection.c (dbus_connection_get_n_messages): remove
+ this, because it's now always broken to use; the number of
+ messages in queue vs. the number still buffered by the message
+ loader is undefined/meaningless. Should use
+ dbus_connection_get_dispatch_state().
+ (dbus_connection_dispatch): rename from
+ dbus_connection_dispatch_message
+
+2003-03-16 Havoc Pennington <hp@pobox.com>
+
* dbus/dbus-string.c (_dbus_string_validate_utf8): copy in a real
implementation
dbus_connection_handle_watch (connection, watch, condition);
- while (dbus_connection_dispatch_message (connection))
- ;
+ bus_connection_dispatch_all_messages (connection);
+
dbus_connection_unref (connection);
}
dbus_timeout_handle (timeout);
- while (dbus_connection_dispatch_message (connection))
- ;
+ bus_connection_dispatch_all_messages (connection);
+
dbus_connection_unref (connection);
}
_dbus_assert (bus_test_client_listed (connection));
/* Run disconnect handler in test.c */
- if (dbus_connection_dispatch_message (connection))
+ if (bus_connection_dispatch_one_message (connection))
_dbus_assert_not_reached ("something received on connection being killed other than the disconnect");
_dbus_assert (!dbus_connection_get_is_connected (connection));
dbus_connection_ref (connection);
dbus_connection_disconnect (connection);
/* dispatching disconnect handler will unref once */
- if (dbus_connection_dispatch_message (connection))
+ if (bus_connection_dispatch_one_message (connection))
_dbus_assert_not_reached ("message other than disconnect dispatched after failure to register");
dbus_connection_unref (connection);
_dbus_assert (!bus_test_client_listed (connection));
check_hello_connection);
dbus_connection_disconnect (foo);
- if (dbus_connection_dispatch_message (foo))
+ if (bus_connection_dispatch_one_message (foo))
_dbus_assert_not_reached ("extra message in queue");
dbus_connection_unref (foo);
_dbus_assert (!bus_test_client_listed (foo));
dbus_connection_disconnect (bar);
- if (dbus_connection_dispatch_message (bar))
+ if (bus_connection_dispatch_one_message (bar))
_dbus_assert_not_reached ("extra message in queue");
dbus_connection_unref (bar);
_dbus_assert (!bus_test_client_listed (bar));
dbus_connection_disconnect (baz);
- if (dbus_connection_dispatch_message (baz))
+ if (bus_connection_dispatch_one_message (baz))
_dbus_assert_not_reached ("extra message in queue");
dbus_connection_unref (baz);
_dbus_assert (!bus_test_client_listed (baz));
#endif
}
+void
+bus_connection_dispatch_all_messages (DBusConnection *connection)
+{
+ while (bus_connection_dispatch_one_message (connection))
+ ;
+}
+
+dbus_bool_t
+bus_connection_dispatch_one_message (DBusConnection *connection)
+{
+ DBusDispatchStatus status;
+
+ while ((status = dbus_connection_dispatch (connection)) == DBUS_DISPATCH_NEED_MEMORY)
+ bus_wait_for_memory ();
+
+ return status == DBUS_DISPATCH_DATA_REMAINS;
+}
#ifndef BUS_UTILS_H
#define BUS_UTILS_H
+#include <dbus/dbus.h>
+
void bus_wait_for_memory (void);
extern const char bus_no_memory_message[];
#define BUS_SET_OOM(error) dbus_set_error ((error), DBUS_ERROR_NO_MEMORY, bus_no_memory_message)
+void bus_connection_dispatch_all_messages (DBusConnection *connection);
+dbus_bool_t bus_connection_dispatch_one_message (DBusConnection *connection);
+
#endif /* BUS_ACTIVATION_H */
*/
typedef struct
{
- char *base_service;
+ char *base_service; /**< Base service name of this connection */
} BusData;
* once per connection.
*
* @param connection the connection
- * @param the base service name
+ * @param base_service the base service name
* @returns #FALSE if not enough memory
*/
dbus_bool_t
DBusMutex *mutex; /**< Lock on the entire DBusConnection */
- dbus_bool_t dispatch_acquired; /**< Protects dispatch_message */
- DBusCondVar *dispatch_cond; /**< Protects dispatch_message */
+ dbus_bool_t dispatch_acquired; /**< Protects dispatch() */
+ DBusCondVar *dispatch_cond; /**< Protects dispatch() */
dbus_bool_t io_path_acquired; /**< Protects transport io path */
DBusCondVar *io_path_cond; /**< Protects transport io path */
static void reply_handler_data_free (ReplyHandlerData *data);
-static void _dbus_connection_remove_timeout_locked (DBusConnection *connection,
- DBusTimeout *timeout);
+static void _dbus_connection_remove_timeout_locked (DBusConnection *connection,
+ DBusTimeout *timeout);
+static DBusDispatchStatus _dbus_connection_get_dispatch_status_unlocked (DBusConnection *connection);
/**
* Acquires the connection lock.
* function on a watch that was not previously added.
*
* @param connection the connection.
- * @param timeout the timeout to toggle.
+ * @param watch the watch to toggle.
* @param enabled whether to enable or disable
*/
void
return TRUE;
}
+
+static DBusMessage*
+check_for_reply_unlocked (DBusConnection *connection,
+ dbus_int32_t client_serial)
+{
+ DBusList *link;
+
+ link = _dbus_list_get_first_link (&connection->incoming_messages);
+
+ while (link != NULL)
+ {
+ DBusMessage *reply = link->data;
+
+ if (dbus_message_get_reply_serial (reply) == client_serial)
+ {
+ _dbus_list_remove_link (&connection->incoming_messages, link);
+ connection->n_incoming -= 1;
+ dbus_message_ref (reply);
+ return reply;
+ }
+ link = _dbus_list_get_next_link (&connection->incoming_messages, link);
+ }
+
+ return NULL;
+}
+
/**
* Sends a message and blocks a certain time period while waiting for a reply.
* This function does not dispatch any message handlers until the main loop
DBusError *error)
{
dbus_int32_t client_serial;
- DBusList *link;
long start_tv_sec, start_tv_usec;
long end_tv_sec, end_tv_usec;
long tv_sec, tv_usec;
-
+ DBusDispatchStatus status;
+
if (timeout_milliseconds == -1)
timeout_milliseconds = DEFAULT_TIMEOUT_VALUE;
end_tv_sec, end_tv_usec);
/* Now we wait... */
- /* THREAD TODO: This is busted. What if a dispatch_message or pop_message
+ /* THREAD TODO: This is busted. What if a dispatch() or pop_message
* gets the message before we do?
*/
- block_again:
-
+ /* always block at least once as we know we don't have the reply yet */
_dbus_connection_do_iteration (connection,
- DBUS_ITERATION_DO_READING |
- DBUS_ITERATION_BLOCK,
- timeout_milliseconds);
+ DBUS_ITERATION_DO_READING |
+ DBUS_ITERATION_BLOCK,
+ timeout_milliseconds);
- /* Check if we've gotten a reply */
- link = _dbus_list_get_first_link (&connection->incoming_messages);
+ recheck_status:
- while (link != NULL)
- {
- DBusMessage *reply = link->data;
+ /* queue messages and get status */
+ status = _dbus_connection_get_dispatch_status_unlocked (connection);
- if (dbus_message_get_reply_serial (reply) == client_serial)
- {
- _dbus_list_remove_link (&connection->incoming_messages, link);
- connection->n_incoming -= 1;
- dbus_message_ref (reply);
-
- dbus_mutex_unlock (connection->mutex);
- return reply;
- }
- link = _dbus_list_get_next_link (&connection->incoming_messages, link);
+ if (status == DBUS_DISPATCH_DATA_REMAINS)
+ {
+ DBusMessage *reply;
+
+ reply = check_for_reply_unlocked (connection, client_serial);
+ if (reply != NULL)
+ {
+ dbus_mutex_unlock (connection->mutex);
+ return reply;
+ }
}
-
+
_dbus_get_current_time (&tv_sec, &tv_usec);
if (tv_sec < start_tv_sec)
_dbus_verbose ("%d milliseconds remain\n", timeout_milliseconds);
_dbus_assert (timeout_milliseconds > 0);
- goto block_again; /* not expired yet */
+ if (status == DBUS_DISPATCH_NEED_MEMORY)
+ {
+ /* Try sleeping a bit, as we aren't sure we need to block for reading,
+ * we may already have a reply in the buffer and just can't process
+ * it.
+ */
+ if (timeout_milliseconds < 100)
+ ; /* just busy loop */
+ else if (timeout_milliseconds <= 1000)
+ _dbus_sleep_milliseconds (timeout_milliseconds / 3);
+ else
+ _dbus_sleep_milliseconds (1000);
+ }
+ else
+ {
+ /* block again, we don't have the reply buffered yet. */
+ _dbus_connection_do_iteration (connection,
+ DBUS_ITERATION_DO_READING |
+ DBUS_ITERATION_BLOCK,
+ timeout_milliseconds);
+ }
+
+ goto recheck_status;
}
if (dbus_connection_get_is_connected (connection))
dbus_mutex_unlock (connection->mutex);
}
-/**
- * Gets the number of messages in the incoming message queue.
- *
- * @param connection the connection.
- * @returns the number of messages in the queue.
- */
-int
-dbus_connection_get_n_messages (DBusConnection *connection)
-{
- int res;
-
- dbus_mutex_lock (connection->mutex);
- res = connection->n_incoming;
- dbus_mutex_unlock (connection->mutex);
- return res;
-}
-
-
/* Call with mutex held. Will drop it while waiting and re-acquire
* before returning
*/
dbus_connection_borrow_message (DBusConnection *connection)
{
DBusMessage *message;
-
+ DBusDispatchStatus status;
+
+ /* this is called for the side effect that it queues
+ * up any messages from the transport
+ */
+ status = dbus_connection_get_dispatch_status (connection);
+ if (status != DBUS_DISPATCH_DATA_REMAINS)
+ return NULL;
+
dbus_mutex_lock (connection->mutex);
if (connection->message_borrowed != NULL)
dbus_connection_pop_message (DBusConnection *connection)
{
DBusMessage *message;
- dbus_mutex_lock (connection->mutex);
+ DBusDispatchStatus status;
+ /* this is called for the side effect that it queues
+ * up any messages from the transport
+ */
+ status = dbus_connection_get_dispatch_status (connection);
+ if (status != DBUS_DISPATCH_DATA_REMAINS)
+ return NULL;
+
+ dbus_mutex_lock (connection->mutex);
+
message = _dbus_connection_pop_message_unlocked (connection);
dbus_mutex_unlock (connection->mutex);
connection->n_incoming += 1;
}
+static DBusDispatchStatus
+_dbus_connection_get_dispatch_status_unlocked (DBusConnection *connection)
+{
+ if (connection->n_incoming > 0)
+ return DBUS_DISPATCH_DATA_REMAINS;
+ else if (!_dbus_transport_queue_messages (connection->transport))
+ return DBUS_DISPATCH_NEED_MEMORY;
+ else
+ {
+ DBusDispatchStatus status;
+
+ status = _dbus_transport_get_dispatch_status (connection->transport);
+
+ if (status != DBUS_DISPATCH_COMPLETE)
+ return status;
+ else if (connection->n_incoming > 0)
+ return DBUS_DISPATCH_DATA_REMAINS;
+ else
+ return DBUS_DISPATCH_COMPLETE;
+ }
+}
+
+/**
+ * Gets the current state (what we would currently return
+ * from dbus_connection_dispatch()) but doesn't actually
+ * dispatch any messages.
+ *
+ * @param connection the connection.
+ * @returns current dispatch status
+ */
+DBusDispatchStatus
+dbus_connection_get_dispatch_status (DBusConnection *connection)
+{
+ DBusDispatchStatus status;
+
+ dbus_mutex_lock (connection->mutex);
+
+ status = _dbus_connection_get_dispatch_status_unlocked (connection);
+
+ dbus_mutex_unlock (connection->mutex);
+
+ return status;
+}
+
/**
- * Pops the first-received message from the current incoming message
- * queue, runs any handlers for it, then unrefs the message.
+ * Processes data buffered while handling watches, queueing zero or
+ * more incoming messages. Then pops the first-received message from
+ * the current incoming message queue, runs any handlers for it, and
+ * unrefs the message. Returns a status indicating whether messages/data
+ * remain, more memory is needed, or all data has been processed.
*
* @param connection the connection
- * @returns #TRUE if the queue is not empty after dispatch
+ * @returns dispatch status
*/
-dbus_bool_t
-dbus_connection_dispatch_message (DBusConnection *connection)
+DBusDispatchStatus
+dbus_connection_dispatch (DBusConnection *connection)
{
DBusMessageHandler *handler;
DBusMessage *message;
ReplyHandlerData *reply_handler_data;
const char *name;
dbus_int32_t reply_serial;
+ DBusDispatchStatus status;
- dbus_mutex_lock (connection->mutex);
+ status = dbus_connection_get_dispatch_status (connection);
+ if (status != DBUS_DISPATCH_DATA_REMAINS)
+ return status;
+ dbus_mutex_lock (connection->mutex);
+
/* We need to ref the connection since the callback could potentially
* drop the last ref to it
*/
/* This call may drop the lock during the execution (if waiting for
* borrowed messages to be returned) but the order of message
- * dispatch if several threads call dispatch_message is still
+ * dispatch if several threads call dispatch() is still
* protected by the lock, since only one will get the lock, and that
* one will finish the message dispatching
*/
message_link = _dbus_connection_pop_message_link_unlocked (connection);
if (message_link == NULL)
{
+ /* another thread dispatched our stuff */
+
_dbus_connection_release_dispatch (connection);
dbus_mutex_unlock (connection->mutex);
+
+ status = dbus_connection_get_dispatch_status (connection);
+
dbus_connection_unref (connection);
- return FALSE;
+
+ return status;
}
message = message_link->data;
dbus_mutex_unlock (connection->mutex);
_dbus_connection_failed_pop (connection, message_link);
dbus_connection_unref (connection);
- return FALSE;
+ return DBUS_DISPATCH_NEED_MEMORY;
}
_dbus_list_foreach (&filter_list_copy,
(DBusForeachFunction)dbus_message_handler_ref,
NULL);
- /* We're still protected from dispatch_message reentrancy here
+ /* We're still protected from dispatch() reentrancy here
* since we acquired the dispatcher
*/
dbus_mutex_unlock (connection->mutex);
name);
if (handler != NULL)
{
- /* We're still protected from dispatch_message reentrancy here
- * since we acquired the dispatcher */
+ /* We're still protected from dispatch() reentrancy here
+ * since we acquired the dispatcher
+ */
dbus_mutex_unlock (connection->mutex);
_dbus_verbose (" running app handler on message %p\n", message);
_dbus_connection_release_dispatch (connection);
dbus_mutex_unlock (connection->mutex);
_dbus_list_free_link (message_link);
+ dbus_message_unref (message); /* don't want the message to count in max message limits
+ * in computing dispatch status
+ */
+
+ status = dbus_connection_get_dispatch_status (connection);
+
dbus_connection_unref (connection);
- dbus_message_unref (message);
- return connection->n_incoming > 0;
+ return status;
}
/**
* and that contains a half-dozen small messages, we may exceed the
* size max by that amount. But this should be inconsequential.
*
+ * This does imply that we can't call read() with a buffer larger
+ * than we're willing to exceed this limit by.
+ *
* @param connection the connection
* @param size the maximum size in bytes of all outstanding messages
*/
* can be present in current state). */
} DBusWatchFlags;
+typedef enum
+{
+ DBUS_DISPATCH_DATA_REMAINS, /**< There is more data to potentially convert to messages. */
+ DBUS_DISPATCH_COMPLETE, /**< All currently available data has been processed. */
+ DBUS_DISPATCH_NEED_MEMORY /**< More memory is needed to continue. */
+} DBusDispatchStatus;
+
typedef dbus_bool_t (* DBusAddWatchFunction) (DBusWatch *watch,
void *data);
typedef void (* DBusWatchToggledFunction) (DBusWatch *watch,
typedef void (* DBusRemoveTimeoutFunction) (DBusTimeout *timeout,
void *data);
-DBusConnection* dbus_connection_open (const char *address,
- DBusResultCode *result);
-void dbus_connection_ref (DBusConnection *connection);
-void dbus_connection_unref (DBusConnection *connection);
-void dbus_connection_disconnect (DBusConnection *connection);
-dbus_bool_t dbus_connection_get_is_connected (DBusConnection *connection);
-dbus_bool_t dbus_connection_get_is_authenticated (DBusConnection *connection);
-void dbus_connection_flush (DBusConnection *connection);
-int dbus_connection_get_n_messages (DBusConnection *connection);
-DBusMessage* dbus_connection_borrow_message (DBusConnection *connection);
-void dbus_connection_return_message (DBusConnection *connection,
- DBusMessage *message);
-void dbus_connection_steal_borrowed_message (DBusConnection *connection,
- DBusMessage *message);
-DBusMessage* dbus_connection_pop_message (DBusConnection *connection);
-dbus_bool_t dbus_connection_dispatch_message (DBusConnection *connection);
-
-
-dbus_bool_t dbus_connection_send (DBusConnection *connection,
- DBusMessage *message,
- dbus_int32_t *client_serial);
-dbus_bool_t dbus_connection_send_with_reply (DBusConnection *connection,
- DBusMessage *message,
- DBusMessageHandler *reply_handler,
- int timeout_milliseconds);
-DBusMessage *dbus_connection_send_with_reply_and_block (DBusConnection *connection,
- DBusMessage *message,
- int timeout_milliseconds,
- DBusError *error);
-
-
-dbus_bool_t dbus_connection_set_watch_functions (DBusConnection *connection,
- DBusAddWatchFunction add_function,
- DBusRemoveWatchFunction remove_function,
- DBusWatchToggledFunction toggled_function,
- void *data,
- DBusFreeFunction free_data_function);
-dbus_bool_t dbus_connection_set_timeout_functions (DBusConnection *connection,
- DBusAddTimeoutFunction add_function,
- DBusRemoveTimeoutFunction remove_function,
- DBusTimeoutToggledFunction toggled_function,
- void *data,
- DBusFreeFunction free_data_function);
-void dbus_connection_set_wakeup_main_function (DBusConnection *connection,
- DBusWakeupMainFunction wakeup_main_function,
- void *data,
- DBusFreeFunction free_data_function);
-void dbus_connection_handle_watch (DBusConnection *connection,
- DBusWatch *watch,
- unsigned int condition);
-
-
+DBusConnection* dbus_connection_open (const char *address,
+ DBusResultCode *result);
+void dbus_connection_ref (DBusConnection *connection);
+void dbus_connection_unref (DBusConnection *connection);
+void dbus_connection_disconnect (DBusConnection *connection);
+dbus_bool_t dbus_connection_get_is_connected (DBusConnection *connection);
+dbus_bool_t dbus_connection_get_is_authenticated (DBusConnection *connection);
+void dbus_connection_flush (DBusConnection *connection);
+DBusMessage* dbus_connection_borrow_message (DBusConnection *connection);
+void dbus_connection_return_message (DBusConnection *connection,
+ DBusMessage *message);
+void dbus_connection_steal_borrowed_message (DBusConnection *connection,
+ DBusMessage *message);
+DBusMessage* dbus_connection_pop_message (DBusConnection *connection);
+DBusDispatchStatus dbus_connection_get_dispatch_status (DBusConnection *connection);
+DBusDispatchStatus dbus_connection_dispatch (DBusConnection *connection);
+dbus_bool_t dbus_connection_send (DBusConnection *connection,
+ DBusMessage *message,
+ dbus_int32_t *client_serial);
+dbus_bool_t dbus_connection_send_with_reply (DBusConnection *connection,
+ DBusMessage *message,
+ DBusMessageHandler *reply_handler,
+ int timeout_milliseconds);
+DBusMessage * dbus_connection_send_with_reply_and_block (DBusConnection *connection,
+ DBusMessage *message,
+ int timeout_milliseconds,
+ DBusError *error);
+dbus_bool_t dbus_connection_set_watch_functions (DBusConnection *connection,
+ DBusAddWatchFunction add_function,
+ DBusRemoveWatchFunction remove_function,
+ DBusWatchToggledFunction toggled_function,
+ void *data,
+ DBusFreeFunction free_data_function);
+dbus_bool_t dbus_connection_set_timeout_functions (DBusConnection *connection,
+ DBusAddTimeoutFunction add_function,
+ DBusRemoveTimeoutFunction remove_function,
+ DBusTimeoutToggledFunction toggled_function,
+ void *data,
+ DBusFreeFunction free_data_function);
+void dbus_connection_set_wakeup_main_function (DBusConnection *connection,
+ DBusWakeupMainFunction wakeup_main_function,
+ void *data,
+ DBusFreeFunction free_data_function);
+void dbus_connection_handle_watch (DBusConnection *connection,
+ DBusWatch *watch,
+ unsigned int condition);
int dbus_watch_get_fd (DBusWatch *watch);
fail_alloc_counter = until_next_fail;
+#if 0
_dbus_verbose ("Set fail alloc counter = %d\n", fail_alloc_counter);
+#endif
}
/**
DBusFreedElement *free_elements; /**< a free list of elements to recycle */
DBusMemBlock *blocks; /**< blocks of memory from malloc() */
- int allocated_elements; /* Count of outstanding allocated elements */
+ int allocated_elements; /**< Count of outstanding allocated elements */
};
/** @} */
dbus_mutex_unlock (message_handler_lock);
/* This function doesn't ref handler/connection/message
- * since that's done in dbus_connection_dispatch_message().
+ * since that's done in dbus_connection_dispatch().
*/
if (function != NULL)
return (* function) (handler, connection, message, user_data);
void _dbus_message_loader_return_buffer (DBusMessageLoader *loader,
DBusString *buffer,
int bytes_read);
-
+dbus_bool_t _dbus_message_loader_queue_messages (DBusMessageLoader *loader);
+DBusMessage* _dbus_message_loader_peek_message (DBusMessageLoader *loader);
DBusMessage* _dbus_message_loader_pop_message (DBusMessageLoader *loader);
DBusList* _dbus_message_loader_pop_message_link (DBusMessageLoader *loader);
* in. This function must always be called, even if no bytes were
* successfully read.
*
- * @todo if we run out of memory in here, we offer no way for calling
- * code to handle it, i.e. they can't re-run the message parsing
- * attempt. Perhaps much of this code could be moved to pop_message()?
- * But then that may need to distinguish NULL return for no messages
- * from NULL return for errors.
- *
* @param loader the loader.
* @param buffer the buffer.
* @param bytes_read number of bytes that were read into the buffer.
_dbus_assert (buffer == &loader->data);
loader->buffer_outstanding = FALSE;
+}
+/**
+ * Converts buffered data into messages.
+ *
+ * @param loader the loader.
+ * @returns #TRUE if we had enough memory to finish.
+ */
+dbus_bool_t
+_dbus_message_loader_queue_messages (DBusMessageLoader *loader)
+{
if (loader->corrupted)
- return;
+ return TRUE;
while (_dbus_string_get_length (&loader->data) >= 16)
{
_dbus_verbose ("Message has protocol version %d ours is %d\n",
(int) header_data[2], DBUS_MAJOR_PROTOCOL_VERSION);
loader->corrupted = TRUE;
- return;
+ return TRUE;
}
byte_order = header_data[0];
_dbus_verbose ("Message with bad byte order '%c' received\n",
byte_order);
loader->corrupted = TRUE;
- return;
+ return TRUE;
}
header_len_unsigned = _dbus_unpack_uint32 (byte_order, header_data + 4);
_dbus_verbose ("Message had broken too-small header length %u\n",
header_len_unsigned);
loader->corrupted = TRUE;
- return;
+ return TRUE;
}
if (header_len_unsigned > (unsigned) MAX_SANE_MESSAGE_SIZE ||
header_len_unsigned,
body_len_unsigned);
loader->corrupted = TRUE;
- return;
+ return TRUE;
}
/* Now that we know the values are in signed range, get
_dbus_verbose ("header length %d is not aligned to 8 bytes\n",
header_len);
loader->corrupted = TRUE;
- return;
+ return TRUE;
}
if (header_len + body_len > loader->max_message_size)
_dbus_verbose ("Message claimed length header = %d body = %d exceeds max message length %d\n",
header_len, body_len, loader->max_message_size);
loader->corrupted = TRUE;
- return;
+ return TRUE;
}
if (_dbus_string_get_length (&loader->data) >= (header_len + body_len))
{
_dbus_verbose ("Header was invalid\n");
loader->corrupted = TRUE;
- return;
+ return TRUE;
}
next_arg = header_len;
&next_arg))
{
loader->corrupted = TRUE;
- return;
+ return TRUE;
}
_dbus_assert (next_arg > prev);
next_arg, header_len, body_len,
header_len + body_len);
loader->corrupted = TRUE;
- return;
+ return TRUE;
}
message = dbus_message_new_empty_header ();
if (message == NULL)
- break; /* ugh, postpone this I guess. */
+ return FALSE;
message->byte_order = byte_order;
message->header_padding = header_padding;
if (!_dbus_list_append (&loader->messages, message))
{
dbus_message_unref (message);
- break;
+ return FALSE;
}
_dbus_assert (_dbus_string_get_length (&message->header) == 0);
{
_dbus_list_remove_last (&loader->messages, message);
dbus_message_unref (message);
- break;
+ return FALSE;
}
if (!_dbus_string_move_len (&loader->data, 0, body_len, &message->body, 0))
_dbus_list_remove_last (&loader->messages, message);
dbus_message_unref (message);
- break;
+ return FALSE;
}
_dbus_assert (_dbus_string_get_length (&message->header) == header_len);
_dbus_verbose ("Loaded message %p\n", message);
}
else
- break;
+ return TRUE;
}
+
+ return TRUE;
+}
+
+/**
+ * Peeks at first loaded message, returns #NULL if no messages have
+ * been queued.
+ *
+ * @param loader the loader.
+ * @returns the next message, or #NULL if none.
+ */
+DBusMessage*
+_dbus_message_loader_peek_message (DBusMessageLoader *loader)
+{
+ if (loader->messages)
+ return loader->messages->data;
+ else
+ return NULL;
}
/**
* Pops a loaded message (passing ownership of the message
* to the caller). Returns #NULL if no messages have been
- * loaded.
+ * queued.
*
* @param loader the loader.
* @returns the next message, or #NULL if none.
message = NULL;
retval = FALSE;
+
+ if (!_dbus_message_loader_queue_messages (loader))
+ _dbus_assert_not_reached ("no memory to queue messages");
if (_dbus_message_loader_get_is_corrupted (loader))
{
dbus_bool_t retval;
retval = FALSE;
+
+ if (!_dbus_message_loader_queue_messages (loader))
+ _dbus_assert_not_reached ("no memory to queue messages");
if (!_dbus_message_loader_get_is_corrupted (loader))
{
message = NULL;
retval = FALSE;
+
+ if (!_dbus_message_loader_queue_messages (loader))
+ _dbus_assert_not_reached ("no memory to queue messages");
if (_dbus_message_loader_get_is_corrupted (loader))
{
check_loader_results (DBusMessageLoader *loader,
DBusMessageValidity validity)
{
+ if (!_dbus_message_loader_queue_messages (loader))
+ _dbus_assert_not_reached ("no memory to queue messages");
+
switch (validity)
{
case _DBUS_MESSAGE_VALID:
dbus_message_unref (message);
/* Now pop back the message */
+ if (!_dbus_message_loader_queue_messages (loader))
+ _dbus_assert_not_reached ("no memory to queue messages");
+
if (_dbus_message_loader_get_is_corrupted (loader))
_dbus_assert_not_reached ("message loader corrupted");
* function on a watch that was not previously added.
*
* @param server the server.
- * @param timeout the timeout to toggle.
+ * @param watch the watch to toggle.
* @param enabled whether to enable or disable
*/
void
int len)
{
const unsigned char *p;
+ const unsigned char *end;
DBUS_CONST_STRING_PREAMBLE (str);
_dbus_assert (start >= 0);
_dbus_assert (start <= real->len);
if (len > real->len - start)
return FALSE;
- p = real->str;
+ p = real->str + start;
+ end = p + len;
- while (p - real->str < len && *p)
+ while (p < end)
{
int i, mask = 0, char_len;
dbus_unichar_t result;
if (char_len == -1)
break;
- /* check that the expected number of bytes exists in real->str */
- if ((len - (p - real->str)) < char_len)
+ /* check that the expected number of bytes exists in the remaining length */
+ if ((end - p) < char_len)
break;
UTF8_GET (result, p, i, mask, char_len);
if (UTF8_LENGTH (result) != char_len) /* Check for overlong UTF-8 */
- break;
+ break;
if (result == (dbus_unichar_t)-1)
break;
if (!UNICODE_VALID (result))
- break;
+ break;
p += char_len;
}
/* See that we covered the entire length if a length was
* passed in
*/
- if (p != (real->str + len))
+ if (p != end)
return FALSE;
else
return TRUE;
_dbus_transport_unref (transport);
}
-static void
-queue_messages (DBusTransport *transport)
-{
- DBusList *link;
-
- /* Queue any messages */
- while ((link = _dbus_message_loader_pop_message_link (transport->loader)))
- {
- DBusMessage *message;
-
- message = link->data;
-
- _dbus_verbose ("queueing received message %p\n", message);
-
- _dbus_message_add_size_counter (message, transport->live_messages_size);
-
- /* pass ownership of link and message ref to connection */
- _dbus_connection_queue_received_message_link (transport->connection,
- link);
- }
-
- if (_dbus_message_loader_get_is_corrupted (transport->loader))
- {
- _dbus_verbose ("Corrupted message stream, disconnecting\n");
- do_io_error (transport);
- }
-
- /* check read watch in case we've now exceeded max outstanding messages */
- check_read_watch (transport);
-}
-
/* return value is whether we successfully read any new data. */
static dbus_bool_t
read_data_into_auth (DBusTransport *transport)
orig_len);
}
- queue_messages (transport);
-
return;
nomem:
total += bytes_read;
- queue_messages (transport);
+ if (_dbus_transport_queue_messages (transport) == DBUS_DISPATCH_NEED_MEMORY)
+ goto out;
/* Try reading more data until we get EAGAIN and return, or
* exceed max bytes per iteration. If in blocking mode of
}
/**
+ * Reports our current dispatch status (whether there's buffered
+ * data to be queued as messages, or not, or we need memory).
+ *
+ * @param transport the transport
+ * @returns current status
+ */
+DBusDispatchStatus
+_dbus_transport_get_dispatch_status (DBusTransport *transport)
+{
+ if (_dbus_counter_get_value (transport->live_messages_size) >= transport->max_live_messages_size)
+ return DBUS_DISPATCH_COMPLETE; /* complete for now */
+
+ if (!_dbus_message_loader_queue_messages (transport->loader))
+ return DBUS_DISPATCH_NEED_MEMORY;
+
+ if (_dbus_message_loader_peek_message (transport->loader) != NULL)
+ return DBUS_DISPATCH_DATA_REMAINS;
+ else
+ return DBUS_DISPATCH_COMPLETE;
+}
+
+/**
+ * Processes data we've read while handling a watch, potentially
+ * converting some of it to messages and queueing those messages on
+ * the connection.
+ *
+ * @param transport the transport
+ * @returns #TRUE if we had enough memory to queue all messages
+ */
+dbus_bool_t
+_dbus_transport_queue_messages (DBusTransport *transport)
+{
+ DBusDispatchStatus status;
+
+ /* Queue any messages */
+ while ((status = _dbus_transport_get_dispatch_status (transport)) == DBUS_DISPATCH_DATA_REMAINS)
+ {
+ DBusMessage *message;
+ DBusList *link;
+
+ link = _dbus_message_loader_pop_message_link (transport->loader);
+ _dbus_assert (link != NULL);
+
+ message = link->data;
+
+ _dbus_verbose ("queueing received message %p\n", message);
+
+ _dbus_message_add_size_counter (message, transport->live_messages_size);
+
+ /* pass ownership of link and message ref to connection */
+ _dbus_connection_queue_received_message_link (transport->connection,
+ link);
+ }
+
+ if (_dbus_message_loader_get_is_corrupted (transport->loader))
+ {
+ _dbus_verbose ("Corrupted message stream, disconnecting\n");
+ _dbus_transport_disconnect (transport);
+ }
+
+ return status != DBUS_DISPATCH_NEED_MEMORY;
+}
+
+/**
* See dbus_connection_set_max_message_size().
*
* @param transport the transport
typedef struct DBusTransport DBusTransport;
-DBusTransport* _dbus_transport_open (const char *address,
- DBusResultCode *result);
-void _dbus_transport_ref (DBusTransport *transport);
-void _dbus_transport_unref (DBusTransport *transport);
-void _dbus_transport_disconnect (DBusTransport *transport);
-dbus_bool_t _dbus_transport_get_is_connected (DBusTransport *transport);
-dbus_bool_t _dbus_transport_get_is_authenticated (DBusTransport *transport);
-void _dbus_transport_handle_watch (DBusTransport *transport,
- DBusWatch *watch,
- unsigned int condition);
-dbus_bool_t _dbus_transport_set_connection (DBusTransport *transport,
- DBusConnection *connection);
-void _dbus_transport_messages_pending (DBusTransport *transport,
- int queue_length);
-void _dbus_transport_do_iteration (DBusTransport *transport,
- unsigned int flags,
- int timeout_milliseconds);
-void _dbus_transport_set_max_message_size (DBusTransport *transport,
- long size);
-long _dbus_transport_get_max_message_size (DBusTransport *transport);
-void _dbus_transport_set_max_live_messages_size (DBusTransport *transport,
- long size);
-long _dbus_transport_get_max_live_messages_size (DBusTransport *transport);
+DBusTransport* _dbus_transport_open (const char *address,
+ DBusResultCode *result);
+void _dbus_transport_ref (DBusTransport *transport);
+void _dbus_transport_unref (DBusTransport *transport);
+void _dbus_transport_disconnect (DBusTransport *transport);
+dbus_bool_t _dbus_transport_get_is_connected (DBusTransport *transport);
+dbus_bool_t _dbus_transport_get_is_authenticated (DBusTransport *transport);
+void _dbus_transport_handle_watch (DBusTransport *transport,
+ DBusWatch *watch,
+ unsigned int condition);
+dbus_bool_t _dbus_transport_set_connection (DBusTransport *transport,
+ DBusConnection *connection);
+void _dbus_transport_messages_pending (DBusTransport *transport,
+ int queue_length);
+void _dbus_transport_do_iteration (DBusTransport *transport,
+ unsigned int flags,
+ int timeout_milliseconds);
+DBusDispatchStatus _dbus_transport_get_dispatch_status (DBusTransport *transport);
+dbus_bool_t _dbus_transport_queue_messages (DBusTransport *transport);
+void _dbus_transport_set_max_message_size (DBusTransport *transport,
+ long size);
+long _dbus_transport_get_max_message_size (DBusTransport *transport);
+void _dbus_transport_set_max_live_messages_size (DBusTransport *transport,
+ long size);
+long _dbus_transport_get_max_live_messages_size (DBusTransport *transport);
DBUS_END_DECLS;
static GStaticMutex server_slot_lock = G_STATIC_MUTEX_INIT;
static int server_slot = -1;
-static gboolean dbus_connection_prepare (GSource *source,
- gint *timeout);
-static gboolean dbus_connection_check (GSource *source);
-static gboolean dbus_connection_dispatch (GSource *source,
- GSourceFunc callback,
- gpointer user_data);
-static gboolean dbus_server_prepare (GSource *source,
- gint *timeout);
-static gboolean dbus_server_check (GSource *source);
-static gboolean dbus_server_dispatch (GSource *source,
- GSourceFunc callback,
- gpointer user_data);
+static gboolean gsource_connection_prepare (GSource *source,
+ gint *timeout);
+static gboolean gsource_connection_check (GSource *source);
+static gboolean gsource_connection_dispatch (GSource *source,
+ GSourceFunc callback,
+ gpointer user_data);
+static gboolean gsource_server_prepare (GSource *source,
+ gint *timeout);
+static gboolean gsource_server_check (GSource *source);
+static gboolean gsource_server_dispatch (GSource *source,
+ GSourceFunc callback,
+ gpointer user_data);
static GSourceFuncs dbus_connection_funcs = {
- dbus_connection_prepare,
- dbus_connection_check,
- dbus_connection_dispatch,
+ gsource_connection_prepare,
+ gsource_connection_check,
+ gsource_connection_dispatch,
NULL
};
static GSourceFuncs dbus_server_funcs = {
- dbus_server_prepare,
- dbus_server_check,
- dbus_server_dispatch,
+ gsource_server_prepare,
+ gsource_server_check,
+ gsource_server_dispatch,
NULL
};
static gboolean
-dbus_connection_prepare (GSource *source,
- gint *timeout)
+gsource_connection_prepare (GSource *source,
+ gint *timeout)
{
DBusConnection *connection = ((DBusGSource *)source)->connection_or_server;
*timeout = -1;
- return (dbus_connection_get_n_messages (connection) > 0);
+ return (dbus_connection_get_dispatch_status (connection) == DBUS_DISPATCH_DATA_REMAINS);
}
static gboolean
-dbus_server_prepare (GSource *source,
- gint *timeout)
+gsource_server_prepare (GSource *source,
+ gint *timeout)
{
*timeout = -1;
}
static gboolean
-dbus_connection_check (GSource *source)
+gsource_connection_check (GSource *source)
{
return dbus_gsource_check (source);
}
static gboolean
-dbus_server_check (GSource *source)
+gsource_server_check (GSource *source)
{
return dbus_gsource_check (source);
}
}
static gboolean
-dbus_connection_dispatch (GSource *source,
- GSourceFunc callback,
- gpointer user_data)
+gsource_connection_dispatch (GSource *source,
+ GSourceFunc callback,
+ gpointer user_data)
{
DBusGSource *dbus_source = (DBusGSource *)source;
DBusConnection *connection = dbus_source->connection_or_server;
FALSE);
/* Dispatch messages */
- while (dbus_connection_dispatch_message (connection))
+ while (dbus_connection_dispatch (connection) == DBUS_DISPATCH_DATA_REMAINS)
;
dbus_connection_unref (connection);
}
static gboolean
-dbus_server_dispatch (GSource *source,
- GSourceFunc callback,
- gpointer user_data)
+gsource_server_dispatch (GSource *source,
+ GSourceFunc callback,
+ gpointer user_data)
{
DBusGSource *dbus_source = (DBusGSource *)source;
DBusServer *server = dbus_source->connection_or_server;
projects / platform / upstream / dbus.git / commitdiff