/* KInterbasDB Python Package - Implementation of Events Support
**
** Version 3.1
**
** The following contributors hold Copyright (C) over their respective
** portions of code (see license.txt for details):
**
** [Original Author (maintained through version 2.0-0.3.1):]
**   1998-2001 [alex]  Alexander Kuznetsov   <alexan@users.sourceforge.net>
** [Maintainers (after version 2.0-0.3.1):]
**   2001-2002 [maz]   Marek Isalski         <kinterbasdb@maz.nu>
**   2002-2004 [dsr]   David Rushby          <woodsplitter@rocketmail.com>
** [Contributors:]
**   2001      [eac]   Evgeny A. Cherkashin  <eugeneai@icc.ru>
**   2001-2002 [janez] Janez Jere            <janez.jere@void.si>
*/

#include "_kievents.h"

#ifdef ENABLE_DB_EVENT_SUPPORT

#include "_kinterbasdb_exceptions.h"

#include "_kievents_platform_defs.h"


#define WAIT_INFINITELY 0.0

/****************** "PRIVATE" DECLARATIONS:BEGIN *******************/
/* Note this file's "public" functions are declared in _kievents.h. */

static EventConduitObject *_event_conduit_new(
    PyObject *event_names, ConnectionObject *conn
  );

void pyob_event_conduit_del(PyObject *conduit);
static void _event_conduit_delete(EventConduitObject *conduit);

static void _event_queue_delete(EventQueue *queue);

static int _event_conduit_cancel(EventConduitObject *conduit);

static long _event_queue_flush(EventQueue *queue);

static int _event_conduit_allocate_event_count_buffers (
    EventConduitObject *conduit, PyObject *event_names
  );

static PyObject *_construct_event_count_dict(
    PyObject *py_event_names, long *event_occurrence_counts
  );

static int _event_callback( EventConduitObject *conduit,
    unsigned short updated_buffer_length, const char *updated_buffer
  );

static int _event_conduit_enqueue_handler( EventConduitObject *conduit,
    boolean allowed_to_raise_exception
  );

/****************** "PRIVATE" DECLARATIONS:END *******************/


/****************** TYPE DEFINITIONS:BEGIN *******************/
PyTypeObject EventConduitType = {
  PyObject_HEAD_INIT(NULL)

  0,
  "kinterbasdb.EventConduit",
  sizeof(EventConduitObject),
  0,
  pyob_event_conduit_del,
  0,
  0,
  0,
  0,
  0,
  0,
  0,
  0,
  0
};
/****************** TYPE DEFINITIONS:END *******************/


/**************** PLATFORM-SPECIFIC EVENT FUNCTIONS:BEGIN *****************/
PyObject *abstract_wait(EventConduitObject *conduit, long timeout_millis);

/* Import the platform-specific event handling implementation. */
#ifdef MS_WIN32
  #include "_kievents_wait_windows.c"
#else
  /* All non-Win32 OSes are assumed to be UNIX-ish. */
  #include "_kievents_wait_unix.c"
#endif

/**************** PLATFORM-SPECIFIC EVENT FUNCTIONS:END *****************/


/********************** COMMON EVENT FUNCTIONS:BEGIN ***********************/

PyObject *pyob_event_conduit_new(PyObject *self, PyObject *args) {
  EventConduitObject *conduit;
  PyObject *event_names;
  ConnectionObject *conn;

  /* Begin parameter validation section. */
  if ( !PyArg_ParseTuple(args, "OO!", &event_names, &ConnectionType, &conn) ) {
    return NULL;
  }
  /* event_names has already been validated by the surrounding Python code. */

  conduit = _event_conduit_new(event_names, conn);
  /* If there was an error, the error handler in _event_conduit_new will have
  ** already freed the conduit and its subordinate memory, and
  ** _event_conduit_new will return NULL. */

  /* The event conduit object has now been created, but it will not be
  ** "hooked into the event system" until its wait() method is called. */
  return (PyObject *) conduit;
} /* pyob_event_conduit_new */


