Use msgpack's streaming feed feature to build up a set of lists.
        
        The lists should then contain the messagepack-rpc specified items.

        This should be outrageously fast.
358448
def data(self, data):
        
        self.unpacker.feed(data)
        for msg in self.unpacker:
            self.handlers[msg[0]](*msg)
1025784
def to_msgpack(options = {})
      options = { :root => Rabl.configuration.include_msgpack_root }.merge(options)
      result = to_dumpable(options)
      Rabl.configuration.msgpack_engine.pack(result)
    end
1187249
def incoming_messages(self) -> t.List[t.Tuple[float, bytes]]:
        
        approximate_messages = self._receive_buffer.qsize()
        messages = []
        for _ in range(approximate_messages):
            try:
                messages.append(self._receive_buffer.get_nowait())
            except queue.Empty:
                break
        return messages
554703
def feed(self, data):
        
        self.buffer += data
        while len(self.buffer) >= 6:
            self.next_packet()
1232254
@Override
    protected void write(List<Message> msgList) throws IOException {
        for (Message msg : msgList) {
            writer.writeTo(msg);

            String routingKey = normalizeRoutingKey(msg);

            DynamicCounter.increment(
                    MonitorConfig.builder("writtenMessages")
                            .withTag(TagKey.DATA_SOURCE, routingKey)
                            .build());
            ++writtenMessages;
            DynamicCounter.increment(
                    MonitorConfig.builder("writtenBytes")
                            .withTag(TagKey.DATA_SOURCE, routingKey)
                            .build(), msg.getPayload().length);
            writtenBytes += msg.getPayload().length;

            messageWrittenInRotation = true;
        }

        writer.sync();

        throughput.increment(msgList.size());
    }
670543
async def recv_multipart(self):
        
        if not self._sockets:
            return []

        results = []

        async def recv_and_store(socket):
            frames = await socket.recv_multipart()
            results.append((socket, frames))

        tasks = [
            asyncio.ensure_future(recv_and_store(socket), loop=self.loop)
            for socket in self._sockets
        ]

        try:
            await asyncio.wait(
                tasks,
                return_when=asyncio.FIRST_COMPLETED,
                loop=self.loop,
            )
        finally:
            for task in tasks:
                task.cancel()

        return results
