<commit_before>import logging
from twisted.python import failure
from twisted.internet import error, defer
from twisted.internet.protocol import Protocol, ServerFactory
import json

from lbrynet.core.utils import is_valid_blobhash


log = logging.getLogger(__name__)


class ReflectorServer(Protocol):

    def connectionMade(self):
        peer_info = self.transport.getPeer()
        log.debug('Connection made to %s', peer_info)
        self.peer = self.factory.peer_manager.get_peer(peer_info.host, peer_info.port)
        self.blob_manager = self.factory.blob_manager
        self.received_handshake = False
        self.peer_version = None
        self.receiving_blob = False
        self.incoming_blob = None
        self.blob_write = None
        self.blob_finished_d = None
        self.cancel_write = None
        self.request_buff = ""

    def connectionLost(self, reason=failure.Failure(error.ConnectionDone())):
        pass

    def dataReceived(self, data):
        if self.receiving_blob:
            log.debug('Writing data to blob')
            self.blob_write(data)
        else:
            log.debug('Not yet recieving blob, data needs further processing')
            self.request_buff += data
            msg, extra_data = self._get_valid_response(self.request_buff)
            if msg is not None:
                self.request_buff = ''
                d = self.handle_request(msg)
                d.addCallbacks(self.send_response, self.handle_error)
                if self.receiving_blob and extra_data:
                    log.debug('Writing extra data to blob')
                    self.blob_write(extra_data)

    def _get_valid_response(self, response_msg):
        extra_data = None
        response = None
        curr_pos = 0
        while True:
            next_close_paren = response_msg.find('}', curr_pos)
            if next_close_paren != -1:
                curr_pos = next_close_paren + 1
                try:
                    response = json.loads(response_msg[:curr_pos])
                except ValueError:
                    if curr_pos > 100:
                        raise Exception("error decoding response")
                    else:
                        pass
                else:
                    extra_data = response_msg[curr_pos:]
                    break
            else:
                break
        return response, extra_data

    def handle_request(self, request_dict):
        if self.received_handshake is False:
            return self.handle_handshake(request_dict)
        else:
            return self.handle_normal_request(request_dict)

    def handle_handshake(self, request_dict):
        log.debug('Handling handshake')
        if 'version' not in request_dict:
            raise ValueError("Client should send version")
        self.peer_version = int(request_dict['version'])
        if self.peer_version != 0:
            raise ValueError("I don't know that version!")
        self.received_handshake = True
        return defer.succeed({'version': 0})

    def determine_blob_needed(self, blob):
        if blob.is_validated():
            return {'send_blob': False}
        else:
            self.incoming_blob = blob
            self.blob_finished_d, self.blob_write, self.cancel_write = blob.open_for_writing(self.peer)
            return {'send_blob': True}

    def close_blob(self):
        self.blob_finished_d = None
        self.blob_write = None
        self.cancel_write = None
        self.incoming_blob = None
        self.receiving_blob = False<commit_msg>blob_completed before moving to next one<commit_after>import logging
from twisted.python import failure
from twisted.internet import error, defer
from twisted.internet.protocol import Protocol, ServerFactory
import json

from lbrynet.core.utils import is_valid_blobhash


log = logging.getLogger(__name__)


class ReflectorServer(Protocol):

    def connectionMade(self):
        peer_info = self.transport.getPeer()
        log.debug('Connection made to %s', peer_info)
        self.peer = self.factory.peer_manager.get_peer(peer_info.host, peer_info.port)
        self.blob_manager = self.factory.blob_manager
        self.received_handshake = False
        self.peer_version = None
        self.receiving_blob = False
        self.incoming_blob = None
        self.blob_write = None
        self.blob_finished_d = None
        self.cancel_write = None
        self.request_buff = ""

    def connectionLost(self, reason=failure.Failure(error.ConnectionDone())):
        pass

    def dataReceived(self, data):
        if self.receiving_blob:
            log.debug('Writing data to blob')
            self.blob_write(data)
        else:
            log.debug('Not yet recieving blob, data needs further processing')
            self.request_buff += data
            msg, extra_data = self._get_valid_response(self.request_buff)
            if msg is not None:
                self.request_buff = ''
                d = self.handle_request(msg)
                d.addCallbacks(self.send_response, self.handle_error)
                if self.receiving_blob and extra_data:
                    log.debug('Writing extra data to blob')
                    self.blob_write(extra_data)

    def _get_valid_response(self, response_msg):
        extra_data = None
        response = None
        curr_pos = 0
        while True:
            next_close_paren = response_msg.find('}', curr_pos)
            if next_close_paren != -1:
                curr_pos = next_close_paren + 1
                try:
                    response = json.loads(response_msg[:curr_pos])
                except ValueError:
                    if curr_pos > 100:
                        raise Exception("error decoding response")
                    else:
                        pass
                else:
                    extra_data = response_msg[curr_pos:]
                    break
            else:
                break
        return response, extra_data

    def handle_request(self, request_dict):
        if self.received_handshake is False:
            return self.handle_handshake(request_dict)
        else:
            return self.handle_normal_request(request_dict)

    def handle_handshake(self, request_dict):
        log.debug('Handling handshake')
        if 'version' not in request_dict:
            raise ValueError("Client should send version")
        self.peer_version = int(request_dict['version'])
        if self.peer_version != 0:
            raise ValueError("I don't know that version!")
        self.received_handshake = True
        return defer.succeed({'version': 0})

    def determine_blob_needed(self, blob):
        if blob.is_validated():
            return {'send_blob': False}
        else:
            self.incoming_blob = blob
            self.blob_finished_d, self.blob_write, self.cancel_write = blob.open_for_writing(self.peer)
            self.blob_finished_d.addCallback(lambda _ :self.blob_manager.blob_completed(blob))
            return {'send_blob': True}

    def close_blob(self):
        self.blob_finished_d = None
        self.blob_write = None
        self.cancel_write = None
        self.incoming_blob = None
        self.receiving_blob = False
{'fraction_non_alphanumeric': 0.07778391820945012, 'fraction_numerical': 0.0022105554020447637, 'mean_word_length': 2.02507312996239, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 5, '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': '5267059', 'n_tokens_mistral': 2004, 'n_tokens_neox': 1917, 'n_words': 444}