def get_messages(self, count=<NUM_LIT:1>, block=True, timeout=<NUM_LIT:0.1>):
messages = []<EOL>if timeout is not None:<EOL><INDENT>timeout += time.time()<EOL><DEDENT>new_offsets = {}<EOL>log.debug('<STR_LIT>', count)<EOL>while len(messages) < count:<EOL><INDENT>block_time = timeout - time.time()<EOL>log.debug('<STR_LIT>', block, block_time)<EOL>block_next_call = block is True or block > len(messages)<EOL>result = self._get_message(block_next_call, block_time,<EOL>get_partition_info=True,<EOL>update_offset=False)<EOL>log.debug('<STR_LIT>', result)<EOL>if not result:<EOL><INDENT>if block_next_call and (timeout is None or time.time() <= timeout):<EOL><INDENT>continue<EOL><DEDENT>break<EOL><DEDENT>partition, message = result<EOL>_msg = (partition, message) if self.partition_info else message<EOL>messages.append(_msg)<EOL>new_offsets[partition] = message.offset + <NUM_LIT:1><EOL><DEDENT>self.offsets.update(new_offsets)<EOL>self.count_since_commit += len(messages)<EOL>self._auto_commit()<EOL>log.debug('<STR_LIT>', len(messages), messages)<EOL>return messages<EOL>
Fetch the specified number of messages

Keyword Arguments:
    count: Indicates the maximum number of messages to be fetched
    block: If True, the API will block till all messages are fetched.
        If block is a positive integer the API will block until that
        many messages are fetched.
    timeout: When blocking is requested the function will block for
        the specified time (in seconds) until count messages is
        fetched. If None, it will block forever.