Class: DiscordRDA::RequestQueue

Inherits:
Object
  • Object
show all
Defined in:
lib/discord_rda/connection/request_queue.rb

Overview

Production-ready Request Queue with proper Async integration. Handles Discord API requests with automatic rate limiting and retry logic.

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(rest, url:, identifier:, delete_queue_delay: 60_000, request_timeout: 30) ⇒ RequestQueue

Initialize request queue

Parameters:

  • rest (RestClient) —

    REST client

  • url (String) —

    URL identifier

  • identifier (String) —

    Queue identifier

  • delete_queue_delay (Integer) (defaults to: 60_000) —

    Delay before deleting empty queue (ms)

  • request_timeout (Integer) (defaults to: 30) —

    Request timeout in seconds



41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
 
# File 'lib/discord_rda/connection/request_queue.rb', line 41

def initialize(rest, url:, identifier:, delete_queue_delay: 60_000, request_timeout: 30)
  @rest = rest
  @url = url
  @identifier = identifier
  @delete_queue_delay = delete_queue_delay
  @request_timeout = request_timeout
  @pending = []
  @max = 1
  @remaining = 1
  @interval = 0
  @frozen_at = nil
  @processing = false
  @reset_timer = nil
  @delete_timeout = nil
  @mutex = Mutex.new
  @first_request = true
  @retry_count = Hash.new(0)
  @max_retries = 3
end

Instance Attribute Details

#delete_queue_delay ⇒ Integer (readonly)

Returns Delay before deleting empty queue (ms).

Returns:

  • (Integer) —

    Delay before deleting empty queue (ms)



30
31
32
 
# File 'lib/discord_rda/connection/request_queue.rb', line 30

def delete_queue_delay
  @delete_queue_delay
end

#identifier ⇒ String

Returns Queue identifier (token prefix).

Returns:

  • (String) —

    Queue identifier (token prefix)



15
16
17
 
# File 'lib/discord_rda/connection/request_queue.rb', line 15

def identifier
  @identifier
end

#interval ⇒ Integer (readonly)

Returns Time window in milliseconds.

Returns:

  • (Integer) —

    Time window in milliseconds



27
28
29
 
# File 'lib/discord_rda/connection/request_queue.rb', line 27

def interval
  @interval
end

#max ⇒ Integer (readonly)

Returns Maximum requests per interval.

Returns:

  • (Integer) —

    Maximum requests per interval



21
22
23
 
# File 'lib/discord_rda/connection/request_queue.rb', line 21

def max
  @max
end

#pending ⇒ Array<Hash> (readonly)

Returns Pending requests.

Returns:

  • (Array<Hash>) —

    Pending requests



18
19
20
 
# File 'lib/discord_rda/connection/request_queue.rb', line 18

def pending
  @pending
end

#remaining ⇒ Integer (readonly)

Returns Requests remaining in current window.

Returns:

  • (Integer) —

    Requests remaining in current window



24
25
26
 
# File 'lib/discord_rda/connection/request_queue.rb', line 24

def remaining
  @remaining
end

#request_timeout ⇒ Integer (readonly)

Returns Request timeout in seconds.

Returns:

  • (Integer) —

    Request timeout in seconds



33
34
35
 
# File 'lib/discord_rda/connection/request_queue.rb', line 33

def request_timeout
  @request_timeout
end

#rest ⇒ RestClient (readonly)

Returns REST client.

Returns:



9
10
11
 
# File 'lib/discord_rda/connection/request_queue.rb', line 9

def rest
  @rest
end

#url ⇒ String (readonly)

Returns Queue URL identifier.

Returns:

  • (String) —

    Queue URL identifier



12
13
14
 
# File 'lib/discord_rda/connection/request_queue.rb', line 12

def url
  @url
end

Instance Method Details

#calculate_wait_time ⇒ Float

Calculate wait time for rate limit

Returns:

  • (Float) —

    Seconds to wait



263
264
265
266
267
268
269
270
271
 
