<commit_before>from __future__ import absolute_import, unicode_literals

import ssl
import zlib
import errno
import atexit
import select
import socket
import logging
import msgpack
import threading
import collections

logger = logging.getLogger(__name__)

import synapse.common as s_common
import synapse.lib.scope as s_scope
import synapse.lib.threads as s_threads
import synapse.lib.thisplat as s_thisplat

from synapse.eventbus import EventBus

def sockgzip(byts):
    blen = len(byts)
    byts = zlib.compress(byts)
    #print('GZIP DELTA: %d -> %d' % (blen,len(byts)))
    return s_common.msgenpack(('sock:gzip', {'data': byts}))

class Socket(EventBus):
    '''
    Wrapper for the builtin socket.Socket class.

    Args:
        sock socket.socket: socket to wrap
        **info:
    '''

    def __init__(self, sock, **info):
        EventBus.__init__(self)

        self.sock = sock  # type: socket.socket
        self.unpk = msgpack.Unpacker(use_list=0, encoding='utf8')
        self.iden = s_common.guid()

        self.info = info
        self.blocking = True    # sockets are blocking by default

        if self.info.get('nodelay', True):
            self._tryTcpNoDelay()

        self.txbuf = None   # the remainder of a partially sent byts
        self.txque = collections.deque()
        self.rxque = collections.deque()

        self.onfini(self._finiSocket)

    def addTxByts(self, byts):
        '''
        Add bytes to the non-blocking TX queue

        Args:

            byts (bytes): The bytes to send

        '''
        self.txque.append(byts)
        self.fire('sock:tx:add')

    def runTxLoop(self):
        '''
        Run a pass through the non-blocking tx loop.

        Returns:
            (bool): True if there is still more work to do
        '''
        while True:

            if not self.txbuf:

                if not self.txque:
                    break

                self.txbuf = self.txque.popleft()
                self.fire('sock:tx:pop')

            sent = self.send(self.txbuf)
            self.txbuf = self.txbuf[sent:]

            # if we still have a txbuf after sending
            # we could only send part of the buffer
            if self.txbuf:
                break

        if not self.txbuf and not self.txque:
            return False

        return True

    def get(self, prop, defval=None):
        '''
        Retrieve a property from the socket's info dict.

        Example:

            if sock.get('listen'):
                dostuff()

        '''
        return self.info.get(prop, defval)

    def set(self, prop, valu):
        '''
        Set a property on the Socket by name.

        Example:

            sock.set('woot', 30)

        '''
        self.info[prop] = valu

    def recvall(self, size):
        '''
        Recieve the exact number of bytes requested.
        Returns None on if socket closes early.

        Example:

            byts = sock.recvall(300)
            if byts == None:
                return

            dostuff(byts)

        Notes:
            * this API will trigger fini() on close

        '''
        byts = b''
        remain = size

        try:
            while remain:
                x = self.sock.recv(remain)
                if not x:
                    return None
                byts += x
                remain -= len(x)

        except socket.error as e:
            # fini triggered above.
            return None

        return byts

    def recvobj(self):
        for mesg in self:
            return mesg

    def addTxByts(self, byts):
        self.txque.append(byts)
        self.fire('sock:tx:add', sock=self)

    def setblocking(self, valu):
        valu = bool(valu)
        self.blocking = valu
        self.sock.setblocking(valu)

    def tx(self, mesg):
        '''
        Transmit a mesg tufo ( type, info ) via the socket using msgpack.
        If present this API is safe for use with a socket in a Plex().
        '''

        byts = s_common.msgenpack(mesg)
        return self.txbytes(byts)

    def txbytes(self, byts):

        # we may support gzip on the socket message
        if len(byts) > 50000 and self.get('sock:can:gzip'):
            byts = sockgzip(byts)

        # if the socket is non-blocking assume someone is managing
        # the socket via sock:tx:add events
        if not self.blocking:
            self.addTxByts(byts)
            return True

        try:
            self.sendall(byts)
            return True
        except socket.error as e:
            self.fini()
            return False

    def rx(self):<commit_msg>added docstring to setblocking() and made addTxByts internal<commit_after>from __future__ import absolute_import, unicode_literals

