def _schedule(self, batch):
        """
        Row - portion of the queue for each partition id created at some point in time
        Row Key - partition id + score interval + timestamp
        Column Qualifier - discrete score (first three digits after dot, e.g. 0.001_0.002, 0.002_0.003, ...)
        Value - QueueCell msgpack blob

        Where score is mapped from 0.0 to 1.0
        score intervals are
          [0.01-0.02)
          [0.02-0.03)
          [0.03-0.04)
         ...
          [0.99-1.00]
        timestamp - the time when links was scheduled for retrieval.

        :param batch: list of tuples(score, fingerprint, domain, url)
        :return:
        """
        def get_interval(score, resolution):
            if score < 0.0 or score > 1.0:
                raise OverflowError

            i = int(score / resolution)
            if i % 10 == 0 and i > 0:
                i = i - 1  # last interval is inclusive from right
            return (i * resolution, (i + 1) * resolution)

        timestamp = int(time() * 1E+6)
        data = dict()
        for score, fingerprint, domain, url in batch:
            if type(domain) == dict:
                partition_id = self.partitioner.partition(domain['name'], self.partitions)
                host_crc32 = get_crc32(domain['name'])
            elif type(domain) == int:
                partition_id = self.partitioner.partition_by_hash(domain, self.partitions)
                host_crc32 = domain
            else:
                raise TypeError("domain of unknown type.")
            item = (unhexlify(fingerprint), host_crc32, url, score)
            score = 1 - score  # because of lexicographical sort in HBase
            rk = "%d_%s_%d" % (partition_id, "%0.2f_%0.2f" % get_interval(score, 0.01), timestamp)
            data.setdefault(rk, []).append((score, item))

        table = self.connection.table(self.table_name)
        with table.batch(transaction=True) as b:
            for rk, tuples in data.iteritems():
                obj = dict()
                for score, item in tuples:
                    column = 'f:%0.3f_%0.3f' % get_interval(score, 0.001)
                    obj.setdefault(column, []).append(item)

                final = dict()
                packer = Packer()
                for column, items in obj.iteritems():
                    stream = BytesIO()
                    for item in items:
                        stream.write(packer.pack(item))
                    final[column] = stream.getvalue()
                b.put(rk, final)

    def __init__(self, mod_conf, pub_endpoint, serialize_to):
        from zmq import Context, PUB

        BaseModule.__init__(self, mod_conf)
        self.pub_endpoint = pub_endpoint
        self.serialize_to = serialize_to
        logger.info("[Zmq Broker] Binding to endpoint " + self.pub_endpoint)

        # This doesn't work properly in init()
        # sometimes it ends up beings called several
        # times and the address becomes already in use.
        self.context = Context()
        self.s_pub = self.context.socket(PUB)
        self.s_pub.bind(self.pub_endpoint)

        # Load the correct serialization function
        # depending on the serialization method
        # chosen in the configuration.
        if self.serialize_to == "msgpack":
            from msgpack import Packer

            packer = Packer(default=encode_monitoring_data)
            self.serialize = lambda msg: packer.pack(msg)
        elif self.serialize_to == "json":
            self.serialize = lambda msg: json.dumps(msg, cls=SetEncoder)
        else:
            raise Exception("[Zmq Broker] No valid serialization method defined (Got " + str(self.serialize_to) + ")!")

def main(name):
   global socket_map, gps_lock, font, caution_written
   socket_map = generate_map(name)
   gps_lock = Lock()

   t1 = Thread(target = cpuavg)
   t1.daemon = True
   t1.start()

   t2 = Thread(target = pmreader)
   t2.daemon = True
   t2.start()

   t3 = Thread(target = gps)
   t3.daemon = True
   t3.start()

   socket = generate_map('aircomm_app')['out']
   packer = Packer(use_single_float = True)
   while True:
      try:
         data = [BCAST_NOFW, HEARTBEAT, int(voltage * 10), int(current * 10), int(load), mem_used(), critical]
         with gps_lock:
            try:
               if gps_data.fix >= 2:
                  data += [gps_data.lon, gps_data.lat]
            except:
               pass
         socket.send(packer.pack(data))
      except Exception, e:
         pass
      sleep(1.0)