PyObject *pyob_event_conduit_wait(PyObject *self, PyObject *args) {
  EventConduitObject *conduit;
  PyObject *wait_result;
  float timeout_secs = WAIT_INFINITELY;
  long timeout_millis = (long) WAIT_INFINITELY;

  if ( !PyArg_ParseTuple(args, "O!f", &EventConduitType, &conduit, &timeout_secs) ) {
    return NULL;
  }

  if (timeout_secs < 0.0) {
    raise_exception(ProgrammingError, "Negative timeout is not valid.");
    return NULL;
  }

  if (timeout_secs != WAIT_INFINITELY) {
    timeout_millis = (long) (timeout_secs * 1000);
  }

  wait_result = abstract_wait(conduit, timeout_millis);

  return wait_result;
} /* pyob_event_conduit_wait */


PyObject *pyob_event_conduit_flush_queue(PyObject *self, PyObject *args) {
  /* Returns an integer containing the number of event notification objects
  ** that were flushed from the queue. */
  EventConduitObject *conduit;
  EventQueue *queue;
  PyObject *nodes_flushed_count;

  if ( !PyArg_ParseTuple(args, "O!", &EventConduitType, &conduit) ) {
    return NULL;
  }

  /* In the future, each queue will probably have its own thread lock, but
  ** that's not necessary right now because the DB thread lock serializes
  ** access to queues. */
  ENTER_DB_WITHOUT_LEAVING_PYTHON

  queue = conduit->queue;
  if (queue == NULL) {
    nodes_flushed_count = PyInt_FromLong(0);
  } else {
    int flush_result = _event_queue_flush(queue);
    if (flush_result < 0) {
      LEAVE_DB_WITHOUT_ENTERING_PYTHON
      return NULL;
    }
    nodes_flushed_count = PyInt_FromLong(flush_result);
    if (nodes_flushed_count == NULL) {
      LEAVE_DB_WITHOUT_ENTERING_PYTHON
      return PyErr_NoMemory();
    }
  }

  LEAVE_DB_WITHOUT_ENTERING_PYTHON

  return nodes_flushed_count;
} /* pyob_event_conduit_flush_queue */


PyObject *pyob_event_conduit_cancel(PyObject *self, PyObject *args) {
  EventConduitObject *conduit;

  if ( !PyArg_ParseTuple(args, "O!", &EventConduitType, &conduit) ) {
    return NULL;
  }

  if (_event_conduit_cancel(conduit) != 0) {
    return NULL;
  }

  RETURN_PY_NONE;
} /* pyob_event_conduit_cancel */

/********************** COMMON EVENT FUNCTIONS:BEGIN ***********************/


/******************** UTILITY FUNCTIONS:BEGIN ********************/

static EventConduitObject *_event_conduit_new(PyObject *event_names, ConnectionObject *conn) {
  EventConduitObject *conduit;

  conduit = PyObject_New(EventConduitObject, &EventConduitType);
  if (conduit == NULL) {
    return (EventConduitObject *) PyErr_NoMemory();
  }
  /* Nullify all fields of the conduit struct out so that in case an error
  ** arises before this function is finished initializing the conduit,
  ** _event_conduit_delete can release only what's appropriate without running
  ** afoul of uninitialized memory: */
  conduit->queue = NULL;
  conduit->event_id = NULL;
  conduit->event_buffer = NULL;
  conduit->result_buffer = NULL;
  conduit->buffer_length = -1;
  conduit->py_event_names = NULL;
  conduit->status = none;
  conduit->connection = NULL;


 /* Create the event queue associated with this conduit. */
  conduit->queue = kimem_main_malloc(sizeof(EventQueue));
  if (conduit->queue == NULL) {
    PyErr_NoMemory();
    goto EVENT_CONDUIT_NEW_ERROR;
  }

  conduit->queue->event = platform_create_event_object();
  if (conduit->queue->event == NULL) {
    raise_exception(OperationalError, "Unable to create native event object.");
    goto EVENT_CONDUIT_NEW_ERROR;
  }

  /* Set the head of the linked list to NULL to indicate that the list is empty. */
  conduit->queue->head = NULL;
 /* Done creating event queue. */

  /* The _event_conduit_allocate_event_count_buffers function call allocates
  ** conduit->event_buffer, conduit->result_buffer, and conduit->buffer_length: */
  if (_event_conduit_allocate_event_count_buffers(conduit, event_names) != 0) {
    goto EVENT_CONDUIT_NEW_ERROR;
  }

  conduit->event_id = kimem_main_malloc(sizeof(ISC_LONG *));
  if (conduit->event_id == NULL) {
    PyErr_NoMemory();
    goto EVENT_CONDUIT_NEW_ERROR;
  }
  *conduit->event_id = -1;

  conduit->status = none;

  Py_INCREF(event_names);
  conduit->py_event_names = event_names;

  Py_INCREF(conn);
  conduit->connection = conn;

  return conduit;

EVENT_CONDUIT_NEW_ERROR:
  Py_DECREF((PyObject *) conduit); /* 2005.05.20: Don't call pyob_event_conduit_del explicitly. */
  return NULL;
} /* _event_conduit_new */


