def buffer_flush(options={})
      force = options[:force] || options[:final]
      final = options[:final]

      # final flush will wait for lock, so we are sure to flush out all buffered events
      if options[:final]
        @buffer_state[:flush_mutex].lock
      elsif ! @buffer_state[:flush_mutex].try_lock # failed to get lock, another flush already in progress
        return 0
      end

      items_flushed = 0

      begin
        time_since_last_flush = (Time.now - @buffer_state[:last_flush])

        return 0 if @buffer_state[:pending_count] == 0
        return 0 if (!force) &&
           (@buffer_state[:pending_count] < @buffer_config[:max_items]) &&
           (time_since_last_flush < @buffer_config[:max_interval])

        @buffer_state[:pending_mutex].synchronize do
          @buffer_state[:outgoing_items] = @buffer_state[:pending_items]
          @buffer_state[:outgoing_count] = @buffer_state[:pending_count]
          buffer_clear_pending
        end

        @buffer_config[:logger].debug do
          debug_output = {
            :outgoing_count => @buffer_state[:outgoing_count],
            :time_since_last_flush => time_since_last_flush,
            :outgoing_events => @buffer_state[:outgoing_items],
            :batch_timeout => @buffer_config[:max_interval],
            :force => force,
            :final => final
          }
          "Flushing output: #{debug_output}"
        end if @buffer_config[:logger]

        @buffer_state[:outgoing_items].each do |group, events|
          begin
            if group.nil?
              flush(events,final)
            else
              flush(events, group, final)
            end

            @buffer_state[:outgoing_items].delete(group)
            events_size = events.size
            @buffer_state[:outgoing_count] -= events_size
            items_flushed += events_size
            @buffer_state[:last_flush] = Time.now

          rescue => e

            @buffer_config[:logger].warn do
              warn_output = {
                :outgoing_count => @buffer_state[:outgoing_count],
                :exception => e.class.name,
                :backtrace => e.backtrace
              }
              "Failed to flush outgoing items: #{warn_output}"
            end if @buffer_config[:logger]

            if @buffer_config[:has_on_flush_error]
              on_flush_error e
            end

            if @buffer_config[:drop_messages_on_flush_error]
              reset_buffer
            else
              cancel_flush
            end

          end
        end

      ensure
        @buffer_state[:flush_mutex].unlock
      end

      return items_flushed
    end
def buffer_flush(options={})
      force = options[:force] || options[:final]
      final = options[:final]

      # final flush will wait for lock, so we are sure to flush out all buffered events
      if options[:final]
        @buffer_state[:flush_mutex].lock
      elsif ! @buffer_state[:flush_mutex].try_lock # failed to get lock, another flush already in progress
        return 0
      end

      items_flushed = 0

      begin
        time_since_last_flush = (Time.now - @buffer_state[:last_flush])

        return 0 if @buffer_state[:pending_count] == 0
        return 0 if (!force) &&
           (@buffer_state[:pending_count] < @buffer_config[:max_items]) &&
           (time_since_last_flush < @buffer_config[:max_interval])

        @buffer_state[:pending_mutex].synchronize do
          @buffer_state[:outgoing_items] = @buffer_state[:pending_items]
          @buffer_state[:outgoing_count] = @buffer_state[:pending_count]
          buffer_clear_pending
        end

        @buffer_config[:logger].debug do
          debug_output = {
            :outgoing_count => @buffer_state[:outgoing_count],
            :time_since_last_flush => time_since_last_flush,
            :outgoing_events => @buffer_state[:outgoing_items],
            :batch_timeout => @buffer_config[:max_interval],
            :force => force,
            :final => final
          }
          "Flushing output: #{debug_output}"
        end if @buffer_config[:logger]

        @buffer_state[:outgoing_items].each do |group, events|
          begin
            if group.nil?
              flush(events,final)
            else
              flush(events, group, final)
            end

            @buffer_state[:outgoing_items].delete(group)
            events_size = events.size
            @buffer_state[:outgoing_count] -= events_size
            items_flushed += events_size
            @buffer_state[:last_flush] = Time.now

          rescue => e

            @buffer_config[:logger].warn do
              warn_output = {
                :outgoing_count => @buffer_state[:outgoing_count],
                :exception => e.class.name,
                :backtrace => e.backtrace
              }
              "Failed to flush outgoing items: #{warn_output}"
            end if @buffer_config[:logger]

            if @buffer_config[:has_on_flush_error]
              on_flush_error e
            end

            if @buffer_config[:drop_messages_on_flush_error]
              reset_buffer
            else
              cancel_flush
            end

          end
        end

      ensure
        @buffer_state[:flush_mutex].unlock
      end

      return items_flushed
    end
Try to flush events.

 Returns immediately if flushing is not necessary/possible at the moment:
 * :max_items have not been accumulated
 * :max_interval seconds have not elapased since the last flush
 * another flush is in progress

 <code>buffer_flush(:force => true)</code> will cause a flush to occur even
 if +:max_items+ or +:max_interval+ have not been reached. A forced flush
 will still return immediately (without flushing) if another flush is
 currently in progress.

 <code>buffer_flush(:final => true)</code> is identical to <code>buffer_flush(:force => true)</code>,
 except that if another flush is already in progress, <code>buffer_flush(:final => true)</code>
 will block/wait for the other flush to finish before proceeding.

 @param [Hash] options Optional. May be <code>{:force => true}</code> or <code>{:final => true}</code>.
 @return [Fixnum] The number of items successfully passed to +flush+.