FIXME: fix Main loop instead
 Get and remove one message from the buffer.
 Return the message or nil.
336317
def message_from_buffer_nonblock
      return nil if @buffer.empty?
      ret = nil
      begin
        ret, size = Message.new.unmarshall_buffer(@buffer)
        @buffer.slice!(0, size)
      rescue IncompleteBufferException
        # fall through, let ret be null
      end
      ret
    end
238432
def help_now!(message=nil)
      exception = OptionParser::ParseError.new(message)
      class << exception
        def exit_code; 64; end
      end
      raise exception
    end
1252259
def error(message = nil, &block)
      logger.add(Logger::ERROR, message, self.logging_prefix, &block)
    end
712579
def end
      unless ready? then
        warn "attempt to close uninitialized stream; ignored."
        return nil
      end

      if in_stream? then
        warn "attempt to close unfinished zstream; reset forced"
        reset
      end

      reset_input

      err = Zlib.send @func_end, pointer

      Zlib.handle_error err, message

      @flags = 0

      # HACK this may be wrong
      @output = nil
      @next_out.free unless @next_out.nil?
      @next_out = nil

      nil
    end
450499
def <<(message)
      @mutex.lock
      begin
        if mailbox_full || @dead
          dead_letter(message)
          return
        end
        if message.is_a?(SystemEvent)
          # SystemEvents are high priority messages so they get added to the
          # head of our message queue instead of the end
          @messages.unshift message
        else
          @messages << message
        end

        @condition.signal
        nil
      ensure
        begin
          @mutex.unlock
        rescue
          nil
        end
      end
    end
911976
def remove_message(self, message):
        '''
        
        '''
        if message in self.__messages:
            self.__messages.remove(message)
1058024
def clear_messages(topic:, partition:)
      return unless @buffer.key?(topic) && @buffer[topic].key?(partition)

      @size -= @buffer[topic][partition].count
      @bytesize -= @buffer[topic][partition].map(&:bytesize).reduce(0, :+)

      @buffer[topic].delete(partition)
      @buffer.delete(topic) if @buffer[topic].empty?
    end
912599
def run
      # before blocking, empty the buffers
      # https://bugzilla.novell.com/show_bug.cgi?id=537401
      @buses.each_value do |b|
        while (m = b.message_queue.message_from_buffer_nonblock)
          b.process(m)
        end
      end
      while !@quitting && !@buses.empty?
        ready = IO.select(@buses.keys, [], [], 5) # timeout 5 seconds
        next unless ready # timeout exceeds so continue unless quitting
        ready.first.each do |socket|
          b = @buses[socket]
          begin
            b.message_queue.buffer_from_socket_nonblock
          rescue EOFError, SystemCallError
            @buses.delete socket # this bus died
            next
          end
          while (m = b.message_queue.message_from_buffer_nonblock)
            b.process(m)
          end
        end
      end
    end
1298964
def fatal(message = nil, ex = nil, data = nil, &block)
      log(FATAL, message, ex, data, block)
    end
285493
def debug(message = nil, &block)
      logger.add(Logger::DEBUG, message, self.logging_prefix, &block)
    end
345682
def warn(message = nil, ex = nil, data = nil, &block)
      log(WARN, message, ex, data, block)
    end
498647
def remove_message(self, message):
        
        import time
        _logger(self.__class__).log(5, 'removing message %s' % message)
        t = time.time()
        usd = message.block.userData()
        if usd:
            try:
                usd.messages.remove(message)
            except (AttributeError, ValueError):
                pass
        if message.decoration:
            self.editor.decorations.remove(message.decoration)
        self._messages.remove(message)
191352
def receive(self):
        '''
        
        '''
        while self.buffer.empty() and self.__up:
            # This sequence is skipped when the buffer is not empty.
            sleep_ms = random.randint(0, 1000)
            # log.debug('The message queue is empty, waiting %d miliseconds', sleep_ms)
            # disabled ^ as it was too noisy
            time.sleep(sleep_ms / 1000.0)
        if not self.buffer.empty():
            return self.buffer.get(block=False)
        return '', ''
636027
def format_system_message(errno):
    
    # first some flags used by FormatMessageW
    ALLOCATE_BUFFER = 0x100
    ARGUMENT_ARRAY = 0x2000
    FROM_HMODULE = 0x800
    FROM_STRING = 0x400
    FROM_SYSTEM = 0x1000
    IGNORE_INSERTS = 0x200

    # Let FormatMessageW allocate the buffer (we'll free it below)
    # Also, let it know we want a system error message.
    flags = ALLOCATE_BUFFER | FROM_SYSTEM
    source = None
    message_id = errno
    language_id = 0
    result_buffer = ctypes.wintypes.LPWSTR()
    buffer_size = 0
    arguments = None
    format_bytes = ctypes.windll.kernel32.FormatMessageW(
        flags,
        source,
        message_id,
        language_id,
        ctypes.byref(result_buffer),
        buffer_size,
        arguments,
    )
    # note the following will cause an infinite loop if GetLastError
    #  repeatedly returns an error that cannot be formatted, although
    #  this should not happen.
    handle_nonzero_success(format_bytes)
    message = result_buffer.value
    ctypes.windll.kernel32.LocalFree(result_buffer)
    return message
155885
def reprompt_error(self, message=None):
        
        try:
            session_id = session.sessionId
            self.session_machines.rollback_fsm(session_id)
            current_state = self.session_machines.current_state(session_id)
            if message is None:
                err_msg = choice(self._scenario_steps[current_state]['reprompt'])
            else:
                err_msg = message
            return question(err_msg)
        except UninitializedStateMachine as e:
            logger.error(e)
            return statement(INTERNAL_ERROR_MSG)
1168502
def error(message)
      length = (position > 0 ? Tokenizer.query(symbols[0..(position - 1)]).size + 1 : 0)
      raise "parsing query failed\n#{message}\n\n#{Tokenizer.query(symbols)}\n#{' ' * length}^"
    rescue
      logger.error $!
      raise $!
    end
202467
def process_messages_loop_internal(self):
        
        while self.receiving_messages:
            # connect to AMQP server and listen for 1 message then disconnect
            self.work_request = None
            self.connection.receive_loop_with_callback(self.queue_name, self.save_work_request_and_close)
            if self.work_request:
                self.process_work_request()