# File 'lib/discord_rda/connection/request_queue.rb', line 263

def calculate_wait_time
  @mutex.synchronize do
    return 0 unless @frozen_at

    now = Time.now.to_f * 1000
    future = @frozen_at + @interval
    [(future - now) / 1000.0, 1000].max / 1000.0
  end
end

#cleanup ⇒ void

This method returns an undefined value.

Clean up queue if empty



275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
 
# File 'lib/discord_rda/connection/request_queue.rb', line 275

def cleanup
  return unless clearable?

  @rest.logger&.debug("Queue #{@url} scheduling cleanup in #{@delete_queue_delay}ms")

  @delete_timeout&.stop rescue nil

  @delete_timeout = Async do |task|
    task.sleep(@delete_queue_delay / 1000.0)

    unless clearable?
      @rest.logger&.debug("Queue #{@url} no longer clearable, restarting processing")
      schedule_processing
      return
    end

    @rest.logger&.debug("Queue #{@url} deleting")
    @rest.queues.delete("#{@identifier}#{@url}")
  end
end

#clearable? ⇒ Boolean

Check if queue can be cleared

Returns:

  • (Boolean) —

    True if queue can be deleted



298
299
300
301
302
 
# File 'lib/discord_rda/connection/request_queue.rb', line 298

def clearable?
  @mutex.synchronize do
    @pending.empty? && !@processing
  end
end

#handle_completed_request(headers) ⇒ void

This method returns an undefined value.

Handle completed request response (update rate limit info)

Parameters:

  • headers (Hash) —

    Response headers



208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
 
# File 'lib/discord_rda/connection/request_queue.rb', line 208

def handle_completed_request(headers)
  @mutex.synchronize do
    if headers[:max] == 0
      @remaining += 1
      return
    end

    @frozen_at ||= Time.now.to_f * 1000
    @interval = headers[:interval] if headers[:interval]
    @remaining = headers[:remaining] if headers[:remaining]

    if @remaining <= 1 && headers[:interval]
      schedule_reset(headers[:interval])
    end
  end
end

#make_request(request) ⇒ void

This method returns an undefined value.

Add a request to the queue and process

Parameters:

  • request (Hash) —

    Request options



64
65
66
67
68
69
70
71
 
# File 'lib/discord_rda/connection/request_queue.rb', line 64

def make_request(request)
  wait_until_request_available

  @mutex.synchronize do
    @pending << request
    schedule_processing unless @processing
  end
end

#process_pending ⇒ void

This method returns an undefined value.

Process pending requests in the queue with proper async



81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
 
# File 'lib/discord_rda/connection/request_queue.rb', line 81

def process_pending
  return if @processing || @pending.empty?

  @mutex.synchronize { @processing = true }

  loop do
    break if @pending.empty?

    @rest.logger&.debug("Queue #{@url} processing #{@pending.length} pending requests")

    # Check if we can make a request
    unless @first_request || request_allowed?
      wait_time = calculate_wait_time
      if wait_time > 0
        Async { |t| t.sleep(wait_time) }
        next
      end
    end

    request = @mutex.synchronize { @pending.first }
    break unless request

    # Check rate limits before sending
    basic_url = @rest.simplify_url(request[:route], request[:method])

    # Check URL rate limits with async waiting
    url_reset = @rest.check_rate_limits(basic_url, @identifier)
    if url_reset && url_reset > 0
      Async { |t| t.sleep(url_reset / 1000.0) }
      next
    end

    # Check bucket rate limits
    if request[:bucket_id]
      bucket_reset = @rest.check_rate_limits(request[:bucket_id], @identifier)
      if bucket_reset && bucket_reset > 0
        Async { |t| t.sleep(bucket_reset / 1000.0) }
        next
      end
    end

    # Wait for invalid bucket with proper async
    unless @rest.invalid_bucket.request_allowed?
      @rest.invalid_bucket.wait_until_request_available
    end

    @first_request = false
    @remaining -= 1

    # Schedule reset if depleted
    if @remaining == 0 && @interval > 0
      schedule_reset
    end

    # Remove from queue and send
    @mutex.synchronize { @pending.shift }

    # Send request with timeout and retry logic
    send_with_retry(request)
  end

  @mutex.synchronize { @processing = false }
  cleanup
