# Path: twimp/utils.py
# class GeneratorWrapperProtocol(protocol.Protocol):
#     def __init__(self, proto=None):
#         self._buf = vecbuf.VecBuf()
#         self._toread = 0
# 
#         self._handler_changed = False
# 
#         self._handler = proto
#         if proto:
#             self._toread = self._handler.send(None)
# 
#     def init_handler(self, proto=None, do_init=True):
#         if proto:
#             if self._handler:
#                 # need to indicate we should retry on StopIteration...
#                 self._handler_changed = True
#             self._handler = proto
#         if do_init:
#             self._toread = self._handler.send(None)
# 
#     def connectionLost(self, reason=protocol.connectionDone):
#         # clean up
#         self._buf.read(len(self._buf))
#         self._handler = None
#         self._toread = None
# 
#     def dataReceived(self, data):
#         self._buf.write(data)
# 
#         while 1: # need to be able to retry immediately when handlers change
#             try:
#                 while len(self._buf) >= self._toread:
#                     self._toread = self._handler.send(self._buf)
#                 break           # no data, no retry
#             except StopIteration:
#                 if self._handler_changed:
#                     self._handler_changed = False
#                 else:
#                     self._toread = None
#                     self._handler = None
#                     break
#             except:
#                 # undocumented, but real; this way we can signal
#                 # protocol errors directly, and by not passing/raising
#                 # directly we avoid a stack trace log, with the same
#                 # effect - connectionLost
#                 f = failure.Failure()
#                 log.info(f.value, exc_info=log.isEnabledFor(logging.DEBUG))
#                 return f
# 
# class FrameSorter(object):
#     def __init__(self, callback, keys):
#         self._max = len(keys)
#         assert self._max > 1, 'need at least two sources'
#         self._queues = dict((k, [None, deque(), k]) for k in keys)
#         self._active = 0
# 
#         self._callback = callback
# 
#         self._key_getter = itemgetter(0)
#         self._grpos_getter = itemgetter(0)
# 
#     def add(self, key, grpos, data):
#         sq = self._queues[key]
#         if not sq[1]:
#             self._active += 1
#             sq[0] = grpos
#         sq[1].append((grpos, data))
#         self._check_send()
# 
#     def _check_send(self):
#         if self._active < self._max:
#             return
# 
#         while self._active >= self._max:
#             first_q = min(self._queues.itervalues(), key=self._key_getter)
#             key = first_q[2]
#             second_q = min((q for (k, q) in self._queues.iteritems()
#                             if k != key), key=self._key_getter)
# 
#             switch_grpos = second_q[0]
#             frame_queue = first_q[1]
#             while frame_queue:
#                 if frame_queue[0][0] > switch_grpos:
#                     break
#                 (grpos, frame) = frame_queue.popleft()
#                 self._callback(grpos, key, frame)
# 
#             if not frame_queue:
#                 self._active -= 1
#             else:
#                 first_q[0] = frame_queue[0][0]
# 
#     def flush(self):
#         sorted_frames = sorted(((grpos, key, frame)
#                                 for (min_grpos, frame_queue, key)
#                                     in self._queues.values()
#                                 for (grpos, frame) in frame_queue),
#                                key=self._grpos_getter)
#         self._active = 0
#         self._queues = dict((k, [None, deque(), k]) for k in self._queues)
# 
#         for f in sorted_frames:
#             self._callback(*f)


from twisted.trial import unittest
from twimp.utils import GeneratorWrapperProtocol, FrameSorter
from helpers import StringTransport

    def test_add_flush_add(self):
        fs = FrameSorter(self.frames_callback, [1, 2])

        fs.add(1, 0, 'a')
        fs.add(2, 1, 'b')
        fs.add(1, 2, 'a')

        self.assertEquals(self.frames, [(0, 1, 'a'),
                                        (1, 2, 'b')])
        self.clear()

        fs.flush()

        self.assertEquals(self.frames, [(2, 1, 'a')])
        self.clear()

        fs.add(2, 3, 'b')
        fs.add(1, 4, 'a')
        fs.add(2, 5, 'b')

        self.assertEquals(self.frames, [(3, 2, 'b'),
                                        (4, 1, 'a')])
        self.clear()

        fs.flush()

        self.assertEquals(self.frames, [(5, 2, 'b')])

    def test_not_sorted_input(self):

