static void pcmk_dbus_connection_dispatch(DBusConnection *connection, DBusDispatchStatus new_status, void *data){
    crm_trace("status %d for %p", new_status, data);
    if (new_status == DBUS_DISPATCH_DATA_REMAINS){
        dbus_connection_dispatch(connection);

        while (dbus_connection_get_dispatch_status(connection) == DBUS_DISPATCH_DATA_REMAINS) {
            dbus_connection_dispatch(connection);
        }
    }
}

/* Callback: "glib says dbus fd is active" */
static gboolean handle_dbus_fd(GIOChannel *gio, GIOCondition condition, gpointer data)
{
    DBusWatch *watch = (DBusWatch*)data;

    log_debug("%s(gio, condition:%x [bits:IN/PRI/OUT/ERR/HUP...], data)", __func__, (int)condition);

    /* Notify the D-Bus library when a previously-added watch
     * is ready for reading or writing, or has an exception such as a hangup.
     */
    int glib_flags = (int)condition;
    int dbus_flags = 0;
    if (glib_flags & G_IO_IN)  dbus_flags |= DBUS_WATCH_READABLE;
    if (glib_flags & G_IO_OUT) dbus_flags |= DBUS_WATCH_WRITABLE;
    if (glib_flags & G_IO_ERR) dbus_flags |= DBUS_WATCH_ERROR;
    if (glib_flags & G_IO_HUP) dbus_flags |= DBUS_WATCH_HANGUP;
    /*
     * TODO:
     * If dbus_watch_handle returns FALSE, then the file descriptor
     * may still be ready for reading or writing, but more memory
     * is needed in order to do the reading or writing. If you ignore
     * the FALSE return, your application may spin in a busy loop
     * on the file descriptor until memory becomes available,
     * but nothing more catastrophic should happen.
     */
    dbus_watch_handle(watch, dbus_flags);

    while (dbus_connection_dispatch(g_dbus_conn) == DBUS_DISPATCH_DATA_REMAINS)
        log_debug("%s: more data to process, looping", __func__);
    return TRUE; /* "glib, do not remove this event source!" */
}

static void read_watch_cb(int fd, void* d) { 
	/* E_DEBUG(E_STRLOC ": read_watch_cb()\n"); */

	EdbusConnImpl* dc = (EdbusConnImpl*)d;
	E_ASSERT(dc != NULL);
	E_ASSERT(dc->watch_list != NULL);

	WatchListIter it = dc->watch_list->begin(), it_end = dc->watch_list->end();
	while(it != it_end) {
		if(dbus_watch_get_fd(*it) == fd && dbus_watch_get_enabled(*it)) {
			if(!dbus_watch_handle(*it, DBUS_WATCH_READABLE))
				E_WARNING(E_STRLOC ": Out of memory\n");
			break;
		}
		++it;
	}

	/*
	 * Check if there are more incomming data and process them. Note that 
	 * dbus_connection_dispatch() call will also remove data from queue. 
	 * This means that (here) timer will not be installed if only one unprocessed
	 * message is in queue; opposite, after installment it will process the rest
	 * of the messages without interrupting read_watch_cb() flow.
	 *
	 * If this is not set (e.g. all data are processed here), we can miss initial
	 * (or later) messages that are sent to us. Also, timer will be triggered faster
	 * as it can (seems that 0.5 as timer value misses some data...).
	 */
	if(dbus_connection_dispatch(dc->conn) == DBUS_DISPATCH_DATA_REMAINS)
		Fl::add_timeout(0.2, dispatch_cb, dc);
}

static void dispatch_cb(pa_mainloop_api *ea, pa_defer_event *ev, void *userdata) {
    DBusConnection *conn = userdata;

    if (dbus_connection_dispatch(conn) == DBUS_DISPATCH_COMPLETE)
        /* no more data to process, disable the deferred */
        ea->defer_enable(ev, 0);
}

void check_dbus_listeners(fd_set *rset, fd_set *wset, fd_set *eset)
{
  DBusConnection *connection = (DBusConnection *)daemon->dbus;
  struct watch *w;

  for (w = daemon->watches; w; w = w->next)
    if (dbus_watch_get_enabled(w->watch))
      {
	unsigned int flags = 0;
	int fd = dbus_watch_get_unix_fd(w->watch);
	
	if (FD_ISSET(fd, rset))
	  flags |= DBUS_WATCH_READABLE;
	
	if (FD_ISSET(fd, wset))
	  flags |= DBUS_WATCH_WRITABLE;
	
	if (FD_ISSET(fd, eset))
	  flags |= DBUS_WATCH_ERROR;

	if (flags != 0)
	  dbus_watch_handle(w->watch, flags);
      }

  if (connection)
    {
      dbus_connection_ref (connection);
      while (dbus_connection_dispatch (connection) == DBUS_DISPATCH_DATA_REMAINS);
      dbus_connection_unref (connection);
    }
}

