/* KInterbasDB Python Package - Implementation of Events Support
 *
 * Version 3.3
 *
 * 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-2007 [dsr]   David Rushby          <woodsplitter@rocketmail.com>
 * [Contributors:]
 *   2001      [eac]   Evgeny A. Cherkashin  <eugeneai@icc.ru>
 *   2001-2002 [janez] Janez Jere            <janez.jere@void.si>            */

#ifdef ENABLE_DB_EVENT_SUPPORT

#include "_kisupport_threadsafe_fifo_queue.c"

#include "_kievents.h"
#include "_kievents_infra.c"

static PyObject *pyob_EventConduit_close(EventConduit *);

static int _update_event_count_dict(PyObject *, PyObject *,
    Py_ssize_t, Py_ssize_t, long *
  );
static long _event_context_allocate_event_count_buffers(
    PyObject *, Py_ssize_t, Py_ssize_t, EventRequestBlock *
  );

/* Global variables that are "private" to the event subsystem: */
PyObject *events__PyInt_zero;

#define EN_OFFSET_FROM_BLOCK_NUMBER(bn) (EVENT_BLOCK_SIZE * (bn))
#define EN_UPPER_LIMIT_FROM_BLOCK_NUMBER(bn, n_event_names) ( \
    (EN_OFFSET_FROM_BLOCK_NUMBER(bn) + EVENT_BLOCK_SIZE) > (n_event_names) \
    ? n_event_names : (EN_OFFSET_FROM_BLOCK_NUMBER(bn) + EVENT_BLOCK_SIZE) \
  )

/************* EventConduit METHODS INACCESSIBLE TO PYTHON:BEGIN *************/
#define EventConduit_is_closed(self) ((self)->state != CONDUIT_STATE_OPEN)

static long _EventConduit_require_open(
    EventConduit *self, char *failure_message
  )
{
  /* If self is not an open event conduit, raises the supplied error message
   * (or a default if no error message is supplied).
   * Returns 0 if the cursor was open; -1 if it was closed. */
  if (!EventConduit_is_closed(self)) { return 0; }

  if (failure_message == NULL) {
    failure_message = "Invalid EventConduit state.  The conduit must be OPEN"
      " to perform this operation.";
  }
  raise_exception(ConduitWasClosed, failure_message);
  return -1;
} /* _EventConduit_require_open */

#define CONDUIT_REQUIRE_OPEN_WITH_FAILURE(self, failure_action) \
  if (_EventConduit_require_open(self, NULL) != 0) { failure_action; }

#define CONDUIT_REQUIRE_OPEN(self) \
  CONDUIT_REQUIRE_OPEN_WITH_FAILURE(self, return NULL)
/************** EventConduit METHODS INACCESSIBLE TO PYTHON:END **************/