void pyob_event_conduit_del(PyObject *conduit) {
  EventConduitObject *conduit_cast = (EventConduitObject *) conduit;

  _event_conduit_cancel(conduit_cast);

  /* Free the memory held by the member pointers of the structure, then free
  ** the structure itself. */
  _event_conduit_delete(conduit_cast);
  PyObject_Del(conduit);
} /* pyob_event_conduit_del */


static void _event_conduit_delete(EventConduitObject *conduit) {
  if (conduit->queue != NULL) {
    _event_queue_delete(conduit->queue);
    kimem_main_free(conduit->queue);
    conduit->queue = NULL;
  }

  /* The event buffer and the result buffer are allocated by the DB client
  ** library (in the isc_event_block function, specifically).  They *must* be
  ** freed by the database client library's own memory handler. */
  ENTER_DB
  if (conduit->event_buffer != NULL) {
    kimem_db_client_free(conduit->event_buffer);
    conduit->event_buffer = NULL;
  }
  if (conduit->result_buffer != NULL) {
    kimem_db_client_free(conduit->result_buffer);
    conduit->result_buffer = NULL;
  }
  LEAVE_DB

  /* Note that in a typical garbage collection of an EventConduit object,
  ** _event_conduit_cancel is called by pyob_event_conduit_del before this
  ** function is called.  This function should concern itself *solely* with
  ** memory deallocation, not with the cancellation of event registrations
  ** or other high-level issues. */
  if (conduit->event_id != NULL) {
    kimem_main_free(conduit->event_id);
    conduit->event_id = NULL;
  }

  Py_XDECREF(conduit->py_event_names);
  conduit->py_event_names = NULL;

  Py_XDECREF(conduit->connection);
  conduit->connection = NULL;
} /* _event_conduit_delete */


static void _event_queue_delete(EventQueue *queue) {
  /* Free any event notification objects in the queue (this must be done before
  ** the queue's event object is destroyed, because _event_queue_flush uses
  ** that event object). */
  _event_queue_flush(queue);

  /* Free the queue's event object. */
  platform_free_event_object(queue->event);
  queue->event = NULL;
} /* _event_conduit_delete */


static long _event_queue_flush(EventQueue *queue) {
  /* Walk through the queue (a linked list) and delete all of its nodes. */
  EventQueueItem *iter_free, *iter_next;
  long nodes_freed_count = 0;

  iter_free = queue->head;
  while (iter_free != NULL) {
    nodes_freed_count++;

    iter_next = iter_free->next;
    /* Plain free, rather than kimem_main_free, is appropriate here because
    ** we don't hold the GIL. */
    kimem_plain_free(iter_free);
    iter_free = iter_next;
  }
  queue->head = NULL;

  /* Clear the queue's event object so that the next wait() call will work
  ** properly. */
  {
    int unsignal_result = event_queue_unsignal(queue);
    if (unsignal_result < 0) {
      raise_exception(OperationalError, "Could not clear native event object.");
      return unsignal_result;
    }
  }

  return nodes_freed_count;
} /* _event_queue_flush */