1345897
def read(self, read_list, verbose=False, log=False, n_per_proc=None):
        ""
        ret = []
        self.prep_input(read_list)
        L = len(self.file_list)
        if L == 0:
            return ret

        logger.info("Beginning to run sparser.")
        output_file_list = []
        if log:
            log_name = 'sparser_run_%s.log' % _time_stamp()
            outbuf = open(log_name, 'wb')
        else:
            outbuf = None
        try:
            if self.n_proc == 1:
                for fpath in self.file_list:
                    outpath, _ = self.read_one(fpath, outbuf, verbose)
                    if outpath is not None:
                        output_file_list.append(outpath)
            else:
                if n_per_proc is None:
                    n_per_proc = max(1, min(1000, L//self.n_proc//2))
                pool = None
                try:
                    pool = Pool(self.n_proc)
                    if n_per_proc is not 1:
                        batches = [self.file_list[n*n_per_proc:(n+1)*n_per_proc]
                                   for n in range(L//n_per_proc + 1)]
                        out_lists_and_buffs = pool.map(self.read_some,
                                                       batches)
                    else:
                        out_files_and_buffs = pool.map(self.read_one,
                                                       self.file_list)
                        out_lists_and_buffs = [([out_files], buffs)
                                               for out_files, buffs
                                               in out_files_and_buffs]
                finally:
                    if pool is not None:
                        pool.close()
                        pool.join()
                for i, (out_list, buff) in enumerate(out_lists_and_buffs):
                    if out_list is not None:
                        output_file_list += out_list
                    if log:
                        outbuf.write(b'Log for producing output %d/%d.\n'
                                     % (i, len(out_lists_and_buffs)))
                        if buff is not None:
                            buff.seek(0)
                            outbuf.write(buff.read() + b'\n')
                        else:
                            outbuf.write(b'ERROR: no buffer was None. '
                                         b'No logs available.\n')
                        outbuf.flush()
        finally:
            if log:
                outbuf.close()
                if verbose:
                    logger.info("Sparser logs may be found at %s." %
                                log_name)
        ret = self.get_output(output_file_list)
        return ret
293290
def pack_list(fmt, data):
    '''
    '''
    func = struct.Struct(fmt).pack
    return create_string_buffer(''.join([func(x) for x in data]))
250501
protected static List<String> listDeliveryStreams() {
        ListDeliveryStreamsRequest listDeliveryStreamsRequest = new ListDeliveryStreamsRequest();
        ListDeliveryStreamsResult listDeliveryStreamsResult =
                firehoseClient.listDeliveryStreams(listDeliveryStreamsRequest);
        List<String> deliveryStreamNames = listDeliveryStreamsResult.getDeliveryStreamNames();
        while (listDeliveryStreamsResult.isHasMoreDeliveryStreams()) {
            if (deliveryStreamNames.size() > 0) {
                listDeliveryStreamsRequest.setExclusiveStartDeliveryStreamName(deliveryStreamNames.get(deliveryStreamNames.size() - 1));
            }

            listDeliveryStreamsResult = firehoseClient.listDeliveryStreams(listDeliveryStreamsRequest);
            deliveryStreamNames.addAll(listDeliveryStreamsResult.getDeliveryStreamNames());

        }
        return deliveryStreamNames;
    }
594903
def list(self, channels=None, server=""):
        """"""
        self.send_items('LIST', ','.join(always_iterable(channels)), server)
997792
def put_stream(self, rel_path, metadata=None, cb=None):
        ''' '''

        import Queue
        import time
        import threading

        md5 = metadata.get('md5', None) if metadata else None

        # Horrible, but doing it anyway because I can.
        acl = ('public-read' if metadata.get('public',
                                             False) else metadata.get('acl',
                                                                      'public-read')) if metadata else 'public-read'

        path = self._prefix(self._rename(rel_path))

        class ThreadUploader(threading.Thread):

            """Thread class for uploading a part to S3"""

            def __init__(self, n, queue):
                threading.Thread.__init__(self)
                self.n = n
                self.queue = queue

            def run(self):

                while True:
                    mp, part_number, buf = self.queue.get()
                    if mp is None:  # Signal to die
                        logger.debug(
                            "put_stream: Thread {} exiting".format(
                                self.n))
                        self.queue.task_done()
                        return
                    logger.debug(
                        "put_stream: Thread {}: processing part: {}".format(
                            self.n,
                            part_number))
                    t1 = time.time()
                    try:
                        mp.upload_part_from_file(buf, part_number)
                    finally:
                        self.queue.task_done()
                        t2 = time.time()
                        logger.debug("put_stream: Thread {}, part {}. time = {} rate =  {} b/s" .format(
                            self.n, part_number, round(t2 - t1, 3), round((float(buf.tell()) / (t2 - t1)), 2)))

        if metadata is None:
            metadata = {}

        if md5:
            metadata['md5'] = md5  # Multipart uploads don't use md5 for etag

        # Some libraries, including apparently, boto, have troubles properly calcing the hash when
        # metadata values are null, empty, 0, etc.
        # https://forums.aws.amazon.com/thread.jspa?threadID=117580
        for k, v in metadata.items():
            if not v:
                del metadata[k]

        this = self

        buffer_size = 50 * 1024 * 1024  # Min part size is 5MB
        num_threads = 4
        thread_upload_queue = Queue.Queue(maxsize=100)

        for i in range(num_threads):
            t = ThreadUploader(i, thread_upload_queue)
            t.setDaemon(True)
            t.start()

        class flo:

            '''Object that is returned to the caller, for the caller to issue
            write() or writeline() calls on '''

            def __init__(self, rel_path):
                import io

                self.mp = this.bucket.initiate_multipart_upload(
                    path,
                    metadata=metadata)
                self.part_number = 1
                self.buffer = io.BytesIO()
                self.total_size = 0

                self.rel_path = rel_path

            def _send_buffer(self):
                '''Schedules a buffer to be sent in a thread by queuing it'''
                logger.debug(
                    "_send_buffer: sending part {} to thread pool size: {}, total_size = {}" .format(
                        self.part_number,
                        self.buffer.tell(),
                        self.total_size))
                self.buffer.seek(0)
                thread_upload_queue.put(
                    (self.mp, self.part_number, self.buffer))

            def write(self, d):
                import io

                self.buffer.write(d)  # Load the requested data into a buffer
                self.total_size += len(d)
                # After the buffer is large enough, send it, then create a new
                # buffer.
                if self.buffer.tell() > buffer_size:
                    self._send_buffer()

                    self.part_number += 1
                    self.buffer = io.BytesIO()

            def writelines(self, lines):
                raise NotImplemented()

            def close(self):

                if self.buffer.tell() > 0:
                    self._send_buffer()

                # Wait for all of th upload to complete
                thread_upload_queue.join()

                for i in range(num_threads):
                    thread_upload_queue.put(
                        (None, None, None))  # Tell all of the threads to die

                # Wait for all of the threads to exit
                thread_upload_queue.join()

                # Multi-part uploads throw a 400 Bad Request if they are
                # completed without writing any data.
                if self.total_size > 0:
                    self.mp.complete_upload()

                    this.bucket.set_acl(acl, path)

                this.put_metadata(self.rel_path, metadata)

            def __enter__(self):
                return self

            def __exit__(self, type_, value, traceback):
                if type_:
                    return False

                self.close()

        return flo(rel_path)
519427
function getListItems(listKey, from, to, callback) {
	this.redisClient.lrange(listKey, from, to, function(err, itemBytesArray) {
		if (err) { return callback(err, null) }
		if (!itemBytesArray) { return callback(null, []) }
		for (var items=[], i=0; i<itemBytesArray.length; i++) {
			items.push(itemBytesArray[i].toString())
		}
		callback(null, items)
	})
}
434475
def _message_generator(self):
        """"""
        while self._next_partition_records or self._completed_fetches:

            if not self._next_partition_records:
                completion = self._completed_fetches.popleft()
                self._next_partition_records = self._parse_fetched_data(completion)
                continue

            # Send additional FetchRequests when the internal queue is low
            # this should enable moderate pipelining
            if len(self._completed_fetches) <= self.config['iterator_refetch_records']:
                self.send_fetches()

            tp = self._next_partition_records.topic_partition

            # We can ignore any prior signal to drop pending message sets
            # because we are starting from a fresh one where fetch_offset == position
            # i.e., the user seek()'d to this position
            self._subscriptions.assignment[tp].drop_pending_message_set = False

            for msg in self._next_partition_records.take():

                # Because we are in a generator, it is possible for
                # subscription state to change between yield calls
                # so we need to re-check on each loop
                # this should catch assignment changes, pauses
                # and resets via seek_to_beginning / seek_to_end
                if not self._subscriptions.is_fetchable(tp):
                    log.debug("Not returning fetched records for partition %s"
                              " since it is no longer fetchable", tp)
                    self._next_partition_records = None
                    break

                # If there is a seek during message iteration,
                # we should stop unpacking this message set and
                # wait for a new fetch response that aligns with the
                # new seek position
                elif self._subscriptions.assignment[tp].drop_pending_message_set:
                    log.debug("Skipping remainder of message set for partition %s", tp)
                    self._subscriptions.assignment[tp].drop_pending_message_set = False
                    self._next_partition_records = None
                    break

                # Compressed messagesets may include earlier messages
                elif msg.offset < self._subscriptions.assignment[tp].position:
                    log.debug("Skipping message offset: %s (expecting %s)",
                              msg.offset,
                              self._subscriptions.assignment[tp].position)
                    continue

                self._subscriptions.assignment[tp].position = msg.offset + 1
                yield msg

            self._next_partition_records = None
545648
def process_incoming(self):
        

        model = self.model
        sockets = self.sockets

        # Check for new messages
        if not sockets:
            return
        # unpack sockets
        poller = sockets['poller']
        rep = sockets['rep']
        # pull = sockets['pull']
        pub = sockets['pub']
        items = poller.poll(10)
        for sock, n in items:
            for i in range(n):

                A, metadata = recv_array(sock)
                var = None

                # bmi actions
                if "update" in metadata:
                    dt = float(metadata["update"])
                    model.update(dt)
                    metadata["dt"] = dt
                elif "get_var" in metadata:
                    name = metadata["get_var"]
                    # temporary implementation
                    if metadata.get("copy", False):
                        var = model.get_var(name).copy()
                    else:
                        var = model.get_var(name)
                    if var is None:
                        logger.warning("Get_var returns None for %s" % name)
                    metadata['name'] = name
                    # assert socket is req socket
                elif "get_var_count" in metadata:
                    # temporary implementation
                    n = model.get_var_count()
                    metadata['get_var_count'] = n
                    # assert socket is req socket
                elif "get_var_rank" in metadata:
                    # temporary implementation
                    var_name = metadata['get_var_rank']
                    n = model.get_var_rank(var_name)
                    metadata['get_var_rank'] = n
                    # assert socket is req socket
                elif "get_var_shape" in metadata:
                    # temporary implementation
                    var_name = metadata['get_var_shape']
                    n = model.get_var_shape(var_name)
                    metadata['get_var_shape'] = tuple([int(item) for item in n])
                    # assert socket is req socket
                elif "get_var_type" in metadata:
                    # temporary implementation
                    var_name = metadata['get_var_type']
                    n = model.get_var_type(var_name)
                    metadata['get_var_type'] = n
                    # assert socket is req socket
                elif "get_var_name" in metadata:
                    i = int(metadata["get_var_name"])
                    name = model.get_var_name(i)
                    metadata['get_var_name'] = name
                    # assert socket is req socket
                elif "set_current_time" in metadata:
                    t = float(metadata["set_current_time"])
                    model.set_current_time(t)
                elif "set_var" in metadata:
                    name = metadata["set_var"]
                    # logger.debug("setting variable %s", name)
                    model.set_var(name, A)
                    metadata["name"] = name  # !?
                elif "set_var_slice" in metadata:
                    name = metadata["set_var_slice"]
                    logger.debug("setting variable %s", name)
                    start = metadata["start"]
                    count = metadata["count"]
                    model.set_var_slice(name, start, count, A)
                    metadata["name"] = name  # !?
                elif "set_var_index" in metadata:
                    name = metadata["set_var_index"]
                    logger.debug("setting variable %s using index index", name)
                    index = metadata["index"]
                    # TODO: test if this is fast enough.
                    # Otherwise move to BMI ++ but that is
                    # a bit of a burden on implementers
                    var = model.get_var(name).copy()
                    var.flat[index] = A
                    model.set_var(name, var)
                    metadata["name"] = name  # !?
                elif "get_current_time" in metadata:
                    metadata["get_current_time"]
                    t = model.get_current_time()
                    metadata['get_current_time'] = t
                elif "get_time_step" in metadata:
                    metadata["get_time_step"]
                    dt = model.get_time_step()
                    metadata['get_time_step'] = dt
                elif "get_end_time" in metadata:
                    metadata["get_end_time"]
                    t = model.get_end_time()
                    metadata['get_end_time'] = t
                elif "get_start_time" in metadata:
                    metadata["get_start_time"]
                    t = model.get_start_time()
                    metadata['get_start_time'] = t
                    # assert socket is req socket
                # custom actions
                elif "remote" in metadata:
                    assert metadata["remote"] in {
                        "play", "stop", "pause", "rewind", "quit"}
                    model.state = metadata["remote"]
                elif "operator" in metadata:
                    # TODO: support same operators as MPI_ops here....,
                    # TODO: reduce before apply
                    # TODO: assert pull socket
                    pass
                    # S = tuple(slice(*x) for x in action['slice'])
                    # print(repr(arr[S]))
                    # if action['operator'] == 'setitem':
                    #     arr[S] = data
                    # elif action['operator'] == 'add':
                    #     arr[S] += data
                elif "initialize" in metadata:
                    config_file = metadata["initialize"]
                    model.initialize(str(config_file))
                elif "finalize" in metadata:
                    model.finalize()
                else:
                    logger.warn("got unknown message {} from socket {}".format(str(metadata), sock))
                if sock.socket_type == zmq.REP:
                    # reply
                    send_array(rep, var, metadata=metadata)
                # any getter requested through the pull socket?
                elif any(x.startswith("get_") for x in metadata) and sock.socket_type == zmq.PULL:
                    # return through the pub socket
                    send_array(pub, var, metadata=metadata)
1203541
def pack(self):
        
        headers = ''.join(
            ['%s:%s\n' % (f, v) for f, v in sorted(self.headers.items())]
        )
        stomp_message = "%s\n%s\n%s%s\n" % (self._cmd, headers, self.body, NULL)

#        import pprint
#        print "stomp_message: ", pprint.pprint(stomp_message)

        return stomp_message
1201579
def StreamInChunks(self, callback=None, finish_callback=None,
                       additional_headers=None):
        """"""
        self.StreamMedia(callback=callback, finish_callback=finish_callback,
                         additional_headers=additional_headers,
                         use_chunks=True)
1163401
def load(cls, bmessage):
        
        message_dict = msgpack.loads(bmessage, encoding="utf8", use_list=False)

        try:
            obj = MessageMeta._registered_messages[message_dict["type"]].__new__(MessageMeta._registered_messages[message_dict["type"]])
            object.__setattr__(obj, "__dict__", message_dict)
        except:
            raise TypeError("Unknown message type") from None

        if not obj._verify():  # pylint: disable=protected-access
            raise TypeError("Invalid message content")

        return obj