Class: TCellAgent::QueueWithTimeout

Inherits:
Object
  • Object
show all
Defined in:
lib/tcell_agent/utils/queue_with_timeout.rb

Instance Method Summary collapse

Constructor Details

#initialize ⇒ QueueWithTimeout 



87
88
89
90
91
92
 
# File 'lib/tcell_agent/utils/queue_with_timeout.rb', line 87

def initialize
  @mutex = Mutex.new
  @queue = []
  @response_time_table = {}
  @recieved = ConditionVariable.new
end

Instance Method Details

#<<(x) ⇒ Object

rubocop:disable Naming/UncommunicativeMethodParamName



94
95
96
97
98
99
 
# File 'lib/tcell_agent/utils/queue_with_timeout.rb', line 94

def <<(x) # rubocop:disable Naming/UncommunicativeMethodParamName
  @mutex.synchronize do
    @queue << x
    @recieved.signal
  end
end

#add_response_time(route_id, response_time) ⇒ Object 



101
102
103
104
105
106
107
108
109
110
 
# File 'lib/tcell_agent/utils/queue_with_timeout.rb', line 101

def add_response_time(route_id, response_time)
  @mutex.synchronize do
    route_id = '?' if route_id.nil? || route_id == ''
    @response_time_table[route_id] = @response_time_table.fetch(route_id, {})
    @response_time_table[route_id]['c'] = @response_time_table[route_id].fetch('c', 0) + 1
    @response_time_table[route_id]['mx'] = [@response_time_table[route_id].fetch('mx', 0), response_time].max
    @response_time_table[route_id]['mn'] = [@response_time_table[route_id].fetch('mn', response_time), response_time].min
    @response_time_table[route_id]['t'] = ((@response_time_table[route_id].fetch('t', 0) * (@response_time_table[route_id]['c'] - 1)) + response_time) / @response_time_table[route_id]['c']
  end
end

#get_response_time_table ⇒ Object 



116
117
118
 
# File 'lib/tcell_agent/utils/queue_with_timeout.rb', line 116

def get_response_time_table
  @response_time_table
end

#length ⇒ Object 



112
113
114
 
# File 'lib/tcell_agent/utils/queue_with_timeout.rb', line 112

def length
  @queue.length
end

#pop(non_block = false) ⇒ Object 



126
127
128
 
# File 'lib/tcell_agent/utils/queue_with_timeout.rb', line 126

def pop(non_block = false)
  pop_with_timeout(non_block ? 0 : nil)
end

#pop_with_timeout(timeout = nil) ⇒ Object 



130
131
132
133
134
135
136
137
138
139
140
 
# File 'lib/tcell_agent/utils/queue_with_timeout.rb', line 130

def pop_with_timeout(timeout = nil)
  @mutex.synchronize do
    if @queue.empty?
      @recieved.wait(@mutex, timeout) if timeout != 0
      # if we're still empty after the timeout, raise exception
      return nil if @queue.empty?
      # raise ThreadError, "queue empty" if @queue.empty?
    end
    @queue.shift
  end
end

#reset_response_time_table ⇒ Object 



120
121
122
123
124
 
# File 'lib/tcell_agent/utils/queue_with_timeout.rb', line 120

def reset_response_time_table
  @mutex.synchronize do
    @response_time_table = {}
  end
end