<commit_before>from datetime import datetime, timedelta
from gevent.event import AsyncResult
from gevent.queue import Channel, Queue


from .protocol import STATE_NEW, STATE_OPEN, STATE_CLOSING, STATE_CLOSED
from .protocol import FRAME_OPEN, FRAME_CLOSE
from .protocol import FRAME_MESSAGE, FRAME_MESSAGE_BLOB, FRAME_HEARTBEAT
from .exceptions import SessionIsAcquired, SessionIsClosed

from .protocol import MSG_CLOSE, MSG_MESSAGE
from .protocol import close_frame, message_frame, messages_frame
from .protocol import SockjsMessage, OpenMessage, ClosedMessage

import collections
import logging
import gevent

log = logging.getLogger('sockjs_flask')


class Session(object):
    """ SockJS session object
    ``state``: Session state
    ``manager``: Session manager that hold this session
    ``acquired``: Acquired state, indicates that transport is using session
    ``timeout``: Session timeout
    """

    manager = None
    acquired = False
    state = STATE_NEW
    interrupted = False
    exception = None

    def __init__(self, id, handler, *, timeout=timedelta(seconds=5), debug=True):
        self.id = id
        self.handler = handler
        self.expired = False
        self.timeout = timeout
        self.expires = datetime.now() + timeout
        self._hits = 0
        self._heartbeats = 0
        self._heartbeat_transport = False
        self._debug = debug
        self._waiter = None
        self._queue = Queue()

    def __str__(self):
        result = ['id=%r' % (self.id,)]
        if self.state == STATE_OPEN:
            result.append('connected')
        elif self.state == STATE_CLOSED:
            result.append('closed')
        else:
            result.append('disconnected')
        if self.acquired:
            result.append('acquired')
        if len(self._queue):
            result.append('queue[%s]' % len(self._queue))
        if self._hits:
            result.append('hits=%s' % self._hits)
        if self._heartbeats:
            result.append('heartbeats=%s' % self._heartbeats)
        return ' '.join(result)

    def _tick(self, timeout=None):
        self.expired = False

        if timeout is None:
            self.expires = datetime.now() + self.timeout
        else:
            self.expires = datetime.now() + timeout
        log.info("Update expires time {}".format(self.expires))

    def _acquire(self, manager, heartbeat=True):
        self.acquired = True
        self.manager = manager
        self._heartbeat_transport = heartbeat

        self._tick()
        self._hits += 1

        if self.state == STATE_NEW:
            log.debug('open session: %s', self.id)
            self.state = STATE_OPEN
            self._feed(FRAME_OPEN, FRAME_OPEN)
            try:
                gevent.spawn(self.handler, OpenMessage, self)
            except Exception as exc:
                self.state = STATE_CLOSING
                self.exception = exc
                self.interrupted = True
                self._feed(FRAME_CLOSE, (3000, 'Internal error'))
                log.exception('Exception in open session handling.')

    def _release(self):
        self.acquired = False
        self.manager = None
        self._heartbeat_transport = False

    def _heartbeat(self):
        self.expired = False
        self._tick()
        self._heartbeats += 1
        if self._heartbeat:
            self._feed(FRAME_HEARTBEAT, FRAME_HEARTBEAT)

    def _feed(self, frame, data):
        log.info('session closed: %s', self.id)
        self._queue.put_nowait((frame, data))
        #if frame == FRAME_MESSAGE:
        #    if self._queue and self._queue[-1][0] == FRAME_MESSAGE:
        #        self._queue[-1][1].append(data)
        #    else:
        #        self._queue.append((frame, [data]))
        #else:
        #    self._queue.append((frame, data))
        waiter = self._waiter
        if waiter is not None:
            self._waiter = None
            if not waiter.cancelled():
                waiter.set_result(True)

    def _wait(self, pack=True):
        log.info("Waiter {} session and queue {}".format(self._waiter, self._queue))
        if not self._queue and self.state != STATE_CLOSED:
            assert not self._waiter
            self._waiter = AsyncResult()

        if self._queue:
            frame, payload = self._queue.get()
            if pack:
                if frame == FRAME_CLOSE:
                    return FRAME_CLOSE, close_frame(*payload)
                elif frame == FRAME_MESSAGE:
                    return FRAME_MESSAGE, messages_frame(payload)
            return frame, payload
        else:
            raise SessionIsClosed()

    def _remote_close(self, exc=None):
        """
        close session from remote.
        """
        if self.state in (STATE_CLOSING, STATE_CLOSED):
            return

        log.info('close session: %s', self.id)
        self.state = STATE_CLOSING
        if exc is not None:
            self.exception = exc
            self.interrupted = True
        try:
            self.handler(SockjsMessage(MSG_CLOSE, exc), self)
        except:
            log.exception('Exception in close handler.')

    def _remote_closed(self):
        if self.state == STATE_CLOSED:
            return

        log.info('session closed: %s', self.id)
        self.state = STATE_CLOSED
        self.expire()
        try:
            gevent.spawn(self.handler, ClosedMessage, self)
        except:
            log.exception('Exception in closed handler.')

        waiter = self._waiter
        if waiter is not None:
            self._waiter = None
            if not waiter.cancelled():
                waiter.set_result(True)

    def _remote_message(self, msg):
        log.debug('incoming message: %s, %s', self.id, msg[:200])
        self._tick()
        try:
            self.handler(SockjsMessage(MSG_MESSAGE, msg), self)
        except:
            log.exception('Exception in message handler.')

    def _remote_messages(self, messages):
        self._tick()
        for msg in messages:
            log.debug('incoming message: %s, %s', self.id, msg[:200])
            try:
                if self.manager is not None:
                    gevent.spawn(self.handler, SockjsMessage(MSG_MESSAGE, msg), self)
            except:
                log.exception('Exception in message handler.')

    def expire(self):
        self.expired = True
