<commit_before>
"""
..module:: file_queue
"""

import os
import sys
import time
import struct
import cStringIO
import random

from twisted.internet import reactor

from logs import lg

from lib import udp
from system import tmpfile
from main import settings
from lib import misc
from contacts import contactsdb

import udp_interface
import udp_session
import udp_stream

#------------------------------------------------------------------------------ 

MAX_SIMULTANEOUS_STREAMS_PER_SESSION = 16
NUMBER_OF_STREAMS_TO_REMEMBER = MAX_SIMULTANEOUS_STREAMS_PER_SESSION * 4

#------------------------------------------------------------------------------ 

_StreamCounter = 0

#------------------------------------------------------------------------------ 

class FileQueue:
    def __init__(self, session):
        self.session = session
        self.streams = {}
        self.outboxFiles = {}
        self.inboxFiles = {}
        self.outboxQueue = []
        self.dead_streams = []

    def make_unique_stream_id(self):
        # return int(str(random.randint(100, 999))+str(int(time.time() * 100.0))[7:])
        global _StreamCounter
        _StreamCounter += 1
        # if _StreamCounter >= 10000:
        #     _StreamCounter = 1
        n = max(contactsdb.contact_position(self.session.peer_idurl), 0) % 89
        return (n + 10) * 1000000 + random.randint(10, 99) * 10000 + _StreamCounter % 10000

    def report_failed_outbox_queue(self, error_message):
        for filename, description, result_defer, single in self.outboxQueue:
            self.on_failed_outbox_queue_item(
                filename, description, error_message, result_defer, single)

    def report_failed_outbox_files(self, error_message):
        for outfile in self.outboxFiles.values():
            outfile.status = 'failed'
            outfile.error_message = error_message
            self.report_outbox_file(outfile)

    def report_failed_inbox_files(self, error_message):
        for infile in self.inboxFiles.values():
            infile.status = 'failed'
            infile.error_message = error_message
            self.report_inbox_file(infile)

    def close(self):
        for stream in self.streams.values():
            stream.on_close()
        self.outboxQueue = []

    def do_send_data(self, stream_id, outfile, output):
        if udp_stream._Debug:
            import random
            if random.randint(1, 100) > 90:
                return True
        newoutput = ''.join((
            struct.pack('i', stream_id),
            struct.pack('i', outfile.size),
            output))
        return self.session.send_packet(udp.CMD_DATA, newoutput)

    def do_send_ack(self, stream_id, infile, ack_data):
        if udp_stream._Debug:
            import random
            if random.randint(1, 100) > 90:
                return True
        newoutput = ''.join((
            struct.pack('i', stream_id),
            ack_data))
        return self.session.send_packet(udp.CMD_ACK, newoutput)

    def append_outbox_file(self, filename, description='', result_defer=None, single=False):
        self.outboxQueue.append((filename, description, result_defer, single))
        lg.out(18, 'udp_file_queue.append_outbox_file %s for %s : %s, streams=%d, queue=%d' % (
            os.path.basename(filename), self.session.peer_id, description,
            len(self.streams), len(self.outboxQueue)))
        udp_session.process_sessions()
        # reactor.callLater(0, self.process_outbox_queue)

    def insert_outbox_file(self, filename, description='', result_defer=None, single=False):
        self.outboxQueue.insert(0, (filename, description, result_defer, single))
        lg.out(18, 'udp_file_queue.insert_outbox_file %s for %s : %s, streams=%d, queue=%d' % (
            os.path.basename(filename), self.session.peer_id, description,
            len(self.streams), len(self.outboxQueue)))
        udp_session.process_sessions()
        # reactor.callLater(0, self.process_outbox_queue)

    def process_outbox_queue(self):<commit_msg>remove debugging feature to drop random udp datagrams<commit_after>
"""
..module:: file_queue
"""

import os
import sys
import time
import struct
import cStringIO
import random

from twisted.internet import reactor

from logs import lg

from lib import udp
from system import tmpfile
from main import settings
from lib import misc
from contacts import contactsdb

import udp_interface
import udp_session
import udp_stream

#------------------------------------------------------------------------------ 

MAX_SIMULTANEOUS_STREAMS_PER_SESSION = 16
NUMBER_OF_STREAMS_TO_REMEMBER = MAX_SIMULTANEOUS_STREAMS_PER_SESSION * 4

#------------------------------------------------------------------------------ 

_StreamCounter = 0

#------------------------------------------------------------------------------ 

class FileQueue:
    def __init__(self, session):
        self.session = session
        self.streams = {}
        self.outboxFiles = {}
        self.inboxFiles = {}
        self.outboxQueue = []
        self.dead_streams = []

    def make_unique_stream_id(self):
        # return int(str(random.randint(100, 999))+str(int(time.time() * 100.0))[7:])
        global _StreamCounter
        _StreamCounter += 1
        # if _StreamCounter >= 10000:
        #     _StreamCounter = 1
        n = max(contactsdb.contact_position(self.session.peer_idurl), 0) % 89
        return (n + 10) * 1000000 + random.randint(10, 99) * 10000 + _StreamCounter % 10000

    def report_failed_outbox_queue(self, error_message):
        for filename, description, result_defer, single in self.outboxQueue:
            self.on_failed_outbox_queue_item(
                filename, description, error_message, result_defer, single)

    def report_failed_outbox_files(self, error_message):
        for outfile in self.outboxFiles.values():
            outfile.status = 'failed'
            outfile.error_message = error_message
            self.report_outbox_file(outfile)

    def report_failed_inbox_files(self, error_message):
        for infile in self.inboxFiles.values():
            infile.status = 'failed'
            infile.error_message = error_message
            self.report_inbox_file(infile)

    def close(self):
        for stream in self.streams.values():
            stream.on_close()
        self.outboxQueue = []

    def do_send_data(self, stream_id, outfile, output):
#         if udp_stream._Debug:
#             import random
#             if random.randint(1, 100) > 90:
#                 return True
        newoutput = ''.join((
            struct.pack('i', stream_id),
            struct.pack('i', outfile.size),
            output))
        return self.session.send_packet(udp.CMD_DATA, newoutput)

    def do_send_ack(self, stream_id, infile, ack_data):
#         if udp_stream._Debug:
#             import random
#             if random.randint(1, 100) > 90:
#                 return True
        newoutput = ''.join((
            struct.pack('i', stream_id),
            ack_data))
        return self.session.send_packet(udp.CMD_ACK, newoutput)

    def append_outbox_file(self, filename, description='', result_defer=None, single=False):
        self.outboxQueue.append((filename, description, result_defer, single))
        lg.out(18, 'udp_file_queue.append_outbox_file %s for %s : %s, streams=%d, queue=%d' % (
            os.path.basename(filename), self.session.peer_id, description,
            len(self.streams), len(self.outboxQueue)))
        udp_session.process_sessions()

    def insert_outbox_file(self, filename, description='', result_defer=None, single=False):
        self.outboxQueue.insert(0, (filename, description, result_defer, single))
        lg.out(18, 'udp_file_queue.insert_outbox_file %s for %s : %s, streams=%d, queue=%d' % (
            os.path.basename(filename), self.session.peer_id, description,
            len(self.streams), len(self.outboxQueue)))
        udp_session.process_sessions()

    def process_outbox_queue(self):
{'fraction_non_alphanumeric': 0.15384615384615385, 'fraction_numerical': 0.016483516483516484, 'mean_word_length': 2.2556910569105693, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 9, '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': '1187661', 'n_tokens_mistral': 2460, 'n_tokens_neox': 2300, 'n_words': 496}