static void handle_watch(struct sock_com *s, short revents)
{
	struct DBusWatch *w = s->data;
	unsigned int flags = 0;

	if(!w)
		return;
	if(!dbus_watch_get_enabled(w)) {
		s->enabled = false;
		return;
	}

	flags |= revents & POLLIN  ? DBUS_WATCH_READABLE : 0;
	flags |= revents & POLLOUT ? DBUS_WATCH_WRITABLE : 0;
	flags |= revents & POLLERR ? DBUS_WATCH_ERROR    : 0;
	flags |= revents & POLLHUP ? DBUS_WATCH_HANGUP   : 0;
	if(flags)
		dbus_watch_handle(w, flags);
	if(idbus_connection)
	{
		dbus_connection_ref(idbus_connection);
		while(DBUS_DISPATCH_DATA_REMAINS == dbus_connection_dispatch(idbus_connection)) {
			/* nop */;
		}
		dbus_connection_unref(idbus_connection);
	}
}

static void mysleep(int msec)
{
  struct timeval tmo;
  struct timeval t1,t2;
  int ms;

  //printf("SLEEP %d\n", msec);
  timeout_init(&tmo, msec);
  getmonotime(&t1,0);

  while( (ms = timeout_left(&tmo)) > 0 )
  {
    printf("dbus wait io %d\n", ms);
    dbus_connection_read_write(connection, ms);
    do
    {
      printf("dbus dispatch\n");
    }
    while( dbus_connection_dispatch(connection) == DBUS_DISPATCH_DATA_REMAINS );
  }

  getmonotime(&t2,0);
  timersub(&t2,&t1,&t1);
  printf("SLEPT  %.3f / %.3f s\n", t1.tv_sec + t1.tv_usec * 1e-6, msec * 1e-3);

}

void DBusProcessEventForConnection(DBusConnection* connection)
{
    if (connection) {
        dbus_connection_ref(connection);
        while (dbus_connection_dispatch(connection) == DBUS_DISPATCH_DATA_REMAINS);
        dbus_connection_unref(connection);
    }
}

/**
 * dispatch_initial_dbus_messages - Dispatch initial dbus messages after
 *     claiming bus name
 * @eloop_ctx: the DBusConnection to dispatch on
 * @timeout_ctx: unused
 *
 * If clients are quick to notice that wpa_supplicant claimed its bus name,
 * there may have been messages that came in before initialization was
 * all finished.  Dispatch those here.
 */
static void dispatch_initial_dbus_messages(void *eloop_ctx, void *timeout_ctx)
{
	DBusConnection *con = eloop_ctx;

	while (dbus_connection_get_dispatch_status(con) ==
	       DBUS_DISPATCH_DATA_REMAINS)
		dbus_connection_dispatch(con);
}

dbus_bool_t
bus_connection_dispatch_one_message  (DBusConnection *connection)
{
  DBusDispatchStatus status;

  while ((status = dbus_connection_dispatch (connection)) == DBUS_DISPATCH_NEED_MEMORY)
    _dbus_wait_for_memory ();
  
  return status == DBUS_DISPATCH_DATA_REMAINS;
}

void
DBusThread::EventLoop()
{
  dbus_connection_set_watch_functions(mConnection, AddWatch,
                                      RemoveWatch, ToggleWatch, this, NULL);
  dbus_connection_set_wakeup_main_function(mConnection, DBusWakeup, this, NULL);
#ifdef DEBUG
  LOG("DBus Event Loop Starting\n");
#endif
  while (1) {
    poll(mPollData.Elements(), mPollData.Length(), -1);

    for (uint32_t i = 0; i < mPollData.Length(); i++) {
      if (!mPollData[i].revents) {
        continue;
      }

      if (mPollData[i].fd == mControlFdR.get()) {
        char data;
        while (recv(mControlFdR.get(), &data, sizeof(char), MSG_DONTWAIT)
               != -1) {
          switch (data) {
          case DBUS_EVENT_LOOP_EXIT:
#ifdef DEBUG
            LOG("DBus Event Loop Exiting\n");
#endif
            dbus_connection_set_watch_functions(mConnection,
                                                NULL, NULL, NULL, NULL, NULL);
            return;
          case DBUS_EVENT_LOOP_ADD:
            HandleWatchAdd(this);
            break;
          case DBUS_EVENT_LOOP_REMOVE:
            HandleWatchRemove(this);
            break;
          case DBUS_EVENT_LOOP_WAKEUP:
            // noop
            break;
          }
        }
      } else {
        short events = mPollData[i].revents;
        unsigned int flags = UnixEventsToDBusFlags(events);
        dbus_watch_handle(mWatchData[i], flags);
        mPollData[i].revents = 0;
        // Break at this point since we don't know if the operation
        // was destructive
        break;
      }
    }
    while (dbus_connection_dispatch(mConnection) ==
           DBUS_DISPATCH_DATA_REMAINS)
    {}
  }
}

