<commit_before>## Copyright (c) 2010, Coptix, Inc.  All rights reserved.
## See the LICENSE file for license terms and warranty disclaimer.

"""state -- xmpp connection state and event management"""

from __future__ import absolute_import
from . import interfaces as i
from .prelude import *
import threading
import logging

__all__ = ('State', )



### State

class State(object):
    """Manage events, synchronize plugins, keep plugin state."""

    def __init__(self, core, plugins=None):
        self.core = core
        self.plugins = plugins or NoPlugins()

        self.locked = False
        self.schedule = deque()
        self.events = ddict(list)
        self.stanzas = {}
        self.state = {}
        self.rlock = threading.RLock()

    def reset(self, clear_events = False):
        return self.flush(True).clear(clear_events).install()

    def install(self):
        self.plugins.install(self)
        return self

    def activate(self):
        self.plugins.activate(self)
        return self

    def clear(self, clear_events = False):
        logging.debug("CLEAR %r %r" % (self.core, self))
        self.locked = False
        self.schedule.clear()
        if clear_events:
            self.events.clear()
        self.stanzas.clear()
        self.state.clear()
        return self

    ## ---------- Plugin State ----------

    def get(self, name, default=None):
        return self.state.get(name, default)

    def set(self, name, value):
        self.state[name] = value
        return self

    ## ---------- Events ----------

    def bind(self, kind, callback):
        logging.debug("BIND core: %r state: %r trigger %r: handlers %r" % (self.core, self, kind, callback))
        self.events[kind].append(callback)
        return self

    def one(self, kind, callback):
        return self.bind(kind, Once(callback))

    def unbind(self, kind, callback):
        logging.debug("UNBIND core: %r state: %r trigger %r: handlers %r" % (self.core, self, kind, callback))
        if kind in self.events:
            try:
                self.events[kind].remove(callback)
            except ValueError:
                pass
        return self

    def trigger(self, event, *args, **kwargs):
        handlers = self.events.get(event)
        logging.debug("core: %r state: %r trigger %r: handlers %r" % (self.core, self, event, handlers))
        if handlers:
            idx = 0; lim = len(handlers)
            while idx < lim:
                handler = handlers[idx]
                logging.debug("trigger %r: run %r" % (event, handler))
                self.run(handler, *args, **kwargs)
                if isinstance(handler, Once):
                    del handlers[idx]
                    lim -= 1
                else:
                    idx += 1
        return self

    ## ---------- Stanzas ----------

    def is_stanza(self, name):
        return name in self.stanzas

    def bind_stanza(self, name, callback, replace=True):
        exists = self.stanzas.get(name)
        if exists and not replace:
            raise ValueError('The %r stanza is handled by %r.' % (
                name,
                exists
            ))
        self.stanzas[name] = callback
        return self

    def one_stanza(self, name, callback, *args, **kwargs):
        return self.bind_stanza(name, Once(callback), *args, **kwargs)

    def unbind_stanza(name, callback=None):
        try:
            if not callback or self.stanzas[name] is callback:
                del self.stanzas[name]
        except KeyError:
            pass
        return self

    def trigger_stanza(self, name, *args, **kwargs):
        handler = self.stanzas.get(name)
        if not handler:
            raise i.StreamError(
                'unsupported-stanza-type',
                'Unrecognized stanza %r.\nstanzas: %r\nevents: %r\nstate: %r\nlocked: %r\ncore: %r\nplugins: %r\ncore.authJID: %r' % (name, self.stanzas, self.events, self.state, self.locked, self.core, self.plugins, self.core.authJID)
            )
        elif isinstance(handler, Once):
            del self.stanzas[name]
        return self.run(handler, *args, **kwargs)

    ## ---------- Synchronization ----------

    @contextmanager
    def lock(self):
        """A re-entrant lock that guards events and writes to the
        stream.  This is useful for coordinating activity across many
        plugins.  When the lock is released, pending jobs are run."""

        orig = self.locked
        self.rlock.acquire()
        try:
            self.locked = True
            yield self
        finally:
            self.rlock.release()
            self.locked = orig
            if not orig:
                self.schedule and self.flush()

    def run(self, method, *args, **kwargs):
        """Run or schedule a job; if delayed, it will be run later
        through flush()."""

        if self.locked:
            self.schedule.append(partial(method, *args, **kwargs))
            return self

        with self.lock():
            if debug_on:
                logging.debug("running  method %r" % method)
                if isinstance(method, partial):
                    logging.debug("running  method %r" % method.func)
            method(*args, **kwargs)
            if debug_on:
                logging.debug("finished method %r" % method)
        return self

    def flush(self, force=False):
        """Try to flush any scheduled jobs."""

        if not self.schedule or (self.locked and not force):
            return self

        try:
            self.locked = True
            while self.schedule:
                if debug_on:
                    name = "%r" % self.schedule[0]
                    logging.debug("delayed running  method %s" % name)
                    if isinstance(self.schedule[0], partial):
                        logging.debug("delayed running  method %r" % self.schedule[0].func)
                self.schedule.popleft()()
                if debug_on:
                    logging.debug("delayed finished method %s" % name)
            return self
        finally:
            self.locked = False

class Once(namedtuple('once', 'callback')):
    """An event handler that should only be called once."""

    def __call__(self, *args, **kwargs):
        return self.callback(*args, **kwargs)

class NoPlugins(i.PluginManager):

    def install(self, state):
        pass

    def activate_default(self, state):
        pass
<commit_msg>use log facilities correctly<commit_after>## Copyright (c) 2010, Coptix, Inc.  All rights reserved.
## See the LICENSE file for license terms and warranty disclaimer.