import ssl
import zlib
import errno
import atexit
import select
import socket
import logging
import msgpack
import threading
import collections

logger = logging.getLogger(__name__)

import synapse.common as s_common
import synapse.lib.scope as s_scope
import synapse.lib.threads as s_threads
import synapse.lib.thisplat as s_thisplat

from synapse.eventbus import EventBus

def sockgzip(byts):
    blen = len(byts)
    byts = zlib.compress(byts)
    #print('GZIP DELTA: %d -> %d' % (blen,len(byts)))
    return s_common.msgenpack(('sock:gzip', {'data': byts}))

class Socket(EventBus):
    '''
    Wrapper for the builtin socket.Socket class.

    Args:
        sock socket.socket: socket to wrap
        **info:
    '''

    def __init__(self, sock, **info):
        EventBus.__init__(self)

        self.sock = sock  # type: socket.socket
        self.unpk = msgpack.Unpacker(use_list=0, encoding='utf8')
        self.iden = s_common.guid()

        self.info = info
        self.blocking = True    # sockets are blocking by default

        if self.info.get('nodelay', True):
            self._tryTcpNoDelay()

        self.txbuf = None   # the remainder of a partially sent byts
        self.txque = collections.deque()
        self.rxque = collections.deque()

        self.onfini(self._finiSocket)

    def _addTxByts(self, byts):
        self.txque.append(byts)
        self.fire('sock:tx:add')

    def runTxLoop(self):
        '''
        Run a pass through the non-blocking tx loop.

        Returns:
            (bool): True if there is still more work to do
        '''
        while True:

            if not self.txbuf:

                if not self.txque:
                    break

                self.txbuf = self.txque.popleft()
                self.fire('sock:tx:pop')

            sent = self.send(self.txbuf)
            self.txbuf = self.txbuf[sent:]

            # if we still have a txbuf after sending
            # we could only send part of the buffer
            if self.txbuf:
                break

        if not self.txbuf and not self.txque:
            return False

        return True

    def get(self, prop, defval=None):
        '''
        Retrieve a property from the socket's info dict.

        Example:

            if sock.get('listen'):
                dostuff()

        '''
        return self.info.get(prop, defval)

    def set(self, prop, valu):
        '''
        Set a property on the Socket by name.

        Example:

            sock.set('woot', 30)

        '''
        self.info[prop] = valu

    def recvall(self, size):
        '''
        Recieve the exact number of bytes requested.
        Returns None on if socket closes early.

        Example:

            byts = sock.recvall(300)
            if byts == None:
                return

            dostuff(byts)

        Notes:
            * this API will trigger fini() on close

        '''
        byts = b''
        remain = size

        try:
            while remain:
                x = self.sock.recv(remain)
                if not x:
                    return None
                byts += x
                remain -= len(x)

        except socket.error as e:
            # fini triggered above.
            return None

        return byts

    def recvobj(self):
        for mesg in self:
            return mesg

    def setblocking(self, valu):
        '''
        Set the socket's blocking mode to True/False.

        Args:
            valu (bool): False to set socket non-blocking
        '''
        valu = bool(valu)
        self.blocking = valu
        self.sock.setblocking(valu)

    def tx(self, mesg):
        '''
        Transmit a mesg tufo ( type, info ) via the socket using msgpack.
        If present this API is safe for use with a socket in a Plex().
        '''

        byts = s_common.msgenpack(mesg)
        return self.txbytes(byts)

    def txbytes(self, byts):

        # we may support gzip on the socket message
        if len(byts) > 50000 and self.get('sock:can:gzip'):
            byts = sockgzip(byts)

        # if the socket is non-blocking assume someone is managing
        # the socket via sock:tx:add events
        if not self.blocking:
            self._addTxByts(byts)
            return True

        try:
            self.sendall(byts)
            return True
        except socket.error as e:
            self.fini()
            return False

    def rx(self):
{'fraction_non_alphanumeric': 0.08831225468818142, 'fraction_numerical': 0.0026166593981683385, 'mean_word_length': 1.6580701246015648, 'pattern_counts': {'":': 0, '<': 3, '<?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': '4587588', 'n_tokens_mistral': 2790, 'n_tokens_neox': 2626, 'n_words': 813}