static int _event_conduit_cancel(EventConduitObject *conduit) {
  ConnectionObject *conn = conduit->connection;
  assert (conn != NULL);

  /* If there is no event id or its value is set to an "inactive" flag, we need
  ** not do anything. */
  if (conduit->event_id != NULL && *conduit->event_id != -1) {
    ISC_LONG *local_event_id;

    /* Set conduit->event_id to NULL in order to indicate to the event callback
    ** thread that the conduit is no longer viable.  Retain a local pointer to
    ** the event id's storage so that it can be freed after the call to
    ** isc_cancel_events below. */
    ENTER_DB
    local_event_id = conduit->event_id;
    conduit->event_id = NULL;
    LEAVE_DB

    /* The DB lock must not be held when we call isc_cancel_events, because
    ** isc_cancel_events causes the event handler thread to do a dummy run of
    ** the event callback function.  The entire event callback function is
    ** synchronized on the (non-reentrant) DB lock, so the event handler thread
    ** deadlocks instead of shutting down if the thread that called
    ** isc_cancel_events is still holding the DB lock. */
    isc_cancel_events(conn->status_vector, &conn->db_handle, local_event_id);

    /* conduit->event_id should already have been nulled at the beginning of
    ** this function; instead, the local variable local_event_id now points to
    ** the storage, which must be freed now. */
    assert(conduit->event_id == NULL);
    kimem_main_free(local_event_id);

    if (DB_API_ERROR(conn->status_vector)) {
      raise_sql_exception( OperationalError,
          "Could not cancel event registration: ", conn->status_vector
        );
      return -1;
    }
  }

  return 0;
} /* _event_conduit_cancel */