"""state -- xmpp connection state and event management"""

from __future__ import absolute_import
from . import interfaces as i
from .prelude import *
import threading

__all__ = ('State', )



### State

class State(object):
    """Manage events, synchronize plugins, keep plugin state."""

    def __init__(self, core, plugins=None):
        self.core = core
        self.plugins = plugins or NoPlugins()

        self.locked = False
        self.schedule = deque()
        self.events = ddict(list)
        self.stanzas = {}
        self.state = {}
        self.rlock = threading.RLock()

    def reset(self, clear_events = False):
        return self.flush(True).clear(clear_events).install()

    def install(self):
        self.plugins.install(self)
        return self

    def activate(self):
        self.plugins.activate(self)
        return self

    def clear(self, clear_events = False):
        log.debug("CLEAR %r %r" % (self.core, self))
        self.locked = False
        self.schedule.clear()
        if clear_events:
            self.events.clear()
        self.stanzas.clear()
        self.state.clear()
        return self

    ## ---------- Plugin State ----------

    def get(self, name, default=None):
        return self.state.get(name, default)

    def set(self, name, value):
        self.state[name] = value
        return self

    ## ---------- Events ----------

    def bind(self, kind, callback):
        log.debug("BIND core: %r state: %r trigger %r: handlers %r" % (self.core, self, kind, callback))
        self.events[kind].append(callback)
        return self

    def one(self, kind, callback):
        return self.bind(kind, Once(callback))

    def unbind(self, kind, callback):
        log.debug("UNBIND core: %r state: %r trigger %r: handlers %r" % (self.core, self, kind, callback))
        if kind in self.events:
            try:
                self.events[kind].remove(callback)
            except ValueError:
                pass
        return self

    def trigger(self, event, *args, **kwargs):
        handlers = self.events.get(event)
        log.debug("core: %r state: %r trigger %r: handlers %r" % (self.core, self, event, handlers))
        if handlers:
            idx = 0; lim = len(handlers)
            while idx < lim:
                handler = handlers[idx]
                log.debug("trigger %r: run %r" % (event, handler))
                self.run(handler, *args, **kwargs)
                if isinstance(handler, Once):
                    del handlers[idx]
                    lim -= 1
                else:
                    idx += 1
        return self

    ## ---------- Stanzas ----------

    def is_stanza(self, name):
        return name in self.stanzas

    def bind_stanza(self, name, callback, replace=True):
        exists = self.stanzas.get(name)
        if exists and not replace:
            raise ValueError('The %r stanza is handled by %r.' % (
                name,
                exists
            ))
        self.stanzas[name] = callback
        return self

    def one_stanza(self, name, callback, *args, **kwargs):
        return self.bind_stanza(name, Once(callback), *args, **kwargs)

    def unbind_stanza(name, callback=None):
        try:
            if not callback or self.stanzas[name] is callback:
                del self.stanzas[name]
        except KeyError:
            pass
        return self

    def trigger_stanza(self, name, *args, **kwargs):
        handler = self.stanzas.get(name)
        if not handler:
            raise i.StreamError(
                'unsupported-stanza-type',
                'Unrecognized stanza %r.\nstanzas: %r\nevents: %r\nstate: %r\nlocked: %r\ncore: %r\nplugins: %r\ncore.authJID: %r' % (name, self.stanzas, self.events, self.state, self.locked, self.core, self.plugins, self.core.authJID)
            )
        elif isinstance(handler, Once):
            del self.stanzas[name]
        return self.run(handler, *args, **kwargs)

    ## ---------- Synchronization ----------

    @contextmanager
    def lock(self):
        """A re-entrant lock that guards events and writes to the
        stream.  This is useful for coordinating activity across many
        plugins.  When the lock is released, pending jobs are run."""

        orig = self.locked
        self.rlock.acquire()
        try:
            self.locked = True
            log.debug("%r %r -> LOCKED" % (self, orig))
            yield self
        finally:
            self.rlock.release()
            self.locked = orig
            log.debug("%r LOCKED -> %r" % (self, orig))
            if not orig:
                log.debug("%r UNLOCKED, flushing" % self)
                self.schedule and self.flush()

    def run(self, method, *args, **kwargs):
        """Run or schedule a job; if delayed, it will be run later
        through flush()."""

        if self.locked:
            self.schedule.append(partial(method, *args, **kwargs))
            return self

        with self.lock():
            log.debug("++++ running  method %r" % method)
            if isinstance(method, partial):
                log.debug("    ====> %r" % method.func)
            method(*args, **kwargs)
            log.debug("---- finished method %r" % method)
        return self

    def flush(self, force=False):
        """Try to flush any scheduled jobs."""

        if not self.schedule or (self.locked and not force):
            return self

        try:
            self.locked = True
            while self.schedule:
                name = "%r" % self.schedule[0]
                log.debug("++++ delayed running method %s" % name)
                if isinstance(self.schedule[0], partial):
                    log.debug("     ====> %r" % self.schedule[0].func)
                self.schedule.popleft()()
                log.debug("---- delayed finished method %s" % name)
            return self
        finally:
            self.locked = False

class Once(namedtuple('once', 'callback')):
    """An event handler that should only be called once."""

    def __call__(self, *args, **kwargs):
        return self.callback(*args, **kwargs)

class NoPlugins(i.PluginManager):

    def install(self, state):
        pass

    def activate_default(self, state):
        pass

{'fraction_non_alphanumeric': 0.12496119217634274, 'fraction_numerical': 0.001552312946289972, 'mean_word_length': 1.8139331731819175, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4174602', 'n_tokens_mistral': 3663, 'n_tokens_neox': 3512, 'n_words': 1052}