<commit_before>#!/usr/bin/env python
# -*- coding: utf8 -*-
import struct
#Local imports
from YakDB.Conversion import ZMQBinaryUtil
from YakDB.Exceptions import ParameterException, YakDBProtocolException
from YakDB.DataProcessor import ClientSidePassiveJob
#Use ZMQPy inside PyPy
import platform
if platform.python_implementation() == "PyPy":
    import zmqpy as zmq
else:
    import zmq


class Connection:
    """
    An instance of this class represents a connection to a YakDB database.
    """
    def __init__(self, endpoints=None, context=None):
        """
        Create a new YakDB connection instance.
        @param endpoints An endpoint string or list of endpoints to connect to.
            New endpoints can be added dynamicallys
        """
        if context is None:
            self.context = zmq.Context()
            self.cleanupContextOnDestruct = True
        else:
            self.context = context
            self.cleanupContextOnDestruct = False
        self.socket = None
        self.numConnections = 0
        #Connect to the endpoints, if any
        if endpoints is None:
            pass
        elif type(endpoints) is str:
            self.connect(endpoints)
        elif type(endpoints) is list:
            for endpoint in endpoints:
                self.connect(endpoint)
        else: #Unknown parameter type
            raise ParameterException("Endpoints parameter is a '%s' but expected a list or a string!" % (str(type(endpoints))))
    def __del__(self):
        """
        Cleanup ZMQ resources.
        Auto-destroys context if the context was created in the constructor.
        """
        if self.socket is not None:
            self.socket.close()
        if self.cleanupContextOnDestruct:
            self.context.destroy()
    def useRequestReplyMode(self):
        """Sets the current ZeroDB connection into Request/reply mode (default)"""
        self.socket = self.context.socket(zmq.REQ)
        self.mode = zmq.REQ
    def usePushMode(self):
        """Sets the current ZeroDB connection into Push/pull mode (default)"""
        self.socket = self.context.socket(zmq.PUSH)
        self.mode = zmq.PUSH
    def usePubMode(self):
        """Sets the current ZeroDB connection into publish/subscribe mode"""
        self.socket = self.context.socket(zmq.PUB)
        self.mode = zmq.PUB
    def connect(self, endpoints):
        """
        Connect to a YakDB server.
        @param endpoint The ZMQ endpoint to connect to, e.g. tcp://localhost:7100.
                An array is also allowed
        """
        if isinstance(endpoints, str):
            endpoints = [endpoints]
        #Use request/reply as default
        if self.socket == None:
            self.useRequestReplyMode()
        for endpoint in endpoints:
            self._checkParameterType(endpoint, str, "[one of the endpoints]")
            self.socket.connect(endpoint)
        self.numConnections += len(endpoints)
    def _checkSingleConnection(self):
        """
        Check if the current instance is connected to
        exactly one server. If not, raise an exception.
        """
        if self.numConnections > 1:
            raise ConnectionStateException("This operation can only be executed with exactly one connection, but currently %d connections are active" % self.numConnections)
    def _checkRequestReply(self):
        """
        Check if the current instance is correctly setup for REQ/REP, else raise an exception
        """
        self._checkConnection()
        if self.mode is not zmq.REQ:
            raise Exception("Only request/reply connections support this message type!")
    def _checkConnection(self):
        """Check if the current instance is correctly connected, else raise an exception"""
        if self.socket is None:
            raise Exception("Please connect to server before using serverInfo (use ZeroDBConnection.connect()!")
        if self.numConnections <= 0:
            raise Exception("Current ZeroDBConnection is setup, but not connected. Please connect before usage!")
    def _sendBinary32(self, value, more=True):
        """
        Send a binary 32-bit number (little-endian) over the current socket
        """
        ZMQBinaryUtil.sendBinary32(self.socket, value, more)
    def _sendBinary64(self, value, more=True):
        """
        Send a binary 32-bit number (little-endian) over the current socket
        """
        ZMQBinaryUtil.sendBinary64(self.socket,  value,  more)
    def _sendRange(self, startKey,  endKey,  more=False):
        """
        Send a dual-frame range over the socket.
        If any of the keys is None or empty, a zero-sized frame is sent
        @param more If this is set to true, not only the range start frame but also the range end frame is sent
            with the ZMQ_SNDMORE flag
        """
        if startKey is not None: startKey = ZMQBinaryUtil.convertToBinary(startKey)
        else: startKey = ""
        if endKey is not None: endKey = ZMQBinaryUtil.convertToBinary(endKey)
        else: endKey = ""
        self.socket.send(startKey, zmq.SNDMORE)
        self.socket.send(endKey,  (zmq.SNDMORE if more else 0))
    def _checkHeaderFrame(self,  msgParts, expectedResponseType):
        """
        Given a list of received message parts, checks the first message part.
        Checks performed:
            - Magic byte (expects 0x31)
            - Version byte (expects 0x01)
            - Response code (expects 0x00, else it assumes that the next frame, is the error msg)
        This function throws an exception on check failure and exits normally on success
        """
        if len(msgParts) == 0:
            raise YakDBProtocolException("Received empty reply message")
        if len(msgParts[0]) < 4:
            raise YakDBProtocolException("Header frame has size of %d, but expected 4"
                                         % len(msgParts[0]))
        if msgParts[0][2] != expectedResponseType:
            raise YakDBProtocolException("Response code received from server is "
                        "%d instead of %d" % (ord(msgParts[0][2]),  ord(expectedResponseType)))
        if msgParts[0][3] != '\x00':
            errorMsg = msgParts[1] if len(msgParts) >= 2 else "<Unknown>"
            raise YakDBProtocolException(
                "Response status code is %d instead of 0x00 (ACK), error message: %s"
                % (ord(msgParts[0][3]),  errorMsg))
        #Parse and return request ID, if any
        if len(msgParts[0]) > 4:
            return msgParts[0][4:]
        return None #No request ID
    def serverInfo(self):
        """
        Send a server info request to the server and return the version string
        @return The server version string
        """
        self._checkRequestReply()
        self._checkSingleConnection()
        self.socket.send("\x31\x01\x00")<commit_msg>Improve size protocol error debuggability<commit_after>#!/usr/bin/env python