static void dispatch_cb(void* d) {
	EdbusConnImpl* dc = (EdbusConnImpl*)d;
	E_ASSERT(dc != NULL);

	/* E_DEBUG(E_STRLOC ": dispatch_cb()\n"); */

	while(dbus_connection_dispatch(dc->conn) == DBUS_DISPATCH_DATA_REMAINS)
		;

	Fl::remove_timeout(dispatch_cb);
}

static gboolean message_dispatch(void *data)
{
	DBusConnection *conn = data;

	/* Dispatch messages */
	while (dbus_connection_dispatch(conn) == DBUS_DISPATCH_DATA_REMAINS);

	dbus_connection_unref(conn);

	return FALSE;
}

static gboolean
message_queue_dispatch (GSource *source, GSourceFunc  callback, gpointer user_data)
{
	DBusConnection *connection = ((DBusGMessageQueue *)source)->connection;
	dbus_connection_ref (connection);

	/* Only dispatch once - we don't want to starve other GSource */
	dbus_connection_dispatch (connection);
	dbus_connection_unref (connection);
	return TRUE;
}

int init_dbus()
{
    DBusConnection* conn;
    DBusError err;

    dbus_error_init(&err);
    VERB3 log("dbus_bus_get");
    conn = dbus_bus_get(DBUS_BUS_SYSTEM, &err);
    handle_dbus_err(conn == NULL, &err);
    // dbus api says:
    // "If dbus_bus_get obtains a new connection object never before returned
    // from dbus_bus_get(), it will call dbus_connection_set_exit_on_disconnect(),
    // so the application will exit if the connection closes. You can undo this
    // by calling dbus_connection_set_exit_on_disconnect() yourself after you get
    // the connection."
    // ...
    // "When a connection is disconnected, you are guaranteed to get a signal
    // "Disconnected" from the interface DBUS_INTERFACE_LOCAL, path DBUS_PATH_LOCAL"
    //
    // dbus-daemon drops connections if it recvs a malformed message
    // (we actually observed this when we sent bad UTF-8 string).
    // Currently, in this case abrtd just exits with exit code 1.
    // (symptom: last two log messages are "abrtd: remove_watch()")
    // If we want to have better logging or other nontrivial handling,
    // here we need to do:
    //
    //dbus_connection_set_exit_on_disconnect(conn, FALSE);
    //dbus_connection_add_filter(conn, handle_message, NULL, NULL);
    //
    // and need to code up handle_message to check for "Disconnected" dbus signal

    /* Also sets g_dbus_conn to conn. */
    attach_dbus_conn_to_glib_main_loop(conn, "/com/redhat/abrt", message_received);

    VERB3 log("dbus_bus_request_name");
    int rc = dbus_bus_request_name(conn, ABRTD_DBUS_NAME, DBUS_NAME_FLAG_REPLACE_EXISTING, &err);
//maybe check that r == DBUS_REQUEST_NAME_REPLY_PRIMARY_OWNER instead?
    handle_dbus_err(rc < 0, &err);
    VERB3 log("dbus init done");

    /* dbus_bus_request_name can already read some data. For example,
     * if we were autostarted, the call which caused autostart arrives
     * at this moment. Thus while dbus fd hasn't any data anymore,
     * dbus library can buffer a message or two.
     * If we don't do this, the data won't be processed
     * until next dbus data arrives.
     */
    int cnt = 10;
    while (dbus_connection_dispatch(conn) != DBUS_DISPATCH_COMPLETE && --cnt)
        VERB3 log("processed initial buffered dbus message");

    return 0;
}