end

#request_allowed? ⇒ Boolean

Check if request is allowed

Returns:

  • (Boolean) —

    True if request can be made



227
228
229
230
231
232
233
234
235
 
# File 'lib/discord_rda/connection/request_queue.rb', line 227

def request_allowed?
  @mutex.synchronize do
    return true if @remaining > 0
    return true unless @frozen_at

    now = Time.now.to_f * 1000
    now >= (@frozen_at + @interval)
  end
end

#schedule_processing ⇒ void

This method returns an undefined value.

Schedule processing asynchronously



75
76
77
 
# File 'lib/discord_rda/connection/request_queue.rb', line 75

def schedule_processing
  Async { process_pending }
end

#send_with_retry(request) ⇒ void

This method returns an undefined value.

Send request with retry logic

Parameters:

  • request (Hash) —

    Request data



149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
 
# File 'lib/discord_rda/connection/request_queue.rb', line 149

def send_with_retry(request)
  request_id = request.object_id
  attempts = 0

  loop do
    attempts += 1

    begin
      # Send with timeout
      result = nil
      Async do |task|
        task.with_timeout(@request_timeout) do
          result = @rest.send_request(request)
        end
      rescue Async::TimeoutError
        raise TimeoutError, "Request timed out after #{@request_timeout}s"
      end

      # Success - reset retry count
      @retry_count.delete(request_id)
      request[:resolve]&.call(result) if request[:resolve]
      return result

    rescue RateLimitedError => e
      # Wait and retry
      wait_time = e.retry_after || 1.0
      @rest.logger&.warn('Rate limited, retrying', route: request[:route], wait: wait_time, attempt: attempts)
      Async { |t| t.sleep(wait_time) }
      next if attempts < @max_retries

      # Max retries reached
      request[:reject]&.call(error: e) if request[:reject]
      raise

    rescue ServerError => e
      # Retry server errors with backoff
      if attempts < @max_retries
        backoff = 2 ** attempts
        @rest.logger&.warn('Server error, retrying with backoff', route: request[:route], backoff: backoff, attempt: attempts)
        Async { |t| t.sleep(backoff) }
        next
      end

      request[:reject]&.call(error: e) if request[:reject]
      raise

    rescue => e
      # Other errors - don't retry
      @retry_count.delete(request_id)
      @rest.logger&.error("Queue #{@url} request failed", error: e, route: request[:route], attempt: attempts)
      request[:reject]&.call(error: e) if request[:reject]
      raise
    end
  end
end

#status ⇒ Hash

Get queue status

Returns:

  • (Hash) —

    Queue status



306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
 
# File 'lib/discord_rda/connection/request_queue.rb', line 306

def status
  @mutex.synchronize do
    {
      url: @url,
      identifier: @identifier,
      pending_count: @pending.length,
      processing: @processing,
      max: @max,
      remaining: @remaining,
      interval: @interval,
      frozen_at: @frozen_at,
      request_allowed: request_allowed?
    }
  end
end

#wait_until_request_available ⇒ void

This method returns an undefined value.

Wait until request is available with async support



239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
 
# File 'lib/discord_rda/connection/request_queue.rb', line 239

def wait_until_request_available
  @mutex.synchronize do
    return if @remaining > 0

    if @frozen_at
      now = Time.now.to_f * 1000
      future = @frozen_at + @interval
      wait_time = [(future - now) / 1000.0, 0].max

      if wait_time > 0
        @mutex.unlock
        if defined?(Async::Task) && Async::Task.current?
          Async::Task.current.sleep(wait_time)
        else
          sleep(wait_time)
        end
        @mutex.lock
      end
    end
  end
end