# Path: twimp/utils.py
# class GeneratorWrapperProtocol(protocol.Protocol):
#     def __init__(self, proto=None):
#         self._buf = vecbuf.VecBuf()
#         self._toread = 0
# 
#         self._handler_changed = False
# 
#         self._handler = proto
#         if proto:
#             self._toread = self._handler.send(None)
# 
#     def init_handler(self, proto=None, do_init=True):
#         if proto:
#             if self._handler:
#                 # need to indicate we should retry on StopIteration...
#                 self._handler_changed = True
#             self._handler = proto
#         if do_init:
#             self._toread = self._handler.send(None)
# 
#     def connectionLost(self, reason=protocol.connectionDone):
#         # clean up
#         self._buf.read(len(self._buf))
#         self._handler = None
#         self._toread = None
# 
#     def dataReceived(self, data):
#         self._buf.write(data)
# 
#         while 1: # need to be able to retry immediately when handlers change
#             try:
#                 while len(self._buf) >= self._toread:
#                     self._toread = self._handler.send(self._buf)
#                 break           # no data, no retry
#             except StopIteration:
#                 if self._handler_changed:
#                     self._handler_changed = False
#                 else:
#                     self._toread = None
#                     self._handler = None
#                     break
#             except:
#                 # undocumented, but real; this way we can signal
#                 # protocol errors directly, and by not passing/raising
#                 # directly we avoid a stack trace log, with the same
#                 # effect - connectionLost
#                 f = failure.Failure()
#                 log.info(f.value, exc_info=log.isEnabledFor(logging.DEBUG))
#                 return f
# 
# class FrameSorter(object):
#     def __init__(self, callback, keys):
#         self._max = len(keys)
#         assert self._max > 1, 'need at least two sources'
#         self._queues = dict((k, [None, deque(), k]) for k in keys)
#         self._active = 0
# 
#         self._callback = callback
# 
#         self._key_getter = itemgetter(0)
#         self._grpos_getter = itemgetter(0)
# 
#     def add(self, key, grpos, data):
#         sq = self._queues[key]
#         if not sq[1]:
#             self._active += 1
#             sq[0] = grpos
#         sq[1].append((grpos, data))
#         self._check_send()
# 
#     def _check_send(self):
#         if self._active < self._max:
#             return
# 
#         while self._active >= self._max:
#             first_q = min(self._queues.itervalues(), key=self._key_getter)
#             key = first_q[2]
#             second_q = min((q for (k, q) in self._queues.iteritems()
#                             if k != key), key=self._key_getter)
# 
#             switch_grpos = second_q[0]
#             frame_queue = first_q[1]
#             while frame_queue:
#                 if frame_queue[0][0] > switch_grpos:
#                     break
#                 (grpos, frame) = frame_queue.popleft()
#                 self._callback(grpos, key, frame)
# 
#             if not frame_queue:
#                 self._active -= 1
#             else:
#                 first_q[0] = frame_queue[0][0]
# 
#     def flush(self):
#         sorted_frames = sorted(((grpos, key, frame)
#                                 for (min_grpos, frame_queue, key)
#                                     in self._queues.values()
#                                 for (grpos, frame) in frame_queue),
#                                key=self._grpos_getter)
#         self._active = 0
#         self._queues = dict((k, [None, deque(), k]) for k in self._queues)
# 
#         for f in sorted_frames:
#             self._callback(*f)


# Path: test/test_utils.py
from twisted.trial import unittest
from twimp.utils import GeneratorWrapperProtocol, FrameSorter
from helpers import StringTransport


    def test_add_flush_add(self):
        fs = FrameSorter(self.frames_callback, [1, 2])

        fs.add(1, 0, 'a')
        fs.add(2, 1, 'b')
        fs.add(1, 2, 'a')

        self.assertEquals(self.frames, [(0, 1, 'a'),
                                        (1, 2, 'b')])
        self.clear()

        fs.flush()

        self.assertEquals(self.frames, [(2, 1, 'a')])
        self.clear()

        fs.add(2, 3, 'b')
        fs.add(1, 4, 'a')
        fs.add(2, 5, 'b')

        self.assertEquals(self.frames, [(3, 2, 'b'),
                                        (4, 1, 'a')])
        self.clear()

        fs.flush()

        self.assertEquals(self.frames, [(5, 2, 'b')])

    def test_not_sorted_input(self):

        fs = FrameSorter(self.frames_callback, [1, 2])