def make_psycopg_green():
    """Configure psycopg2 to call our wait function
    """
    if not hasattr(extensions, 'set_wait_callback'):
        raise ImportError('support for coroutines not available in this psycopg version ({})'
                          .format(psycopg2.__version__))

    extensions.set_wait_callback(psycopg2_wait_cb)

def make_psycopg_green():
    """Configure Psycopg to be used with gevent in non-blocking way."""
    if not hasattr(extensions, 'set_wait_callback'):
        raise ImportError(
            "support for coroutines not available in this Psycopg version (%s)"
            % psycopg2.__version__)

    extensions.set_wait_callback(gevent_wait_callback)

def make_asynchronous():
    try:
        extensions.POLL_OK
    except AttributeError:
        raise ImproperlyConfigured(
            'Psycopg2 does not have support for asynchronous connections. '
            'You need at least version 2.2.0 of Psycopg2.')
    extensions.set_wait_callback(psycopg2_wait_callback)

def unmake_psycopg_green():
    """Undo make_psycopg_green()."""
    if not hasattr(extensions, 'set_wait_callback'):
        raise ImportError(
            "support for coroutines not available in this Psycopg version (%s)"
            % psycopg2.__version__)

    extensions.set_wait_callback(None)

def make_green(engine):
    """
    Set up psycopg2 & SQLAlchemy to be greenlet-friendly.
    Note: psycogreen does not really monkey patch psycopg2 in the
    manner that gevent monkey patches socket.
    """
    log.warn('Making the system green...')

    extensions.set_wait_callback(gevent_wait_callback)

    # Assuming that gevent monkey patched the builtin
    # threading library, we're likely good to use
    # SQLAlchemy's QueuePool, which is the default
    # pool class.  However, we need to make it use
    # threadlocal connections
    #
    #
    engine.pool._use_threadlocal = True

def db_config(name, user, password, host='127.0.0.1', port=5432, pool_size=10, pool_block=1, patch_psycopg2_with_gevent=True, log=None, **kwargs):

    ### _db_config

    _db_config.clear()
    _db_config.update(database=name, user=user, password=password, host=host, port=port, **kwargs)

    ### pool

    global _db_pool
    connections_to_create = pool_size - _db_pool.qsize()

    if connections_to_create > 0:  # Create connections.
        if pool_block > connections_to_create:
            pool_block = connections_to_create
        _connect(pool_block)
        gevent.spawn(_connect, connections_to_create - pool_block)

    else:  # Delete connections.
        for _ in xrange(-connections_to_create):
            try:
                db_conn = _db_pool.get(block=False)
            except Empty:
                break
            try:
                db_conn.close()
            except Exception:
                pass

    ### patch_psycopg2_with_gevent

    if patch_psycopg2_with_gevent:
        extensions.set_wait_callback(_gevent_wait_callback)

    ### log

    global _log
    _log = log

def main():
    global running, channel_mngr, workers

    # Make green psycopg:
    extensions.set_wait_callback(gevent_wait_callback)

    # Decide how many processes to use
    try:
        proc_count = int(sys.argv[1])
    except:
        proc_count = multiprocessing.cpu_count()

    workers = [None] * proc_count

    channel.init(proc_count)
    
    gevent.signal(signal.SIGINT, sig_quit)
    gevent.signal(signal.SIGTERM, sig_quit)
    
    # Remove the sockets directory
    shutil.rmtree(sock_dir, True)
    os.mkdir(sock_dir)

    # Spawn the reloaders for the workers
    reloaders = [gevent.spawn(reloader, i) for i in xrange(proc_count)]

    setproctitle.setproctitle('diggems master')

    # Channel manager process reloader:
    while running:
        print 'Starting channel manager process.'
        proc = gipc.start_process(channel.rpc_dispatcher, daemon=True, name='channel_mngr')
        proc.join()
        print 'Channel manager process has quit.'
    print 'Done with channel manager'

    gevent.joinall(reloaders)
    print 'All done, quiting'

# Cast all database input to unicode automatically.
# See http://initd.org/psycopg/docs/usage.html#unicode-handling for more info.
ext.register_type(ext.UNICODE)
ext.register_type(ext.UNICODEARRAY)
ext.register_type(ext.new_type((705,), "UNKNOWN", ext.UNICODE))
# See https://github.com/dbcli/pgcli/issues/426 for more details.
# This registers a unicode type caster for datatype 'RECORD'.
ext.register_type(ext.new_type((2249,), "RECORD", ext.UNICODE))

# Cast bytea fields to text. By default, this will render as hex strings with
# Postgres 9+ and as escaped binary in earlier versions.
ext.register_type(ext.new_type((17,), 'BYTEA_TEXT', psycopg2.STRING))

