<commit_before>from twisted.internet.protocol import Protocol
import random

class MQTTProtocol(Protocol):
    _packetTypes = { 0x00: "null",    0x01: "connect",     0x02: "connack",
                     0x03: "publish", 0x04: "puback",      0x05: "pubrec",
                     0x06: "pubrel",  0x07: "pubcomp",     0x08: "subscribe",
                     0x09: "suback",  0x0A: "unsubscribe", 0x0B: "unsuback",
                     0x0C: "pingreq", 0x0D: "pingresp",    0x0E: "disconnect" }
    
    buffer = bytearray()
    
    def dataReceived(self, data):
        self._accumulatePacket(data)
        
    def _accumulatePacket(self, data):
        self.buffer.extend(data)
        
        length = None
        
        while len(self.buffer):
            if length is None:
                # Start on a new packet
                
                # Haven't got enough data to start a new packet,
                # wait for some more
                if len(self.buffer) < 2: break 
                
                lenLen = 1
                # Calculate the length of the length field
                while lenLen < len(self.buffer):
                    if not self.buffer[lenLen] & 0x80: break
                    lenLen += 1
                
                # We still haven't got all of the remaining length field
                if lenLen < len(self.buffer) and self.buffer[lenLen] & 0x80: 
                    return
                
                length = self._decodeLength(self.buffer[1:])
                
            if len(self.buffer) >= length:
                chunk = self.buffer[:length + lenLen + 1]
                self._processPacket(chunk)
                self.buffer = self.buffer[length + lenLen + 1:]
                length = None

            else:
                break
                
    def _processPacket(self, packet):
        try:
            type = self._packetTypes[(packet[0] & 0xF0) >> 4]
            dup = (packet[0] & 0x08) == 0x08
            qos = (packet[0] & 0x06) >> 1
            retain = (packet[0] & 0x01) == 0x01
        except:
            # Invalid packet type, throw away this packet
            print "Invalid packet type"
            return
        
        # Strip the fixed header
        lenLen = 1
        while packet[lenLen] & 0x80:
            lenLen += 1
        
        packet = packet[lenLen+1:]
        
        # Get the appropriate handler function
        packetHandler = getattr(self, "_event_%s" % type, None)
        
        if packetHandler:
            packetHandler(packet, qos, dup, retain)
        else:
            # Rocks fall, everyone dies<commit_msg>Ensure the entire MQTT packet is in the buffer before processing the packet.<commit_after>from twisted.internet.protocol import Protocol
import random

class MQTTProtocol(Protocol):
    _packetTypes = { 0x00: "null",    0x01: "connect",     0x02: "connack",
                     0x03: "publish", 0x04: "puback",      0x05: "pubrec",
                     0x06: "pubrel",  0x07: "pubcomp",     0x08: "subscribe",
                     0x09: "suback",  0x0A: "unsubscribe", 0x0B: "unsuback",
                     0x0C: "pingreq", 0x0D: "pingresp",    0x0E: "disconnect" }
    
    buffer = bytearray()
    
    def dataReceived(self, data):
        self._accumulatePacket(data)
        
    def _accumulatePacket(self, data):
        self.buffer.extend(data)
        
        length = None
        
        while len(self.buffer):
            if length is None:
                # Start on a new packet
                
                # Haven't got enough data to start a new packet,
                # wait for some more
                if len(self.buffer) < 2: break 
                
                lenLen = 1
                # Calculate the length of the length field
                while lenLen < len(self.buffer):
                    if not self.buffer[lenLen] & 0x80: break
                    lenLen += 1
                
                # We still haven't got all of the remaining length field
                if lenLen < len(self.buffer) and self.buffer[lenLen] & 0x80: 
                    return
                
                length = self._decodeLength(self.buffer[1:])
                
            if len(self.buffer) >= 1 + lenLen + length:
                chunk = self.buffer[:1 + lenLen + length]
                self._processPacket(chunk)
                self.buffer = self.buffer[1 + lenLen + length:]
                length = None

            else:
                break
                
    def _processPacket(self, packet):
        try:
            type = self._packetTypes[(packet[0] & 0xF0) >> 4]
            dup = (packet[0] & 0x08) == 0x08
            qos = (packet[0] & 0x06) >> 1
            retain = (packet[0] & 0x01) == 0x01
        except:
            # Invalid packet type, throw away this packet
            print "Invalid packet type"
            return
        
        # Strip the fixed header
        lenLen = 1
        while packet[lenLen] & 0x80:
            lenLen += 1
        
        packet = packet[lenLen+1:]
        
        # Get the appropriate handler function
        packetHandler = getattr(self, "_event_%s" % type, None)
        
        if packetHandler:
            packetHandler(packet, qos, dup, retain)
        else:
            # Rocks fall, everyone dies
{'fraction_non_alphanumeric': 0.09017923823749066, 'fraction_numerical': 0.030433159073935772, 'mean_word_length': 1.3352223190932868, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 13, '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': '5332321', 'n_tokens_mistral': 1588, 'n_tokens_neox': 1445, 'n_words': 413}