# -*- coding: utf8 -*-
import struct
#Local imports
from YakDB.Conversion import ZMQBinaryUtil
from YakDB.Exceptions import ParameterException, YakDBProtocolException
from YakDB.DataProcessor import ClientSidePassiveJob
#Use ZMQPy inside PyPy
import platform
if platform.python_implementation() == "PyPy":
    import zmqpy as zmq
else:
    import zmq


class Connection:
    """
    An instance of this class represents a connection to a YakDB database.
    """
    def __init__(self, endpoints=None, context=None):
        """
        Create a new YakDB connection instance.
        @param endpoints An endpoint string or list of endpoints to connect to.
            New endpoints can be added dynamicallys
        """
        if context is None:
            self.context = zmq.Context()
            self.cleanupContextOnDestruct = True
        else:
            self.context = context
            self.cleanupContextOnDestruct = False
        self.socket = None
        self.numConnections = 0
        #Connect to the endpoints, if any
        if endpoints is None:
            pass
        elif type(endpoints) is str:
            self.connect(endpoints)
        elif type(endpoints) is list:
            for endpoint in endpoints:
                self.connect(endpoint)
        else: #Unknown parameter type
            raise ParameterException("Endpoints parameter is a '%s' but expected a list or a string!" % (str(type(endpoints))))
    def __del__(self):
        """
        Cleanup ZMQ resources.
        Auto-destroys context if the context was created in the constructor.
        """
        if self.socket is not None:
            self.socket.close()
        if self.cleanupContextOnDestruct:
            self.context.destroy()
    def useRequestReplyMode(self):
        """Sets the current ZeroDB connection into Request/reply mode (default)"""
        self.socket = self.context.socket(zmq.REQ)
        self.mode = zmq.REQ
    def usePushMode(self):
        """Sets the current ZeroDB connection into Push/pull mode (default)"""
        self.socket = self.context.socket(zmq.PUSH)
        self.mode = zmq.PUSH
    def usePubMode(self):
        """Sets the current ZeroDB connection into publish/subscribe mode"""
        self.socket = self.context.socket(zmq.PUB)
        self.mode = zmq.PUB
    def connect(self, endpoints):
        """
        Connect to a YakDB server.
        @param endpoint The ZMQ endpoint to connect to, e.g. tcp://localhost:7100.
                An array is also allowed
        """
        if isinstance(endpoints, str):
            endpoints = [endpoints]
        #Use request/reply as default
        if self.socket == None:
            self.useRequestReplyMode()
        for endpoint in endpoints:
            self._checkParameterType(endpoint, str, "[one of the endpoints]")
            self.socket.connect(endpoint)
        self.numConnections += len(endpoints)
    def _checkSingleConnection(self):
        """
        Check if the current instance is connected to
        exactly one server. If not, raise an exception.
        """
        if self.numConnections > 1:
            raise ConnectionStateException("This operation can only be executed with exactly one connection, but currently %d connections are active" % self.numConnections)
    def _checkRequestReply(self):
        """
        Check if the current instance is correctly setup for REQ/REP, else raise an exception
        """
        self._checkConnection()
        if self.mode is not zmq.REQ:
            raise Exception("Only request/reply connections support this message type!")
    def _checkConnection(self):
        """Check if the current instance is correctly connected, else raise an exception"""
        if self.socket is None:
            raise Exception("Please connect to server before using serverInfo (use ZeroDBConnection.connect()!")
        if self.numConnections <= 0:
            raise Exception("Current ZeroDBConnection is setup, but not connected. Please connect before usage!")
    def _sendBinary32(self, value, more=True):
        """
        Send a binary 32-bit number (little-endian) over the current socket
        """
        ZMQBinaryUtil.sendBinary32(self.socket, value, more)
    def _sendBinary64(self, value, more=True):
        """
        Send a binary 32-bit number (little-endian) over the current socket
        """
        ZMQBinaryUtil.sendBinary64(self.socket,  value,  more)
    def _sendRange(self, startKey,  endKey,  more=False):
        """
        Send a dual-frame range over the socket.
        If any of the keys is None or empty, a zero-sized frame is sent
        @param more If this is set to true, not only the range start frame but also the range end frame is sent
            with the ZMQ_SNDMORE flag
        """
        if startKey is not None: startKey = ZMQBinaryUtil.convertToBinary(startKey)
        else: startKey = ""
        if endKey is not None: endKey = ZMQBinaryUtil.convertToBinary(endKey)
        else: endKey = ""
        self.socket.send(startKey, zmq.SNDMORE)
        self.socket.send(endKey,  (zmq.SNDMORE if more else 0))
    def _checkHeaderFrame(self,  msgParts, expectedResponseType):
        """
        Given a list of received message parts, checks the first message part.
        Checks performed:
            - Magic byte (expects 0x31)
            - Version byte (expects 0x01)
            - Response code (expects 0x00, else it assumes that the next frame, is the error msg)
        This function throws an exception on check failure and exits normally on success
        """
        if len(msgParts) == 0:
            raise YakDBProtocolException("Received empty reply message")
        if len(msgParts[0]) < 4:
            looksLikeAHeaderFrame = (len(msgParts[0]) >= 1)
            if len(msgParts[0]) >= 1 and msgParts[0][0] != '\x31': looksLikeAHeaderFrame = False
            if len(msgParts[0]) >= 2 and msgParts[0][1] != '\x01': looksLikeAHeaderFrame = False
            raise YakDBProtocolException("Reponse header frame has size of %d, but expected size-4 frame%s"
                                         % (len(msgParts[0]),
                                           (", it doesn't even look like a header frame" if not looksLikeAHeaderFrame
                                               else ", but it looks like some kind of header frame")))
        if msgParts[0][2] != expectedResponseType:
            raise YakDBProtocolException("Response code received from server is "
                        "%d instead of %d" % (ord(msgParts[0][2]),  ord(expectedResponseType)))
        if msgParts[0][3] != '\x00':
            errorMsg = msgParts[1] if len(msgParts) >= 2 else "<Unknown>"
            raise YakDBProtocolException(
                "Response status code is %d instead of 0x00 (ACK), error message: %s"
                % (ord(msgParts[0][3]),  errorMsg))
        #Parse and return request ID, if any
        if len(msgParts[0]) > 4:
            return msgParts[0][4:]
        return None #No request ID
    def serverInfo(self):
        """
        Send a server info request to the server and return the version string
        @return The server version string
        """
        self._checkRequestReply()
        self._checkSingleConnection()
        self.socket.send("\x31\x01\x00")
{'fraction_non_alphanumeric': 0.08520934146854618, 'fraction_numerical': 0.009397573462374641, 'mean_word_length': 2.2826887661141804, 'pattern_counts': {'":': 2, '<': 9, '<?xml version=': 0, '>': 14, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 8, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6019425', 'n_tokens_mistral': 3814, 'n_tokens_neox': 3677, 'n_words': 1321}