# When running a query, make pressing CTRL+C raise a KeyboardInterrupt
# See http://initd.org/psycopg/articles/2014/07/20/cancelling-postgresql-statements-python/
ext.set_wait_callback(psycopg2.extras.wait_select)


def register_date_typecasters(connection):
    """
    Casts date and timestamp values to string, resolves issues with out of
    range dates (e.g. BC) which psycopg2 can't handle
    """
    def cast_date(value, cursor):
        return value
    cursor = connection.cursor()
    cursor.execute('SELECT NULL::date')
    date_oid = cursor.description[0][1]
    cursor.execute('SELECT NULL::timestamp')
    timestamp_oid = cursor.description[0][1]
    cursor.execute('SELECT NULL::timestamptz')

def gevent_wait_callback(conn, timeout=None):
    """A wait callback useful to allow gevent to work with Psycopg."""
    while 1:
        state = conn.poll()
        if state == extensions.POLL_OK:
            break
        elif state == extensions.POLL_READ:
            wait_read(conn.fileno(), timeout=timeout)
        elif state == extensions.POLL_WRITE:
            wait_write(conn.fileno(), timeout=timeout)
        else:
            raise OperationalError(
                "Bad result from poll: %r" % state)


extensions.set_wait_callback(gevent_wait_callback)


class DatabaseConnectionPool(object):

    def __init__(self, maxsize=100):
        if not isinstance(maxsize, (int, long)):
            raise TypeError('Expected integer, got %r' % (maxsize, ))
        self.maxsize = maxsize
        self.pool = Queue()
        self.size = 0

    def get(self):
        pool = self.pool
        if self.size >= self.maxsize or pool.qsize():
            return pool.get()

from sakura.common.access import GRANT_LEVELS

# psycopg should let gevent switch to other greenlets
def wait_callback(conn, timeout=None):
    while True:
        state = conn.poll()
        if state == POLL_OK:
            break
        elif state == POLL_READ:
            wait_read(conn.fileno(), timeout=timeout)
        elif state == POLL_WRITE:
            wait_write(conn.fileno(), timeout=timeout)
        else:
            raise psycopg2.OperationalError("Bad result from poll: %r" % state)

set_wait_callback(wait_callback)

DEBUG_CURSORS=False
#DEBUG_CURSORS=True

TYPES_SAKURA_TO_PG = {
    'int8':     'smallint',
    'int16':    'smallint',
    'int32':    'integer',
    'int64':    'bigint',
    'float32':  'real',
    'float64':  'double precision',
    'string':   'text',
    'bool':     'boolean',
    'date':     'timestamp with time zone'
}

    while 1:
        if panic:
            raise Exception('whatever')

        state = conn.poll()
        if state == extensions.POLL_OK:
            break
        elif state == extensions.POLL_READ:
            trampoline(conn.fileno(), read=True)
        elif state == extensions.POLL_WRITE:
            trampoline(conn.fileno(), write=True)
        else:
            raise psycopg2.OperationalError(
                "Bad result from poll: %r" % state)

extensions.set_wait_callback(wait_cb)


# SIGHUP handler to inject a fail in the callback

def handler(signum, frame):
    panic.append(True)

signal.signal(signal.SIGHUP, handler)


# Simulate another green thread working

def worker():
    while 1:
        print "I'm working"