<commit_msg>up session<commit_after>from datetime import datetime, timedelta
from gevent.event import AsyncResult
from gevent.queue import Channel, Queue


from .protocol import STATE_NEW, STATE_OPEN, STATE_CLOSING, STATE_CLOSED
from .protocol import FRAME_OPEN, FRAME_CLOSE
from .protocol import FRAME_MESSAGE, FRAME_MESSAGE_BLOB, FRAME_HEARTBEAT
from .exceptions import SessionIsAcquired, SessionIsClosed

from .protocol import MSG_CLOSE, MSG_MESSAGE
from .protocol import close_frame, message_frame, messages_frame
from .protocol import SockjsMessage, OpenMessage, ClosedMessage

import logging
import gevent

log = logging.getLogger('sockjs_flask')


class Session(object):
    """ SockJS session object
    ``state``: Session state
    ``manager``: Session manager that hold this session
    ``acquired``: Acquired state, indicates that transport is using session
    ``timeout``: Session timeout
    """

    manager = None
    acquired = False
    state = STATE_NEW
    interrupted = False
    exception = None

    def __init__(self, id, handler, *, timeout=timedelta(seconds=5), debug=True):
        self.id = id
        self.handler = handler
        self.expired = False
        self.timeout = timeout
        self.expires = datetime.now() + timeout
        self._hits = 0
        self._heartbeats = 0
        self._heartbeat_transport = False
        self._debug = debug
        self._waiter = None
        self._queue = Queue()

    def __str__(self):
        result = ['id=%r' % (self.id,)]
        if self.state == STATE_OPEN:
            result.append('connected')
        elif self.state == STATE_CLOSED:
            result.append('closed')
        else:
            result.append('disconnected')
        if self.acquired:
            result.append('acquired')
        if len(self._queue):
            result.append('queue[%s]' % len(self._queue))
        if self._hits:
            result.append('hits=%s' % self._hits)
        if self._heartbeats:
            result.append('heartbeats=%s' % self._heartbeats)
        return ' '.join(result)

    def _tick(self, timeout=None):
        log.info("Update expires time {} for session".format(self.expires, self.id))
        self.expired = False
        if timeout is None:
            self.expires = datetime.now() + self.timeout
        else:
            self.expires = datetime.now() + timeout

    def _acquire(self, manager, heartbeat=True):
        self.acquired = True
        self.manager = manager
        self._heartbeat_transport = heartbeat
        self._tick()
        self._hits += 1
        if self.state == STATE_NEW:
            log.debug('open session: %s', self.id)
            self.state = STATE_OPEN
            self._feed(FRAME_OPEN, FRAME_OPEN)
            try:
                self.handler(OpenMessage, self)
            except Exception as exc:
                self.state = STATE_CLOSING
                self.exception = exc
                self.interrupted = True
                self._feed(FRAME_CLOSE, (3000, 'Internal error'))
                log.exception('Exception in open session handling.')

    def _release(self):
        self.acquired = False
        self.manager = None
        self._heartbeat_transport = False

    def _heartbeat(self):
        self.expired = False
        self._tick()
        self._heartbeats += 1
        if self._heartbeat:
            self._feed(FRAME_HEARTBEAT, FRAME_HEARTBEAT)

    def _feed(self, frame, data):
        log.info('session closed: %s', self.id)
        self._queue.put_nowait((frame, data))
        #if frame == FRAME_MESSAGE:
        #    if self._queue and self._queue[-1][0] == FRAME_MESSAGE:
        #        self._queue[-1][1].append(data)
        #    else:
        #        self._queue.append((frame, [data]))
        #else:
        #    self._queue.append((frame, data))
        waiter = self._waiter
        if waiter is not None:
            self._waiter = None
            if not waiter.cancelled():
                waiter.set_result(True)

    def _wait(self, pack=True):
        log.info("Waiter {} session and queue {}".format(self._waiter, self._queue))
        if not self._queue and self.state != STATE_CLOSED:
            assert not self._waiter
            self._waiter = AsyncResult()

        if self._queue:
            frame, payload = self._queue.get()
            if pack:
                if frame == FRAME_CLOSE:
                    return FRAME_CLOSE, close_frame(*payload)
                elif frame == FRAME_MESSAGE:
                    return FRAME_MESSAGE, messages_frame(payload)
            return frame, payload
        else:
            raise SessionIsClosed()

    def _remote_close(self, exc=None):
        """
        Close session from remote.
        """
        if self.state in (STATE_CLOSING, STATE_CLOSED):
            return

        log.info('close session: %s', self.id)
        self.state = STATE_CLOSING
        if exc is not None:
            self.exception = exc
            self.interrupted = True
        try:
            self.handler(SockjsMessage(MSG_CLOSE, exc), self)
        except Exception as e:
            log.exception('Exception in close handler.')

    def _remote_closed(self):
        if self.state == STATE_CLOSED:
            return

        log.info('session closed: %s', self.id)
        self.state = STATE_CLOSED
        self.expire()
        try:
            self.handler(ClosedMessage, self)
        except Exception as e:
            log.exception('Exception in closed handler.')

        waiter = self._waiter
        if waiter is not None:
            self._waiter = None
            if not waiter.cancelled():
                waiter.set_result(True)

    def _remote_message(self, msg):
        log.debug('incoming message: %s, %s', self.id, msg[:200])
        self._tick()
        try:
            self.handler(SockjsMessage(MSG_MESSAGE, msg), self)
        except:
            log.exception('Exception in message handler.')

    def _remote_messages(self, messages):
        self._tick()
        for msg in messages:
            log.debug('incoming message: %s, %s', self.id, msg[:200])
            try:
                if self.manager is not None:
                    gevent.spawn(self.handler, SockjsMessage(MSG_MESSAGE, msg), self)
            except:
                log.exception('Exception in message handler.')

    def expire(self):
        self.expired = True

{'fraction_non_alphanumeric': 0.09979633401221996, 'fraction_numerical': 0.0029766567444775185, 'mean_word_length': 1.9934349355216883, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, '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': '626463', 'n_tokens_mistral': 3708, 'n_tokens_neox': 3537, 'n_words': 911}