/************** EventConduit METHODS ACCESSIBLE TO PYTHON:BEGIN **************/
static PyObject *pyob_EventConduit_create(PyObject *self_, PyObject *py_args) {
  EventConduit *self = NULL;
  PyObject *py_event_names = NULL;
  int py_event_names_length;
  int i;

  CConnection *originating_con;
  PyObject *py_event_names_orig;
  PyObject *py_con_params;

  boolean init_event_q = FALSE;
  boolean init_op_thread_context = FALSE;
  boolean started_op_thread = FALSE;

  if (!PyArg_ParseTuple(py_args, "O!O!O",
        &ConnectionType, &originating_con,
        &PyTuple_Type, &py_con_params, &py_event_names_orig
     ))
  { return NULL; }

  CON_ACTIVATE(originating_con, return NULL);

 /* Validate py_event_names_orig, storing a tuple representation of it in
  * py_event_names: */
  /* Execute the equivalent of the following Python statement:
   *   py_event_names = tuple(py_event_names_orig)
   * This allows the client programmer to pass any iterable; the tuple
   * constructor takes care of extracting the values.  Importantly, it also
   * prevents the client programmer from passing a mutable sequence object and
   * then modifying it while the EventConduit is operating.
   *
   * We use separate variables for py_event_names and py_event_names_orig so
   * this function's error handler can safely Py_XDECREF(py_event_names). */
  py_event_names = PyObject_CallFunctionObjArgs( (PyObject *) &PyTuple_Type,
      py_event_names_orig, NULL
    );
  if (py_event_names == NULL) { goto fail; }

  { /* The database engine doesn't work properly with anywhere near INT_MAX
     * events, so the limit imposed here is no real loss. */
    const Py_ssize_t py_event_names_length_ss = PyTuple_GET_SIZE(
        py_event_names
      );
    if (py_event_names_length_ss > INT_MAX) {
      raise_exception(NotSupportedError, "At most INT_MAX events supported.");
      goto fail;
    }
    py_event_names_length = (int) py_event_names_length_ss;
  }

  /* Disallow zero event names: */
  if (py_event_names_length == 0) {
    raise_exception(ProgrammingError, "Can't wait for zero events.");
    goto fail;
  }

  /* Disallow non-string event names, and also multiple events with the same
   * name: */
  {
    PyObject *py_event_name_dict = PyDict_New();
    if (py_event_name_dict == NULL) { goto fail; }

    for (i = 0; i < py_event_names_length; i++) {
      PyObject *en = PyTuple_GET_ITEM(py_event_names, i);
      if (!PyString_CheckExact(en)) {
        raise_exception(ProgrammingError, "All event names must be str"
            " objects."
          );
        goto done_checking_event_names;
      }
      if (PyDict_GetItem(py_event_name_dict, en) != NULL) {
        PyObject *py_err_msg = PyString_FromFormat("The following event name"
            " appears more than once in the supplied event_names sequence:"
            " \"%s\"", PyString_AS_STRING(en)
          );
        if (py_err_msg != NULL) {
          raise_exception(ProgrammingError, PyString_AS_STRING(py_err_msg));
          Py_DECREF(py_err_msg);
        }
        goto done_checking_event_names;
      }
      /* We're using py_event_name_dict as a set, so the key matters, but the
       * value does not. */
      if (PyDict_SetItem(py_event_name_dict, en, Py_None) != 0) {
        goto done_checking_event_names;
      }
    }
    /* Fall through to done_checking_event_names: */

    done_checking_event_names:
      Py_DECREF(py_event_name_dict);
      if (PyErr_Occurred()) { goto fail; }
  }

 /* Validate py_con_params.
  * kinterbasdb itself, rather than the client programmer, supplies
  * py_con_params, so we're not very uptight about validating it. */

  /* The PyArg_ParseTuple call toward the beginning of this function should've
   * already ensured the following: */
  assert (PyTuple_CheckExact(py_con_params));

  if (    PyTuple_GET_SIZE(py_con_params) != 3
      || !PyString_CheckExact(PyTuple_GET_ITEM(py_con_params, 0))
      || !PyString_CheckExact(PyTuple_GET_ITEM(py_con_params, 1))
      || !PyInt_CheckExact   (PyTuple_GET_ITEM(py_con_params, 2))
     )
  {
    raise_exception(InternalError, "py_con_params is invalid.");
    goto fail;
  }

 /* The parameters passed to this method from the Python level are now
  * validated; created and initialize the EventConduit object. */
  self = PyObject_New(EventConduit, &EventConduitType);
  if (self == NULL) { goto fail; }

  /* Nullify self's members for safe cleanup before initializing them: */
  self->state = CONDUIT_STATE_CREATED;
  self->py_event_names = NULL;
  self->n_event_names = -1;
  self->n_event_blocks = -1;
  self->py_event_counts_dict_template = NULL;
  self->op_thread_ref = THREAD_REF_INVALID;
  /* Done nullifying. */

  /* Pass ownership of py_event_names to self. */
  assert (PyTuple_CheckExact(py_event_names));
  self->py_event_names = py_event_names;
  py_event_names = NULL;

  self->n_event_names = py_event_names_length;
  self->n_event_blocks = py_event_names_length / EVENT_BLOCK_SIZE;
  if (py_event_names_length % EVENT_BLOCK_SIZE != 0) {
    ++self->n_event_blocks;
  }

  self->py_event_counts_dict_template = PyDict_New();
  if (self->py_event_counts_dict_template == NULL) { goto fail; }
  {
    PyObject *py_ecdt = self->py_event_counts_dict_template;
    PyObject *py_en = self->py_event_names;
    for (i = 0; i < py_event_names_length; i++) {
      if (PyDict_SetItem(py_ecdt,
            PyTuple_GET_ITEM(py_en, i), events__PyInt_zero
          ) != 0
         )
      { goto fail; }
    }
  }

  if (ThreadSafeFIFOQueue_init(&self->event_q) != 0) {
    raise_exception(OperationalError, "Unable to initialize event_q.");
    goto fail;
  }
  init_event_q = TRUE;

  {
    EventOpThreadContext *eotc = &self->op_thread_context;

    /* Initialize the context structure for the EventOpThread: */
    if (EventOpThreadContext_init(eotc, &self->event_q, self->n_event_blocks)
        != 0
       )
    {
      raise_exception(OperationalError,
          "Unable to initialize op_thread_context."
        );
      goto fail;
    }
    init_op_thread_context = TRUE;

    assert (self->n_event_blocks == eotc->n_event_blocks);
    assert (&self->event_q == eotc->event_q);

    /* Start the EventOpThread: */
    self->op_thread_ref = Thread_create(EventOpThread_main, eotc,
        DV_THREADID_PTR(&eotc->event_op_thread_id)
      );
    if (self->op_thread_ref == THREAD_REF_INVALID) {
      raise_exception(OperationalError, "Unable to create EventOpThread.");
      goto fail;
    }
    started_op_thread = TRUE;

    /* Send an opcode to the EventOpThread requesting that it establish a
     * private database connection.  This thread then blocks until the
     * EventOpThread indicates completion by posting to the admin_response_q.
     */
    {
      ConnParamsNode* payload = kimem_plain_malloc(sizeof(ConnParamsNode));
      if (payload == NULL) { goto fail; }
      payload->dsn = NULL;
      payload->dpb = NULL;

      /* The types of the elements of py_con_params were validated earlier.
       * Additionally, when the CConnection they're derived from was
       * originally created (in _kicore_connection.c/pyob_Connection_connect), the
       * contents of the elements were validated.  Therefore, we can be certain
       * that py_con_params contains appropriate values. */
      {
        PyObject *param;

        #define _CONVERT_STR_PARAM(tuple_pos, name) \
          param = PyTuple_GET_ITEM(py_con_params, tuple_pos); \
          payload->name ## _len = (short) PyString_GET_SIZE(param); \
          assert (payload->name ## _len > 0); \
          /* Note that there's no need for a trailing null byte at the end of \
           * the buffer: */ \
          payload->name = kimem_plain_malloc(payload->name ## _len); \
          if (payload->name == NULL) { goto fail_freeing_payload; } \
          memcpy(DV_STR(payload->name), PyString_AS_STRING(param), \
              payload->name ## _len \
            );

        _CONVERT_STR_PARAM(0, dsn);
        _CONVERT_STR_PARAM(1, dpb);

        param = PyTuple_GET_ITEM(py_con_params, 2);
        assert (PyInt_CheckExact(param));
        payload->dialect = (short) PyInt_AS_LONG(param);
      }

      {
        int req_status = -1;
        int res_status = -1;
        ISC_STATUS sql_error_code = 0;
        char *message = NULL;

        LEAVE_GIL_WITHOUT_AFFECTING_DB /* DB lock is immaterial. */
        req_status = EventOpQueue_request(&eotc->op_q, OP_CONNECT, NO_TAG,
            payload
          );

        if (req_status == 0) {
          res_status = AdminResponseQueue_require(&eotc->admin_response_q,
              OP_CONNECT, 0, &sql_error_code, &message, WAIT_INFINITELY_LONG
            );
        }
        ENTER_GIL_WITHOUT_AFFECTING_DB /* DB lock is immaterial. */

        if (req_status != 0 || res_status != 0) {
          if (message != NULL) {
            PyObject *err_msg = PyString_FromFormat("While creating"
                " EventConduit, OP_CONNECT failed:\n%s", message
              );
            kimem_plain_free(message);
            if (err_msg != NULL) {
              raise_exception_with_numeric_error_code(OperationalError,
                  sql_error_code, PyString_AS_STRING(err_msg)
                );
              Py_DECREF(err_msg);
            }
          } else {
            raise_exception(OperationalError, "EventOpThread could not"
                " establish private database connection."
              );
          }

          /* If the EventOpQueue_request call succeeded, then the queue took
           * ownership of the payload. */
          if (req_status == 0) {
            goto fail;
          } else {
            goto fail_freeing_payload;
          }
        }
      }

      goto connect_request_succeeded;

      fail_freeing_payload:
        if (payload->dsn != NULL) {
          kimem_plain_free(DV_STR(payload->dsn));
        }
        if (payload->dpb != NULL) {
          kimem_plain_free(DV_STR(payload->dpb));
        }
        kimem_plain_free(payload);

        goto fail;
    }

    connect_request_succeeded:
    /* On success, the EventOpThread will have freed payload's memory. */

    for (i = 0; i < eotc->n_event_blocks; i++) {
      if (_event_context_allocate_event_count_buffers(
            DV_PYO(self->py_event_names),
            EN_OFFSET_FROM_BLOCK_NUMBER(i),
            EN_UPPER_LIMIT_FROM_BLOCK_NUMBER(i, self->n_event_names),
            DV_ERB(eotc->er_blocks + i)
          ) != 0
         )
      { goto fail; }
    }

    /* Now that the EventOpThread's private connection has been established and
     * the members of EventOpThreadContext necessary to support event
     * registration have been initialized, we can direct the EventOpThread to
     * actually register for event notification. */
    {
      int req_status = -1;
      int res_status = -1;
      ISC_STATUS sql_error_code = 0;
      char *message = NULL;

      LEAVE_GIL_WITHOUT_AFFECTING_DB /* DB lock is immaterial. */
      req_status = EventOpQueue_request(&eotc->op_q, OP_REGISTER, NO_TAG,
          NULL
        );
      if (req_status == 0) {
        res_status = AdminResponseQueue_require(&eotc->admin_response_q,
            OP_REGISTER, 0, &sql_error_code, &message, WAIT_INFINITELY_LONG
          );
      }
      ENTER_GIL_WITHOUT_AFFECTING_DB /* DB lock is immaterial. */

      if (req_status != 0 || res_status != 0) {
        if (message != NULL) {
          PyObject *err_msg = PyString_FromFormat("While creating"
              " EventConduit, OP_REGISTER failed:\n%s", message
            );
          kimem_plain_free(message);
          if (err_msg != NULL) {
            raise_exception_with_numeric_error_code(OperationalError,
                sql_error_code, PyString_AS_STRING(err_msg)
              );
            Py_DECREF(err_msg);
          }
        } else {
          raise_exception(OperationalError, "EventOpThread failed to register"
              " for event notification."
            );
        }

        goto fail;
      }
    }

    assert (EventOpThreadContext_has_state(eotc, OPTHREADSTATE_READY));
  }

  /* Success: */
  assert (self != NULL);
  assert (!PyErr_Occurred());
  self->state = CONDUIT_STATE_OPEN;

  goto clean;
  fail:
    assert (PyErr_Occurred());

    if (self != NULL) {
      if (self->py_event_names != NULL) {
        Py_DECREF(self->py_event_names);
        self->py_event_names = NULL;
      }
      if (self->py_event_counts_dict_template != NULL) {
        Py_DECREF(self->py_event_counts_dict_template);
        self->py_event_counts_dict_template = NULL;
      }

      if (init_op_thread_context) {
        EventOpThreadContext *eotc = &self->op_thread_context;

        /* Must wait until the EventOpThread has actually exited before yanking
         * support structures from under it: */
        if (started_op_thread) {
          Thread_join(self->op_thread_ref);
        }

        ENTER_GDAL
        EventOpThreadContext_free_er_blocks(eotc);
        LEAVE_GDAL

        EventOpThreadContext_close(eotc);

        #ifdef FIREBIRD_1_0_ONLY
          /* 2005.12.13:
           * YYY: KInterbasDB's event-handling system is not working with FB
           * 1.0.x at present, although it works with other versions of the
           * database engine.  KInterbasDB's attempt to establish an event
           * conduit causes an internal error in the FB 1.0.3 server, after
           * which the server becomes totally unresponsive, both to
           * isc_det4ch_db requests and even to server process shutdown
           * requests (the server has to be killed).
           * This workaround is intended to prevent the Python client process
           * from hanging permanently due to the server's refusal to respond to
           * an isc_det4ch_db request. */
          Connection_close(originating_con, TRUE, FALSE);
        #endif
      }
      if (init_event_q) {
        /* self's destructor will close self->event_q, but we ensure that it's
         * in the expected state: */
        assert (!(self->event_q).closed);
      }

      /* DECREFing self here won't cause pyob_EventConduit_close to be executed
       * because pyob_EventConduit___del__ will detect that self's state flag
       * was not yet set to CONDUIT_STATE_OPEN. */
      assert (EventConduit_is_closed(self));
      Py_DECREF(self);
      self = NULL;
    }

    /* This doesn't constitute a "double DECREF" because local variable
     * py_event_names is nullified as soon as ownership of the Python object is
     * transferred to self->py_event_names: */
    Py_XDECREF(py_event_names);

    /* Fall through to clean: */
    assert (PyErr_Occurred());
    assert (self == NULL);

  clean:
    CON_PASSIVATE(originating_con);
    CON_MUST_NOT_BE_ACTIVE(originating_con);
    return (PyObject *) self;
} /* pyob_EventConduit_create */

static void pyob_EventConduit___del__(EventConduit *self) {
  if (!EventConduit_is_closed(self)) {
    PyObject *py_close_result = pyob_EventConduit_close(self);
    if (py_close_result != NULL) {
      Py_DECREF(py_close_result);
    } else {
      SUPPRESS_EXCEPTION;
    }
  }

  /* 2005.09.27: Moved the closure of event_q here from
   * pyob_EventConduit_close. */
  if (!(self->event_q).closed) {
    if (ThreadSafeFIFOQueue_close(&self->event_q) != 0) { SUPPRESS_EXCEPTION; }
  }

  /* Release the EventConduit struct itself: */
  PyObject_Del(self);
} /* pyob_EventConduit___del__ */

static PyObject *pyob_EventConduit_closed_get(
    EventConduit *self, void *closure
  )
{
  /* Obviously, if the result is FALSE, there's no guarantee that it will still
   * be valid by the time it's returned to the caller. */
  return PyBool_FromLong(EventConduit_is_closed(self));
} /* pyob_EventConduit_closed_get */

static PyObject *pyob_EventConduit_close(EventConduit *self) {
  /* Note the explicit serialization in this method. */
  PyObject *res = NULL;
  EventOpThreadContext *eotc = &self->op_thread_context;

  CONDUIT_REQUIRE_OPEN_WITH_FAILURE(self, goto fail);

  if (self->py_event_names != NULL) {
    Py_DECREF(self->py_event_names);
    self->py_event_names = NULL;
  }

  self->n_event_names = -1;
  self->n_event_blocks = -1;

  if (self->py_event_counts_dict_template != NULL) {
    Py_DECREF(self->py_event_counts_dict_template);
    self->py_event_counts_dict_template = NULL;
  }

  if (!EventOpThreadContext_has_state(eotc, OPTHREADSTATE_DEAD)) {
    int status = -1;
    ISC_STATUS sql_error_code = 0;
    char *message = NULL;

    LEAVE_GIL_WITHOUT_AFFECTING_DB /* DB lock is immaterial. */

    status = EventOpQueue_request(&eotc->op_q, OP_DIE, NO_TAG, NULL);
    if (status == 0) {
      status = AdminResponseQueue_require(&eotc->admin_response_q,
          OP_DIE, 0, &sql_error_code, &message, WAIT_INFINITELY_LONG
        );
      /* Regardless of whether the EventOpThread shuts down normally, we need
       * to be sure it's finished before pulling the rug out from under it. */
      Thread_join(self->op_thread_ref);
    }

    ENTER_GIL_WITHOUT_AFFECTING_DB /* DB lock is immaterial. */

    if (status != 0) {
      if (message != NULL) {
        raise_exception_with_numeric_error_code(OperationalError,
            sql_error_code, message
          );
        kimem_plain_free(message);
      }
      goto fail;
    }
  }

  /* Close the members of EventOpThreadContext that're necessary for the most
   * basic communication with the thread(s) that're calling methods of the
   * supervising EventConduit: */
  if (EventOpThreadContext_close(eotc) != 0) { goto fail; }
  /* 2005.09.27: */
  /* Previously, the following call was issued here on the event queue:
   *   ThreadSafeFIFOQueue_close(&self->event_q);
   * However, that method closes its target destructively (freeing container
   * memory, synchronization primitives, etc.).  Therefore, calling it here
   * caused memory corruption if there were one or more threads wait()ing for
   * events when another thread called close().
   * Since our earlier closure of the EventOpThreadContext will already have
   * indirectly *cancelled* the event queue, we needn't do anything more to the
   * event queue here.  Instead, we wait to call ThreadSafeFIFOQueue_close
   * until EventConduit's destructor. */
  assert (ThreadSafeFIFOQueue_is_cancelled(&self->event_q));

  self->state = CONDUIT_STATE_CLOSED;

  res = Py_None;
  Py_INCREF(Py_None);
  goto exit;

  fail:
    assert (res == NULL);
    if (!PyErr_Occurred()) {
      raise_exception(OperationalError, "Unspecified error while closing.");
    }
    /* Fall through to exit: */

  exit:
    return res;
} /* pyob_EventConduit_close */

static PyObject *pyob_EventConduit_flush(EventConduit *self) {
  /* This method needs no explicit serialization because the underlying
   * ThreadSafeFIFOQueue self->event_q provides enough. */
  LONG_LONG n_items_flushed = -1;

  CONDUIT_REQUIRE_OPEN(self);

  if (ThreadSafeFIFOQueue_flush(&self->event_q, &n_items_flushed) != 0) {
    raise_exception(OperationalError, "Underlying event queue flush failed.");
    goto fail;
  }

  assert (n_items_flushed >= 0);
  return PythonIntOrLongFrom64BitValue(n_items_flushed);

  fail:
    assert (PyErr_Occurred());
    return NULL;
} /* pyob_EventConduit_flush */

static PyObject *pyob_EventConduit_wait(
    EventConduit *self, PyObject *args, PyObject *kwargs
  )
{
  /* This method needs no explicit serialization because the underlying
   * ThreadSafeFIFOQueue self->event_q provides enough. */
  static char *kwarg_list[] = {"timeout", NULL};

  PyObject *py_timeout = NULL;
  long timeout_ms = WAIT_INFINITELY_LONG;

  WaitResult wait_res;
  EventFiredNode *n = NULL;
  PyObject *py_count_dict = NULL;

  CONDUIT_REQUIRE_OPEN(self);

  if (!PyArg_ParseTupleAndKeywords(args, kwargs, "|O", kwarg_list,
         &py_timeout
       )
     )
  { goto fail; }

  if (py_timeout != NULL && py_timeout != Py_None) {
    LONG_LONG timeout_ms_LL = py_seconds_to_milliseconds(py_timeout,
        ProgrammingError,
        "'timeout' parameter to EventConduit.wait must be either None or -1.0"
        " to wait infinitely, or a non-negative number specifying the maximum"
        " number of seconds to wait before timing out.  The Python object %s"
        " is not an acceptable input value.",
        -1, LONG_MAX
      );
    if (PyErr_Occurred()) { goto fail; }
    /* py_seconds_to_milliseconds constrained the user-supplied timeout to
     * between -1 and LONG_MAX (inclusive), so the following cast is safe: */
    timeout_ms = (long) timeout_ms_LL;
  }

  LEAVE_GIL_WITHOUT_AFFECTING_DB /* DB lock is immaterial. */

  wait_res = EventFiredQueue_get(&self->event_q, timeout_ms, &n);

  ENTER_GIL_WITHOUT_AFFECTING_DB /* DB lock is immaterial. */

  if (wait_res == WR_WAIT_TIMEOUT) {
    RETURN_PY_NONE;
  } else if (wait_res != WR_WAIT_OK) {
    if (wait_res == WR_WAIT_CANCELLED) {
      raise_exception(ConduitWasClosed,
          "Event conduit was closed before wait completed."
        );
    } else {
      raise_exception(OperationalError,
          "Unspecified fatal error while waiting for events."
        );
    }
    goto fail;
  }

  assert (n != NULL);
  assert (n->block_number >= 0 && n->block_number <= self->n_event_blocks);

  /* Convert the raw q_item to a Python-accessible value. */
  py_count_dict = PyDict_Copy(self->py_event_counts_dict_template);
  if (py_count_dict == NULL) { goto fail; }
  if (_update_event_count_dict(py_count_dict, DV_PYO(self->py_event_names),
        EN_OFFSET_FROM_BLOCK_NUMBER(n->block_number),
        EN_UPPER_LIMIT_FROM_BLOCK_NUMBER(n->block_number, self->n_event_names),
        DV_LONG_PTR(n->counts)
      ) != 0
     )
  { goto fail; }

  assert (py_count_dict != NULL);
  goto exit;

  fail:
    assert (PyErr_Occurred());
    Py_XDECREF(py_count_dict);
    /* Fall through to exit: */

  exit:
    if (n != NULL) {
      EventFiredNode_del(n);
    }

    return py_count_dict;
} /* pyob_EventConduit_wait */
/*************** EventConduit METHODS ACCESSIBLE TO PYTHON:END ***************/

/************************** UTILITY FUNCTIONS:BEGIN **************************/
static long _event_context_allocate_event_count_buffers(
    PyObject *py_event_names, Py_ssize_t en_offset, Py_ssize_t en_upper_limit,
    EventRequestBlock *erb
  )
{
  /* Build event count buffers using isc_event_block, for up to
   * EVENT_BLOCK_SIZE events.
   * I know of no way to gracefully and *portably* "apply(function, sequence)"
   * in C, so this is really ugly. */
  long res = -1;
  char *en[EVENT_BLOCK_SIZE];
  long allocated_buf_len = -1;

  #ifdef FIREBIRD_2_0_OR_LATER
    ISC_UCHAR *
  #else
    char *
  #endif
    res_buf_slot__discarded = NULL;

  short n_to_allocate;
  assert (en_upper_limit - en_offset <= SHRT_MAX);
  n_to_allocate = (short) (en_upper_limit - en_offset);

  /* Before calling this function, EventConduit_create should have verified the
   * following: */
  assert (erb != NULL);
  assert (erb->req_buf == NULL);

  assert (en_offset >= 0);
  assert (en_upper_limit >= 1);
  assert (en_offset < en_upper_limit);
  assert (n_to_allocate >= 1 && n_to_allocate <= EVENT_BLOCK_SIZE);

  assert (py_event_names != NULL);
  assert (PyTuple_CheckExact(py_event_names));
  assert (en_upper_limit <= PyTuple_GET_SIZE(py_event_names));

  {
    Py_ssize_t py_en_tuple_index = en_offset;
    for (; py_en_tuple_index < en_upper_limit; py_en_tuple_index++) {
      PyObject *s = PyTuple_GET_ITEM(py_event_names, py_en_tuple_index);
      /* EventConduit_create should've already verified that every element of
       * py_event_names is a string: */
      assert (PyString_CheckExact(s));
      en[py_en_tuple_index % EVENT_BLOCK_SIZE] = PyString_AS_STRING(s);
    }
  }

  ENTER_GDAL

  #ifdef FIREBIRD_2_0_OR_LATER
    #define _CONVERT_REQBUF_POINTER(p) ((ISC_UCHAR **) DV_STR_PTR(p))
  #else
    #define _CONVERT_REQBUF_POINTER(p) DV_STR_PTR(p)
  #endif

  #define ISC_EVENT_BLOCK_BEGIN \
    allocated_buf_len = isc_event_block( \
      _CONVERT_REQBUF_POINTER(&erb->req_buf), \
      &res_buf_slot__discarded, n_to_allocate,
  #define ISC_EVENT_BLOCK_END ); break;
  #define EN_STR(i) en[i]

  switch (n_to_allocate) {
    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

    default:
      /* No default case is necessary because the length of py_event_names was
       * already validated. */
      assert (FALSE);
  }

  LEAVE_GDAL

  if (   allocated_buf_len <= 0
      || erb->req_buf == NULL
      || res_buf_slot__discarded == NULL
     )
  {
    raise_exception(OperationalError, "isc_event_block: event buffers could"
        " not be allocated."
      );
    goto fail;
  } else if (allocated_buf_len > SHRT_MAX) {
    raise_exception(OperationalError, "isc_event_block: allocated event"
        " buffers have size > SHRT_MAX."
      );
    goto fail;
  }
  erb->req_buf_len = (short) allocated_buf_len;

  res = 0;
  goto exit;

  fail:
    assert (PyErr_Occurred());
    assert (res == -1);

    if (erb->req_buf != NULL) {
      ENTER_GDAL
      kimem_db_client_free(DV_STR(erb->req_buf));
      erb->req_buf = NULL;
      LEAVE_GDAL
    }

    /* Fall through to exit: */

  exit:
    /* isc_event_block requires that it be allowed to allocate this buffer, but
     * because of the migration to a queue-centric thread communication
     * architecture, we don't actually use it. */
    if (res_buf_slot__discarded != NULL) {
      ENTER_GDAL
      kimem_db_client_free((void *) res_buf_slot__discarded);
      LEAVE_GDAL
    }

    return res;
} /* _event_context_allocate_event_count_buffers */

static int _update_event_count_dict(PyObject *py_count_dict,
    PyObject *py_event_names, Py_ssize_t en_offset, Py_ssize_t en_upper_limit,
    long *counts
  )
{
  Py_ssize_t en_pos;
  Py_ssize_t counts_pos;

  /* These conditions should've been verified earlier: */
  assert (py_count_dict != NULL);

  assert (py_event_names != NULL);
  assert (PyTuple_CheckExact(py_event_names));
  assert (PyTuple_GET_SIZE(py_event_names) > 0);
  assert (en_upper_limit <= PyTuple_GET_SIZE(py_event_names));
  assert (en_offset >= 0);
  assert (en_offset < en_upper_limit);

  for (en_pos = en_offset, counts_pos = 0;
       en_pos < en_upper_limit;
       en_pos++, counts_pos++
      )
  {
    long cur_count;
    assert (counts_pos >= 0 && counts_pos < EVENT_BLOCK_SIZE);
    cur_count = counts[counts_pos];
    if (cur_count == 0) {
      /* The count of 0 is already present in py_count_dict, and already zero;
       * we needn't do anything. */
      assert (
          PyObject_Compare(
            events__PyInt_zero,
            PyDict_GetItem(py_count_dict,
              PyTuple_GET_ITEM(py_event_names, en_pos)
            )
          ) == 0
        );
      continue;
    } else {
      int status;
      PyObject *py_key = PyTuple_GET_ITEM(py_event_names, en_pos);
      PyObject *py_value = PyInt_FromLong(cur_count);
      if (py_value == NULL) { goto fail; }
      assert (PyString_CheckExact(py_key));
      status = PyDict_SetItem(py_count_dict, py_key, py_value);
      Py_DECREF(py_value);
      if (status != 0) { goto fail; }
    }
  }

  return 0;

  fail:
    assert (PyErr_Occurred());
    return -1;
} /* _update_event_count_dict */

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

/********** EventConduit CLASS DEFINITION AND INITIALIZATION:BEGIN ***********/
static PyMethodDef EventConduit_methods[] = {
    {"wait", (PyCFunction) pyob_EventConduit_wait, METH_VARARGS|METH_KEYWORDS},
    {"flush", (PyCFunction) pyob_EventConduit_flush, METH_NOARGS},
    {"close", (PyCFunction) pyob_EventConduit_close, METH_NOARGS},
    {NULL}  /* sentinel */
  };

static PyGetSetDef EventConduit_getters_setters[] = {
    {"closed",
        (getter) pyob_EventConduit_closed_get,
        NULL,
        NULL,
      },
    {NULL}  /* sentinel */
  };

PyTypeObject EventConduitType = { /* new-style class */
    PyObject_HEAD_INIT(NULL)
    0,                                  /* ob_size */
    "kinterbasdb.EventConduit",         /* tp_name */
    sizeof(EventConduit),               /* tp_basicsize */
    0,                                  /* tp_itemsize */
    (destructor) pyob_EventConduit___del__, /* tp_dealloc */
    0,                                  /* tp_print */
    0,                                  /* tp_getattr */
    0,                                  /* tp_setattr */
    0,                                  /* tp_compare */
    0,                                  /* tp_repr */
    0,                                  /* tp_as_number */
    0,                                  /* tp_as_sequence */
    0,                                  /* tp_as_mapping */
    0,                                  /* tp_hash */
    0,                                  /* tp_call */
    0,                                  /* tp_str */
    0,                                  /* tp_getattro */
    0,                                  /* tp_setattro */
    0,                                  /* tp_as_buffer */
    0,                                  /* tp_flags */
    0,                                  /* tp_doc */
    0,		                              /* tp_traverse */
    0,		                              /* tp_clear */
    0,		                              /* tp_richcompare */
    0,		                              /* tp_weaklistoffset */
    0,		                              /* tp_iter */
    0,		                              /* tp_iternext */
    EventConduit_methods,               /* tp_methods */
    NULL,                               /* tp_members */
    EventConduit_getters_setters,       /* tp_getset */
    0,                                  /* tp_base */
    0,                                  /* tp_dict */
    0,                                  /* tp_descr_get */
    0,                                  /* tp_descr_set */
    0,                                  /* tp_dictoffset */

    /* Currently using pyob_EventConduit_create instead of a conventional
     * __init__ method: */
    0,                                  /* tp_init */

    0,                                  /* tp_alloc */
    0,                                  /* tp_new */
  };

static int init_kidb_event_system(void) {
  /* EventConduitType is a new-style class, so PyType_Ready must be called
   * before its getters and setters will function. */
  if (PyType_Ready(&EventConduitType) < 0) { goto fail; }

  events__PyInt_zero = PyInt_FromLong(0);
  if (events__PyInt_zero == NULL) { goto fail; }

  return 0;

  fail:
    /* This function is indirectly called by the module loader, which makes no
     * provision for error recovery. */
    return -1;
} /* init_kidb_event_system */

/*********** EventConduit CLASS DEFINITION AND INITIALIZATION:END ************/

#endif /* ENABLE_DB_EVENT_SUPPORT */