static void MainLoop() {
  struct ConnectionData* c = (struct ConnectionData*) FindTask(NULL)->tc_UserData;
  kprintf("MainLoop started %08lx (cd %08lx)\n", FindTask(NULL), c);

  c->signal = AllocSignal(-1);

  if (c->signal == -1) {
    goto exit;
  }
  
  Signal(c->creator, SIGF_SINGLE);
  
  while(TRUE) {
    ULONG signals = Wait(SIGBREAKF_CTRL_C | (1UL << c->signal));

    kprintf("MainLoop got a signal %lx\n", signals);

    if (signals & SIGBREAKF_CTRL_C) {
      break;
    }

    if (signals & (1UL << c->signal)) {
      struct WatchData* w;
      
//      dbus_connection_ref(c->connection);

      kprintf("Checking watches\n");
      for(w = (struct WatchData*) c->watches.mlh_Head;
	  w->node.mln_Succ != NULL;
	  w = (struct WatchData*) w->node.mln_Succ) {
	kprintf("%s watch on fd %ld, flags %lx\n",
		w->enabled ? "Enabled" : "Disabled",
		dbus_watch_get_fd(w->watch), dbus_watch_get_flags(w->watch));
	if (w->enabled) {
	  dbus_watch_handle(w->watch, dbus_watch_get_flags(w->watch));
	}
      }
      
      kprintf("Dispatching messages\n");
      /* Dispatch messages */
      while (dbus_connection_dispatch(c->connection) == DBUS_DISPATCH_DATA_REMAINS) {
	kprintf("More messages available\n");
      }

//      dbus_connection_unref(c->connection);
    }
  }

exit:
  c->main = NULL;
  Signal(c->creator, SIGF_SINGLE);
  kprintf("MainLoop terminating\n");
}

static void receive()
{
	DBusConnection *conn = sedbus_receive(show_alert, "Test");
	// loop listening for messages being emmitted
	while (1) {
		// non blocking read of the next available message
		dbus_connection_read_write(conn, 0);
		while(dbus_connection_dispatch(conn) == DBUS_DISPATCH_DATA_REMAINS)
			;
		sleep(1);
	}
}

int dbus_update(struct dbus* dbus)
{
    DBusDispatchStatus status;

    // handle watches
    Word_t watch_count = 0;
    JLC(watch_count, dbus->watches, 0, -1);
    struct pollfd* pollfds = (struct pollfd*)alloca(sizeof(struct pollfd) * watch_count);
    int fdcount = get_pollfds(dbus, pollfds);

    if(poll(pollfds, fdcount, 0) < 0) {
        return -1;
    }

    // process the watches
    DBusWatch** pwatch;
    Word_t index = 0;
    int c = 0;
    JLF(pwatch, dbus->watches, index);
    while(pwatch != NULL) {
        struct pollfd* poll_result = &pollfds[c];
        struct DBusWatch* watch = *pwatch;

        if(dbus_watch_get_enabled(watch)) {
            assert(poll_result->fd == dbus_watch_get_unix_fd(watch));

            int flags = 0;
            int revents = poll_result->revents;

            if((revents & POLLIN) != 0) flags |= DBUS_WATCH_READABLE;
            if((revents & POLLOUT) != 0) flags |= DBUS_WATCH_WRITABLE;
            if((revents & POLLERR) != 0) flags |= DBUS_WATCH_ERROR;
            if((revents & POLLHUP) != 0) flags |= DBUS_WATCH_HANGUP;

            if(flags != 0) dbus_watch_handle(watch, flags);

            c++;
        }
        JLN(pwatch, dbus->watches, index);
    }

    // dispatch incoming messages
    while((status = dbus_connection_get_dispatch_status(dbus->conn)) != DBUS_DISPATCH_COMPLETE) {
        dbus_connection_dispatch(dbus->conn);
    }

    // Send outgoing messages
    if(dbus_connection_has_messages_to_send(dbus->conn)) {
        dbus_connection_flush(dbus->conn);
    }

    return 0;
}

static Eina_Bool efl_dispatch_dbus(void *data)
{
	DBusConnection *dbus_cnx = data;

	dbus_connection_ref(dbus_cnx);

	while (dbus_connection_dispatch(dbus_cnx) ==
					DBUS_DISPATCH_DATA_REMAINS);

	dbus_connection_unref(dbus_cnx);

	return EINA_FALSE;
}

void asdbus_handleDispatches (){
#ifdef ASDBUS_DISPATCH
	void *data;
	while ((data = extract_first_bidirelem (ASDBus.dispatches)) != NULL){
		ASDBusDispatch *d = (ASDBusDispatch*)data;
		while (dbus_connection_get_dispatch_status(d->data) == DBUS_DISPATCH_DATA_REMAINS){
			dbus_connection_dispatch(d->data);
			show_debug(__FILE__,__FUNCTION__,__LINE__,"dispatching dbus  data=%p\n", d->data);
		}
		free (d);
	}
#endif
}