def get_messages(self, count=<NUM_LIT:1>, block=True, timeout=<NUM_LIT:10>):
messages = []<EOL>self.size.value = count<EOL>self.events.pause.clear()<EOL>if timeout is not None:<EOL><INDENT>max_time = time.time() + timeout<EOL><DEDENT>new_offsets = {}<EOL>while count > <NUM_LIT:0> and (timeout is None or timeout > <NUM_LIT:0>):<EOL><INDENT>if self.queue.empty():<EOL><INDENT>self.events.start.set()<EOL><DEDENT>block_next_call = block is True or block > len(messages)<EOL>try:<EOL><INDENT>partition, message = self.queue.get(block_next_call,<EOL>timeout)<EOL><DEDENT>except queue.Empty:<EOL><INDENT>break<EOL><DEDENT>_msg = (partition, message) if self.partition_info else message<EOL>messages.append(_msg)<EOL>new_offsets[partition] = message.offset + <NUM_LIT:1><EOL>count -= <NUM_LIT:1><EOL>if timeout is not None:<EOL><INDENT>timeout = max_time - time.time()<EOL><DEDENT><DEDENT>self.size.value = <NUM_LIT:0><EOL>self.events.start.clear()<EOL>self.events.pause.set()<EOL>self.offsets.update(new_offsets)<EOL>self.count_since_commit += len(messages)<EOL>self._auto_commit()<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.