<commit_before>import json
import logging
from collections import defaultdict
import traceback

from zmq.eventloop import ioloop
import zmq
from multiprocessing import Process, Queue
from threading import Thread
ioloop.install()
import tornado
import constants

from protocol import goodbye
import network_util
import hashlib
import routingtable
import datastore
from urlparse import urlparse


# Connection to one peer
class PeerConnection(object):
    def __init__(self, transport, address):
        # timeout in seconds
        self._timeout = 10
        self._transport = transport
        self._address = address
        self._log = logging.getLogger(self.__class__.__name__)

    def create_socket(self):
        self._ctx = zmq.Context()
        self._socket = self._ctx.socket(zmq.REQ)
        self._socket.setsockopt(zmq.LINGER, 0)
        self._socket.connect(self._address)

    def cleanup_socket(self):
        self._socket.close()

    def send(self, data):
        self.send_raw(json.dumps(data))

    def send_raw(self, serialized):

        queue = Queue()
        Thread(target=self._send_raw, args=(serialized,queue,)).start()
        msg = queue.get()
        print msg

        pass

    def _send_raw(self, serialized, raw_queue):

        # pyzmq sockets are not threadsafe,
        # they have to run in a separate process
        queue = Queue()
        # queue element is false if something went wrong and the peer
        # has to be removed

        p = Process(target=self._send_raw_process, args=(serialized, queue))
        p.start()
        msg = queue.get()
        if not msg:
            self._log.info("Peer %s timed out." % self._address)
            #self._transport.remove_peer(self._address, self._guid)
        else:
            raw_queue.put(msg)

        p.join()



    def _send_raw_process(self, serialized, queue):

        self.create_socket()
        self._socket.send(serialized)

        poller = zmq.Poller()
        poller.register(self._socket, zmq.POLLIN)
        if poller.poll(self._timeout * 1000):
            msg = self._socket.recv()
            self.on_message(msg)
            self.cleanup_socket()
            queue.put(msg)

        else:
            self._log.info("Node timed out: %s" % self._address)
            self.cleanup_socket()
            queue.put(False)


    def on_message(self, msg, callback=None):
        self._log.info("Message received: %s" % msg)


# Transport layer manages a list of peers
class TransportLayer(object):
    def __init__(self, my_ip, my_port, my_guid):
        self._peers = {}
        self._callbacks = defaultdict(list)
        self._port = my_port
        self._ip = my_ip
        self._guid = my_guid
        self._uri = 'tcp://%s:%s' % (self._ip, self._port)

        # Routing table
        self._routingTable = routingtable.OptimizedTreeRoutingTable(self.guid)
        self._dataStore = datastore.MongoDataStore()
        self._knownNodes = []


        self._log = logging.getLogger(self.__class__.__name__)
        # signal.signal(signal.SIGTERM, lambda x, y: self.broadcast_goodbye())

    def add_callback(self, section, callback):
        self._callbacks[section].append(callback)



    def trigger_callbacks(self, section, *data):
        for cb in self._callbacks[section]:
            cb(*data)
        if not section == 'all':
            for cb in self._callbacks['all']:
                cb(*data)

    def get_profile(self):
        return {'type': 'hello_request', 'uri': self._uri}





    def listen(self):
        t = Thread(target=self._listen)
        t.setDaemon(True)
        t.start()

    def _listen(self):
        self._log.info("Listening at: %s:%s" % (self._ip, self._port))
        self._ctx = zmq.Context()
        self._socket = self._ctx.socket(zmq.REP)

        if network_util.is_loopback_addr(self._ip):
            # we are in local test mode so bind that socket on the
            # specified IP
            self._socket.bind(self._uri)
        else:
            try:
              self._socket.bind('tcp://*:%s' % self._port)
            except:
              pass

        while True:
            message = self._socket.recv()
            self.on_raw_message(message)
            self._socket.send(json.dumps({'type': 'ok', 'senderGUID':self._transport.guid, 'pubkey':self._transport.pubkey}))

    def closed(self, *args):
        self._log.info("client left")

    def _init_peer(self, msg):
        uri = msg['uri']

        if uri not in self._peers:
            self._peers[uri] = CryptoPeerConnection(self, uri)

    def remove_peer(self, uri, guid):
        self._log.info("Removing peer %s", uri)
        ip = urlparse(uri).hostname
        port = urlparse(uri).port
        if (ip, port, guid) in self._shortlist:
            self._shortlist.remove((ip, port, guid))


        self._log.info('Removed')



        # try:
        #     del self._peers[uri]
        #     msg = {
        #         'type': 'peer_remove',
        #         'uri': uri
        #     }
        #     self.trigger_callbacks(msg['type'], msg)<commit_msg>Fixed reference to guid<commit_after>import json
import logging
from collections import defaultdict
import traceback