static int _event_conduit_allocate_event_count_buffers (
    EventConduitObject *conduit, PyObject *event_names
  )
{
  /* Build event count buffer using isc_event_block, for up to
  ** MAX_EVENT_NAMES events.  I know of no way to gracefully and portably
  ** "apply(C_function, Py_Sequence)", so this is really ugly. */
  int i;
  PyObject *en[MAX_EVENT_NAMES];
  /* The following is cast to short is not risky because the length of the
  ** sequence has been constrained by previous code. */
  short event_names_count = (short) PySequence_Length(event_names);
  if (event_names_count == -1) {
    return -1;
  }

  for (i = 0; i < event_names_count; i++) {
    /* PySequence_GetItem creates new references; they are released in a loop
    ** below. */
    en[i] = PySequence_GetItem(event_names, i);
  }

  ENTER_DB_WITHOUT_LEAVING_PYTHON

  #define ISC_EVENT_BLOCK_BEGIN conduit->buffer_length = \
    (short)isc_event_block( \
      &conduit->event_buffer, &conduit->result_buffer, event_names_count,
  #define ISC_EVENT_BLOCK_END ); break;
  #define EN_STR(i) PyString_AsString(en[i])

  switch (event_names_count) {
    case 1:
      ISC_EVENT_BLOCK_BEGIN
        EN_STR(0)
      ISC_EVENT_BLOCK_END
    case 2:
      ISC_EVENT_BLOCK_BEGIN
        EN_STR(0), EN_STR(1)
      ISC_EVENT_BLOCK_END
    case 3:
      ISC_EVENT_BLOCK_BEGIN
        EN_STR(0), EN_STR(1), EN_STR(2)
      ISC_EVENT_BLOCK_END
    case 4:
      ISC_EVENT_BLOCK_BEGIN
        EN_STR(0), EN_STR(1), EN_STR(2), EN_STR(3)
      ISC_EVENT_BLOCK_END
    case 5:
      ISC_EVENT_BLOCK_BEGIN
        EN_STR(0), EN_STR(1), EN_STR(2), EN_STR(3), EN_STR(4)
      ISC_EVENT_BLOCK_END
    case 6:
      ISC_EVENT_BLOCK_BEGIN
        EN_STR(0), EN_STR(1), EN_STR(2), EN_STR(3), EN_STR(4),
        EN_STR(5)
      ISC_EVENT_BLOCK_END
    case 7:
      ISC_EVENT_BLOCK_BEGIN
        EN_STR(0), EN_STR(1), EN_STR(2), EN_STR(3), EN_STR(4),
        EN_STR(5), EN_STR(6)
      ISC_EVENT_BLOCK_END
    case 8:
      ISC_EVENT_BLOCK_BEGIN
        EN_STR(0), EN_STR(1), EN_STR(2), EN_STR(3), EN_STR(4),
        EN_STR(5), EN_STR(6), EN_STR(7)
      ISC_EVENT_BLOCK_END
    case 9:
      ISC_EVENT_BLOCK_BEGIN
        EN_STR(0), EN_STR(1), EN_STR(2), EN_STR(3), EN_STR(4),
        EN_STR(5), EN_STR(6), EN_STR(7), EN_STR(8)
      ISC_EVENT_BLOCK_END
    case 10:
      ISC_EVENT_BLOCK_BEGIN
        EN_STR(0), EN_STR(1), EN_STR(2), EN_STR(3), EN_STR(4),
        EN_STR(5), EN_STR(6), EN_STR(7), EN_STR(8), EN_STR(9)
      ISC_EVENT_BLOCK_END
    case 11:
      ISC_EVENT_BLOCK_BEGIN
        EN_STR(0), EN_STR(1), EN_STR(2), EN_STR(3), EN_STR(4),
        EN_STR(5), EN_STR(6), EN_STR(7), EN_STR(8), EN_STR(9),
        EN_STR(10)
      ISC_EVENT_BLOCK_END
    case 12:
      ISC_EVENT_BLOCK_BEGIN
        EN_STR(0), EN_STR(1), EN_STR(2), EN_STR(3), EN_STR(4),
        EN_STR(5), EN_STR(6), EN_STR(7), EN_STR(8), EN_STR(9),
        EN_STR(10), EN_STR(11)
      ISC_EVENT_BLOCK_END
    case 13:
      ISC_EVENT_BLOCK_BEGIN
        EN_STR(0), EN_STR(1), EN_STR(2), EN_STR(3), EN_STR(4),
        EN_STR(5), EN_STR(6), EN_STR(7), EN_STR(8), EN_STR(9),
        EN_STR(10), EN_STR(11), EN_STR(12)
      ISC_EVENT_BLOCK_END
    case 14:
      ISC_EVENT_BLOCK_BEGIN
        EN_STR(0), EN_STR(1), EN_STR(2), EN_STR(3), EN_STR(4),
        EN_STR(5), EN_STR(6), EN_STR(7), EN_STR(8), EN_STR(9),
        EN_STR(10), EN_STR(11), EN_STR(12), EN_STR(13)
      ISC_EVENT_BLOCK_END
    case 15:
      ISC_EVENT_BLOCK_BEGIN
        EN_STR(0), EN_STR(1), EN_STR(2), EN_STR(3), EN_STR(4),
        EN_STR(5), EN_STR(6), EN_STR(7), EN_STR(8), EN_STR(9),
        EN_STR(10), EN_STR(11), EN_STR(12), EN_STR(13), EN_STR(14)
      ISC_EVENT_BLOCK_END

    /* No default case is necessary because the length of event_names was
    ** already validated. */
  }

  LEAVE_DB_WITHOUT_ENTERING_PYTHON

  /* PySequence_GetItem returned a new reference; we must release it. */
  for (i = 0; i < event_names_count; i++) {
    Py_DECREF(en[i]);
  }

  if (conduit->buffer_length <= 0) {
    PyErr_NoMemory();
    return -1;
  }

  return 0;
} /* _event_conduit_allocate_event_count_buffers */


static PyObject *_construct_event_count_dict(
    PyObject *py_event_names, long *event_occurrence_counts
  )
{
  /* Construct a dictionary that maps
  **   (event name -> number of times that specific event occurred) */
  PyObject *countDict = NULL;
  short event_names_count;
  short i;

  event_names_count = PySequence_Length(py_event_names);
  if (event_names_count == -1) {
    goto fail;
  }

  countDict = PyDict_New();
  if (countDict == NULL) {
    goto fail;
  }

  for (i = 0; i < event_names_count; i++) {
    /* Both PySequence_GetItem and PyInt_FromLong create new references; we
    ** must DECREF them after passing them to PyDict_SetItem, which creates
    ** its new references of its own. */
    PyObject *key, *value;

    key = PySequence_GetItem(py_event_names, i);
    if (key == NULL) {
      goto fail;
    }

    value = PyInt_FromLong(event_occurrence_counts[i]);
    if (value == NULL) {
      Py_DECREF(key);
      goto fail;
    }

    {
      int dictStoreStatus = PyDict_SetItem(countDict, key, value);
      Py_DECREF(key);
      Py_DECREF(value);
      if (dictStoreStatus == -1) {
        goto fail;
      }
    }
  }

  return countDict;

fail:
  if (!PyErr_Occurred()) {
    PyErr_NoMemory();
  }

  Py_XDECREF(countDict);

  return NULL;
} /* _construct_event_count_dict */


/******************** UTILITY FUNCTIONS:END ********************/


static int _event_callback( EventConduitObject *conduit,
    unsigned short updated_buffer_length, const char *updated_buffer
  )
{
/* Notice that this callback NEVER manipulates the Python GIL in any way, so
** it is not safe to call the Python API within it.  However, the entire
** callback is protected by the DB thread lock. */
ENTER_DB_WITHOUT_LEAVING_PYTHON
{ /* begin block that contains the actual callback functionality */

  /* A vector into which event counts will be stored by isc_event_counts: */
  ISC_ULONG event_count_vector[STATUS_VECTOR_SIZE];

  boolean is_dummy_run;
  int signal_result = 0;

  /* When event registration is cancelled via isc_cancel_events, the handler
  ** thread (this thread) is called so that it can clean up after itself.
  ** This event handler doesn't need to clean up after itself; it can just
  ** return (after having released the database lock, of course).  Since this
  ** thread does not requeue itself in this case, it will "die". */
  if (conduit == NULL || conduit->event_id == NULL) {
    goto EVENT_CALLBACK_EXIT;
  }
  assert (conduit->queue != NULL);
  assert (conduit->event_buffer != NULL);
  assert (conduit->result_buffer != NULL);
  assert (conduit->py_event_names != NULL);
  assert (conduit->connection != NULL);

  is_dummy_run = (conduit->status == after_initial_enqueue);

  /* Update the event count buffer to record the fact that this occurrence
  ** (or pseudo-occurrence (i.e., the initial 'dummy run')) of the event was
  ** handled. */
  memcpy(conduit->result_buffer, updated_buffer, updated_buffer_length);

  /* DB threadlock is already held, so no problem in calling isc_* function. */
  isc_event_counts(
      event_count_vector,
      conduit->buffer_length,
      conduit->event_buffer,
      conduit->result_buffer
    );

  if (!is_dummy_run) {
    /* This run is the real thing; insert an object into the event queue. */
    int i;
    EventQueue *queue = conduit->queue;

    /* Create new queue item. */
    /* Since we don't hold the Python GIL, we must use system malloc rather
    ** than kimem_main_malloc.  The thread that retrieves the queue item will
    ** be responsible for freeing it (with kimem_plain_free). */
    EventQueueItem *new_item = kimem_plain_malloc(sizeof(EventQueueItem));
    new_item->next = NULL;

    for (i = 0; i < MAX_EVENT_NAMES; i++) {
      new_item->counts[i] = (long) event_count_vector[i];
    }

    /* Insert the new item into the queue. */
    if (queue->head == NULL) {
      queue->head = new_item;
    } else {
      /* Find the tail of the queue and append new_item there. */
      EventQueueItem *iter = queue->head;

      while (iter->next != NULL) {
        iter = iter->next;
      }
      iter->next = new_item;
    }

    /* Tell the waiting thread that there's a new queue item. */
    signal_result = event_queue_signal(queue);
  }

  /* Re-queue the event handler so that the next occurrence of the event will
  ** trigger the handler again, rather than being ignored.
  ** If signalling the event queue raised an error, however, do not re-queue
  ** the event handler (effectively killing it). */
  if (signal_result >= 0) {
    if (_event_conduit_enqueue_handler(conduit, FALSE) == 0) {
      if (is_dummy_run) {
        conduit->status = dummy_run_complete;
      }
    }
  }

} /* end block that contains the actual callback functionality */
EVENT_CALLBACK_EXIT:
LEAVE_DB_WITHOUT_ENTERING_PYTHON /* Because we were never in Python. */
  return 0;
} /* _event_callback */


static int _event_conduit_enqueue_handler( EventConduitObject *conduit,
    boolean allowed_to_raise_exception
  )
{
  /* If $allowed_to_raise_exception is FALSE, this function will refrain from
  ** raising a Python exception, and will indicate its status solely via the
  ** return code. */

  /* WARNING:  This function places responsibility for the DB thread lock
  ** on the caller!  The caller must have acquired the DB lock before calling
  ** this function.  */

  int enqueue_result = isc_que_events(
      conduit->connection->status_vector,
      &conduit->connection->db_handle,
      conduit->event_id,
      conduit->buffer_length,
      conduit->event_buffer,
      (EVENT_CALLBACK_FUNCTION) _event_callback,
      conduit
    );

  if (allowed_to_raise_exception) {
    if (DB_API_ERROR(conduit->connection->status_vector)) {
      LEAVE_DB_WITHOUT_ENTERING_PYTHON
      raise_sql_exception( OperationalError,
          "Could not queue event handler: ", conduit->connection->status_vector
        );
      ENTER_DB_WITHOUT_LEAVING_PYTHON
    }
  }

  return enqueue_result;
} /* _event_conduit_enqueue_handler */


PyObject *abstract_wait(EventConduitObject *conduit, long timeout_millis) {
  EventQueue *queue = conduit->queue;
  EventQueueItem *q_item = NULL;
  PyObject *py_result = NULL;
  int wait_result = 0;

  ENTER_DB_WITHOUT_LEAVING_PYTHON

  /* If this conduit has never been wait()ed on before, we must kick off the
  ** event listening process. */
  if (conduit->status == none) {
    int enqueue_result = _event_conduit_enqueue_handler(conduit, TRUE);
    if (enqueue_result != 0) {
      /* _event_conduit_enqueue_handler already set Python exception. */
      goto PLATFORM_WAIT_ERROR;
    }
    conduit->status = after_initial_enqueue;
  }

  /* If there's NOT already an event waiting in the queue, wait for one.
  ** Otherwise, retrieve one from the queue and return without waiting. */
  if (queue->head == NULL) {
    LEAVE_DB_WITHOUT_ENTERING_PYTHON

    /* Release the Python GIL and wait for notification of the arrival of an
    ** item in the event queue. */
    LEAVE_PYTHON_WITHOUT_ENTERING_DB
    wait_result = event_queue_wait(queue, timeout_millis); /* We hold NO thread locks at this point. */
    ENTER_PYTHON_WITHOUT_LEAVING_DB

    ENTER_DB_WITHOUT_LEAVING_PYTHON
  }

  if (wait_result == EVENT_ERROR) {
    raise_exception(OperationalError, "Native event-wait encountered error.");
    goto PLATFORM_WAIT_ERROR;
  }

  if (wait_result == EVENT_TIMEOUT) {
    py_result = Py_None;
    Py_INCREF(Py_None);
  } else {
    /* Clear the event object no matter whether we had to wait() on it or not: */
    if ( event_queue_unsignal(queue) < 0 ) {
      raise_exception(OperationalError, "Could not unsignal native event object.");
      goto PLATFORM_WAIT_ERROR;
    }

    /* Retrieve the head of the event queue. */
    assert (queue->head != NULL);
    q_item = queue->head;
    queue->head = queue->head->next;
    /* Physically enfore the logical detachment of q_item from the queue: */
    q_item->next = NULL;

    /* Convert the raw q_item to a Python-accessible value. */
    py_result = _construct_event_count_dict(
        conduit->py_event_names, q_item->counts
      );

    /* The queue item must be garbage collected because it's no longer part
    ** of the queue; it will become unreachable when this function returns. */
    /* Plain free, rather than kimem_main_free, is appropriate here because
    ** the queue item was allocated by _event_callback, which didn't hold the
    ** GIL at the time of the allocation, and thus couldn't use kimem_main_malloc. */
    kimem_plain_free(q_item);
  }

  LEAVE_DB_WITHOUT_ENTERING_PYTHON /* We're already in Python. */
  return py_result;

PLATFORM_WAIT_ERROR:
  LEAVE_DB_WITHOUT_ENTERING_PYTHON /* We're already in Python. */
  return NULL;
} /* abstract_wait */


#endif /* ENABLE_DB_EVENT_SUPPORT */