def testPackUnicode():
    test_data = ["", "abcd", ["defgh"], "Русский текст"]
    for td in test_data:
        re = unpackb(packb(td), use_list=1, raw=False)
        assert re == td
        packer = Packer()
        data = packer.pack(td)
        re = Unpacker(BytesIO(data), raw=False, use_list=1).unpack()
        assert re == td

def testPackUnicode():
    test_data = ["", "abcd", ["defgh"], "Русский текст"]
    for td in test_data:
        re = unpackb(packb(td, encoding='utf-8'), use_list=1, encoding='utf-8')
        assert re == td
        packer = Packer(encoding='utf-8')
        data = packer.pack(td)
        re = Unpacker(BytesIO(data), encoding=str('utf-8'), use_list=1).unpack()
        assert re == td

def testPackUnicode():
    test_data = [six.u(""), six.u("abcd"), [six.u("defgh")], six.u("Русский текст")]
    for td in test_data:
        re = unpackb(packb(td, encoding="utf-8"), use_list=1, encoding="utf-8")
        assert_equal(re, td)
        packer = Packer(encoding="utf-8")
        data = packer.pack(td)
        re = Unpacker(BytesIO(data), encoding="utf-8", use_list=1).unpack()
        assert_equal(re, td)

def test_get_buffer():
    packer = Packer(autoreset=0, use_bin_type=True)
    packer.pack([1, 2])
    strm = BytesIO()
    strm.write(packer.getbuffer())
    written = strm.getvalue()

    expected = packb([1, 2], use_bin_type=True)
    assert written == expected

def testPackUnicode():
    test_data = [
        six.u(""), six.u("abcd"), [six.u("defgh")], six.u("Русский текст"),
        ]
    for td in test_data:
        re = unpackb(packb(td, encoding='utf-8'), use_list=1, encoding='utf-8')
        assert re == td
        packer = Packer(encoding='utf-8')
        data = packer.pack(td)
        re = Unpacker(BytesIO(data), encoding='utf-8', use_list=1).unpack()
        assert re == td

def testPackUnicode():
    test_data = [
        six.u(""), six.u("abcd"), (six.u("defgh"),), six.u("Русский текст"),
        ]
    for td in test_data:
        re = unpackb(packb(td, encoding='utf-8'), encoding='utf-8')
        assert_equal(re, td)
        packer = Packer(encoding='utf-8')
        data = packer.pack(td)
        re = Unpacker(BytesIO(data), encoding='utf-8').unpack()
        assert_equal(re, td)

def gen_segment(name, method_suffix, arg=None, append_arg_to_name=True):
    packer = Packer()
    if arg is None:
        getattr(packer, 'pack' + method_suffix)()
    else:
        getattr(packer, 'pack' + method_suffix)(arg)

    if append_arg_to_name:
        name = name + ' (' + str(arg) + ')'
    return OrderedDict(
        [('name', name), ('method_suffix', method_suffix), ('b64', packer.get_bytes().encode('base64', 'strict').replace('\n', ''))])

def testPackUnicode():
    test_data = [
        "", "abcd", ("defgh",), "Русский текст",
        ]
    for td in test_data:
        re = unpacks(packs(td, encoding='utf-8'), encoding='utf-8')
        assert_equal(re, td)
        packer = Packer(encoding='utf-8')
        data = packer.pack(td)
        re = Unpacker(BytesIO(data), encoding='utf-8').unpack()
        assert_equal(re, td)

def log_events():
    sock = ctx.socket(zmq.SUB)
    sock.bind("inproc://raw_events")
    sock.setsockopt(zmq.SUBSCRIBE, "")
    packer = Packer()
    with open('tweets.%d.msgpack' % int(time.time()), 'wb') as f:
        # intentionally writing the raw bytes and not parsing it here
        while True:
            msg = sock.recv()
            LOG.debug('event received: %d bytes', len(msg))
            f.write(packer.pack({'time' : int(time.time()), 'event' : msg}))
            f.flush()

def testArraySize(sizes=[0, 5, 50, 1000]):
    bio = BytesIO()
    packer = Packer()
    for size in sizes:
        bio.write(packer.pack_array_header(size))
        for i in range(size):
            bio.write(packer.pack(i))

    bio.seek(0)
    unpacker = Unpacker(bio, use_list=1)
    for size in sizes:
        assert unpacker.unpack() == list(range(size))