def make_psycopg_green():
    if not hasattr(extensions, 'set_wait_callback'):
        raise ImportError(
            "Support for coroutines is available only from Psycopg 2.2.0")

    extensions.set_wait_callback(gevent_wait_callback)

    def create_engine(self):
        """Create the database engine"""
        log.debug("Creating ENGINE")
        options = {
            'convert_unicode': self.native_unicode,
            'poolclass': GreenQueuePool,
            'pool_size': self.pool_size,
            'pool_recycle': self.pool_recycle,
            'pool_timeout': self.pool_timeout

        }

        if self.database_uri.startswith('sqlite:'):

            options.pop('pool_timeout')

            match = _sqlite_re.match(self.database_uri)
            if match is None:
                raise ArgumentError('Could not parse rfc1738 URL')
            database, query = match.groups()
            if database is None:
                database = ':memory:'
            if query:
                query = url_decode(query).to_dict()
            else:
                query = {}
            info = URL('sqlite', database=database, query=query)
            pool_size = options.get('pool_size', 0)
            # we go to memory and the pool size was explicitly set to 0
            # which is fail.  Let the user know that
            if info.database in (None, '', ':memory:'):
                if pool_size == 0:
                    raise RuntimeError('SQLite in memory database with an '
                                       'empty queue not possible due to data '
                                       'loss.')
                # if pool size is None or explicitly set to 0 we assume the
                # user did not want a queue for this sqlite connection and
                # hook in the null pool.
                elif not pool_size:
                    log.warn("SQLite database is using the NullPool")
                    from sqlalchemy.pool import NullPool
                    options['poolclass'] = NullPool
        else:
            info = make_url(self.database_uri)
            # if mysql is the database engine, god forbid, and no connection
            # encoding is provided we set it to utf-8
            if info.drivername == 'mysql':
                info.query.setdefault('charset', 'utf8')
                options.setdefault('pool_size', 10)
                options.setdefault('pool_recycle', 7200)
            elif info.drivername.startswith('postgresql+psycopg2'):
                from psycopg2 import extensions, OperationalError
                from gevent.socket import wait_read, wait_write
                options['use_native_unicode'] = self.native_unicode

                def wait_callback(conn, timeout=None):
                    """
                    A wait callback useful to allow gevent to work with Psycopg.
                    https://bitbucket.org/dvarrazzo/psycogreen/src/tip/gevent/
                    """
                    while True:
                        state = conn.poll()
                        if state == extensions.POLL_OK:
                            break
                        elif state == extensions.POLL_READ:
                            wait_read(conn.fileno(), timeout=timeout)
                        elif state == extensions.POLL_WRITE:
                            wait_write(conn.fileno(), timeout=timeout)
                        else:
                            raise OperationalError(
                                "Bad result from poll: %r" % state
                            )
                if hasattr(extensions, 'set_wait_callback'):
                    extensions.set_wait_callback(wait_callback)

        dialect_cls = info.get_dialect()

        # get the correct DBAPI base on connection url
        dbapi_args = {}
        dbapi = dialect_cls.dbapi(**dbapi_args)

        # create the dialect
        dialect_args = {'dbapi':dbapi}
        dialect = dialect_cls(**dialect_args)

        # assemble connection arguments for this dialect
        (cargs, connection_params) = dialect.create_connect_args(info)
        log.debug("CARGS: %s; CONNECTION_PARAMS: %s;", cargs, connection_params)
        log.debug("Creating db engine. info: %s; options: %s;", info, options)
        engine = sqlalchemy.create_engine(info, **options)
        database_engine_created.send(self, engine=engine)

def patch():
    gevent_psycopg2.monkey_patch()
    extensions.set_wait_callback(gevent_wait_callback)

    
def pulsar_wait_callback(conn, timeout=None):
    """A wait callback useful to allow gevent to work with Psycopg."""
    while 1:
        state = conn.poll()
        if state == extensions.POLL_OK:
            break
        elif state == extensions.POLL_READ:
            wait_read(conn.fileno(), timeout=timeout)
        elif state == extensions.POLL_WRITE:
            wait_write(conn.fileno(), timeout=timeout)
        else:
            raise psycopg2.OperationalError(
                "Bad result from poll: %r" % state)

extensions.set_wait_callback(pulsar_wait_callback)


class Async(object):

    def wait(self, callback, errback=None, registered=False):
        exc = None
        loop = self._loop
        try:
            state = self.connection.poll()
        except Exception:
            exc = sys.exc_info()
            if registered:
                loop.remove_connector(self._sock_fd)
        else:
            if state == POLL_OK:

            elif state == POLL_READ:
                select.select([conn.fileno()], [], [], _WAIT_SELECT_TIMEOUT)
            elif state == POLL_WRITE:
                select.select([], [conn.fileno()], [], _WAIT_SELECT_TIMEOUT)
            else:
                raise conn.OperationalError("bad state from poll: %s" % state)
        except KeyboardInterrupt:
            conn.cancel()
            # the loop will be broken by a server error
            continue


# When running a query, make pressing CTRL+C raise a KeyboardInterrupt
# See http://initd.org/psycopg/articles/2014/07/20/cancelling-postgresql-statements-python/
# See also https://github.com/psycopg/psycopg2/issues/468
ext.set_wait_callback(_wait_select)


def register_date_typecasters(connection):
    """
    Casts date and timestamp values to string, resolves issues with out of
    range dates (e.g. BC) which psycopg2 can't handle
    """
    def cast_date(value, cursor):
        return value
    cursor = connection.cursor()
    cursor.execute('SELECT NULL::date')
    date_oid = cursor.description[0][1]
    cursor.execute('SELECT NULL::timestamp')
    timestamp_oid = cursor.description[0][1]
    cursor.execute('SELECT NULL::timestamp with time zone')

def monkey_patch():
    """Configure psycopg2 to be used with gevent in non-blocking way."""

    extensions.set_wait_callback(gevent_wait_callback)