from zmq.eventloop import ioloop
import zmq
from multiprocessing import Process, Queue
from threading import Thread
ioloop.install()
import tornado
import constants

from protocol import goodbye
import network_util
import hashlib
import routingtable
import datastore
from urlparse import urlparse


# Connection to one peer
class PeerConnection(object):
    def __init__(self, transport, address):
        # timeout in seconds
        self._timeout = 10
        self._transport = transport
        self._address = address
        self._log = logging.getLogger(self.__class__.__name__)

    def create_socket(self):
        self._ctx = zmq.Context()
        self._socket = self._ctx.socket(zmq.REQ)
        self._socket.setsockopt(zmq.LINGER, 0)
        self._socket.connect(self._address)

    def cleanup_socket(self):
        self._socket.close()

    def send(self, data):
        self.send_raw(json.dumps(data))

    def send_raw(self, serialized):

        queue = Queue()
        Thread(target=self._send_raw, args=(serialized,queue,)).start()
        msg = queue.get()
        print msg

        pass

    def _send_raw(self, serialized, raw_queue):

        # pyzmq sockets are not threadsafe,
        # they have to run in a separate process
        queue = Queue()
        # queue element is false if something went wrong and the peer
        # has to be removed

        p = Process(target=self._send_raw_process, args=(serialized, queue))
        p.start()
        msg = queue.get()
        if not msg:
            self._log.info("Peer %s timed out." % self._address)
            #self._transport.remove_peer(self._address, self._guid)
        else:
            raw_queue.put(msg)

        p.join()



    def _send_raw_process(self, serialized, queue):

        self.create_socket()
        self._socket.send(serialized)

        poller = zmq.Poller()
        poller.register(self._socket, zmq.POLLIN)
        if poller.poll(self._timeout * 1000):
            msg = self._socket.recv()
            self.on_message(msg)
            self.cleanup_socket()
            queue.put(msg)

        else:
            self._log.info("Node timed out: %s" % self._address)
            self.cleanup_socket()
            queue.put(False)


    def on_message(self, msg, callback=None):
        self._log.info("Message received: %s" % msg)


# Transport layer manages a list of peers
class TransportLayer(object):
    def __init__(self, my_ip, my_port, my_guid):
        self._peers = {}
        self._callbacks = defaultdict(list)
        self._port = my_port
        self._ip = my_ip
        self._guid = my_guid
        self._uri = 'tcp://%s:%s' % (self._ip, self._port)

        # Routing table
        self._routingTable = routingtable.OptimizedTreeRoutingTable(self.guid)
        self._dataStore = datastore.MongoDataStore()
        self._knownNodes = []


        self._log = logging.getLogger(self.__class__.__name__)
        # signal.signal(signal.SIGTERM, lambda x, y: self.broadcast_goodbye())

    def add_callback(self, section, callback):
        self._callbacks[section].append(callback)



    def trigger_callbacks(self, section, *data):
        for cb in self._callbacks[section]:
            cb(*data)
        if not section == 'all':
            for cb in self._callbacks['all']:
                cb(*data)

    def get_profile(self):
        return {'type': 'hello_request', 'uri': self._uri}





    def listen(self):
        t = Thread(target=self._listen)
        t.setDaemon(True)
        t.start()

    def _listen(self):
        self._log.info("Listening at: %s:%s" % (self._ip, self._port))
        self._ctx = zmq.Context()
        self._socket = self._ctx.socket(zmq.REP)

        if network_util.is_loopback_addr(self._ip):
            # we are in local test mode so bind that socket on the
            # specified IP
            self._socket.bind(self._uri)
        else:
            try:
              self._socket.bind('tcp://*:%s' % self._port)
            except:
              pass

        while True:
            message = self._socket.recv()
            self.on_raw_message(message)
            self._socket.send(json.dumps({'type': 'ok', 'senderGUID':self._guid, 'pubkey':''}))

    def closed(self, *args):
        self._log.info("client left")

    def _init_peer(self, msg):
        uri = msg['uri']

        if uri not in self._peers:
            self._peers[uri] = CryptoPeerConnection(self, uri)

    def remove_peer(self, uri, guid):
        self._log.info("Removing peer %s", uri)
        ip = urlparse(uri).hostname
        port = urlparse(uri).port
        if (ip, port, guid) in self._shortlist:
            self._shortlist.remove((ip, port, guid))


        self._log.info('Removed')



        # try:
        #     del self._peers[uri]
        #     msg = {
        #         'type': 'peer_remove',
        #         'uri': uri
        #     }
        #     self.trigger_callbacks(msg['type'], msg)
{'fraction_non_alphanumeric': 0.11118770927713216, 'fraction_numerical': 0.001378766988378964, 'mean_word_length': 2.17046518888542, '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': '5722105', 'n_tokens_mistral': 3089, 'n_tokens_neox': 2971, 'n_words': 708}