def testMapSize(sizes=[0, 5, 50, 1000]):
    bio = BytesIO()
    packer = Packer()
    for size in sizes:
        bio.write(packer.pack_map_header(size))
        for i in range(size):
            bio.write(packer.pack(i)) # key
            bio.write(packer.pack(i * 2)) # value

    bio.seek(0)
    unpacker = Unpacker(bio)
    for size in sizes:
        assert unpacker.unpack() == dict((i, i * 2) for i in range(size))

    def test_packer_unpacker(self):
        buf = BytesIO()
        packer = Packer()
        buf.write(packer.pack(1))
        buf.write(packer.pack('2'))
        buf.write(packer.pack({}))
        buf.seek(0)
        unpacker = Unpacker(buf)
        v1 = unpacker.unpack()
        self.assertEqual(1, v1)

        v2 = unpacker.unpack()
        self.assertEqual('2', v2)

        v3 = unpacker.unpack()
        self.assertTrue(isinstance(v3, dict))

 def __init__(self, event_loop):
     """Wrap `event_loop` on a msgpack-aware interface."""
     self.loop = event_loop
     self._packer = Packer(encoding='utf-8',
                           unicode_errors=unicode_errors_default)
     self._unpacker = Unpacker()
     self._message_cb = None

 def __init__(self, event_loop):
     """Wrap `event_loop` on a msgpack-aware interface."""
     self._event_loop = event_loop
     self._posted = deque()
     self._packer = Packer(use_bin_type=True)
     self._unpacker = Unpacker()
     self._message_cb = None
     self._stopped = False

def test_manualreset(sizes=[0, 5, 50, 1000]):
    packer = Packer(autoreset=False)
    for size in sizes:
        packer.pack_array_header(size)
        for i in range(size):
            packer.pack(i)

    bio = BytesIO(packer.bytes())
    unpacker = Unpacker(bio, use_list=1)
    for size in sizes:
        assert unpacker.unpack() == list(range(size))

    packer.reset()
    assert packer.bytes() == b''

def test_basic_segment(key):
    method_suffix = test_segment['suffix']
    read_bytes = test_segment['b64'].decode('base64', 'strict')
    unpacker = Unpacker(read_bytes)
    read_value = getattr(unpacker, 'unpack' + method_suffix)()
    logging.debug('[%s] read %s', key, read_value)
    packer = Packer()
    if read_value is None:
        getattr(packer, 'pack' + method_suffix)()
    else:
        getattr(packer, 'pack' + method_suffix)(read_value)

    write_bytes = packer.get_bytes()
    out_b64 = write_bytes.encode('base64', 'strict').replace('\n', '')

    if out_b64 != test_segment['b64']:
        compare_bytes(convert_bytes(write_bytes), convert_bytes(read_bytes))

    assert out_b64 == test_segment['b64']

class MsgpackStream(object):

    """Two-way msgpack stream that wraps a event loop byte stream.

    This wraps the event loop interface for reading/writing bytes and
    exposes an interface for reading/writing msgpack documents.
    """

    def __init__(self, event_loop):
        """Wrap `event_loop` on a msgpack-aware interface."""
        self.loop = event_loop
        self._packer = Packer(encoding='utf-8',
                              unicode_errors=unicode_errors_default)
        self._unpacker = Unpacker()
        self._message_cb = None

    def threadsafe_call(self, fn):
        """Wrapper around `BaseEventLoop.threadsafe_call`."""
        self.loop.threadsafe_call(fn)

    def send(self, msg):
        """Queue `msg` for sending to Nvim."""
        debug('sent %s', msg)
        self.loop.send(self._packer.pack(msg))

    def run(self, message_cb):
        """Run the event loop to receive messages from Nvim.

        While the event loop is running, `message_cb` will be called whenever
        a message has been successfully parsed from the input stream.
        """
        self._message_cb = message_cb
        self.loop.run(self._on_data)
        self._message_cb = None

    def stop(self):
        """Stop the event loop."""
        self.loop.stop()

    def close(self):
        """Close the event loop."""
        self.loop.close()

    def _on_data(self, data):
        self._unpacker.feed(data)
        while True:
            try:
                debug('waiting for message...')
                msg = next(self._unpacker)
                debug('received message: %s', msg)
                self._message_cb(msg)
            except StopIteration:
                debug('unpacker needs more data...')
                break