Class: Ticketing::Conn::Session

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

Overview

In-flight requests per (op, key), FIFO — matches server reply order. (op, key)별 FIFO 대기 — 서버 응답 순서와 일치.

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize ⇒ Session

Returns a new instance of Session.



295
296
297
298
299
300
301
 
# File 'lib/ticketing/connection.rb', line 295

def initialize
  @queue = SizedQueue.new(REQ_QUEUE)
  @closed = false
  @mutex = Mutex.new
  @acquire = {}
  @release = {}
end

Instance Attribute Details

#closed ⇒ Object

Returns the value of attribute closed.



293
294
295
 
# File 'lib/ticketing/connection.rb', line 293

def closed
  @closed
end

#queue ⇒ Object (readonly)

Returns the value of attribute queue.



292
293
294
 
# File 'lib/ticketing/connection.rb', line 292

def queue
  @queue
end

Instance Method Details

#abort_queued ⇒ Object 



333
334
335
336
337
338
339
340
341
342
343
344
 
# File 'lib/ticketing/connection.rb', line 333

def abort_queued
  loop do
    item = begin
      @queue.pop(true)
    rescue ThreadError
      break
    end
    break if item.nil?

    item[3] << TicketError.aborted
  end
end

#fail_all ⇒ Object

Abort every in-flight request so the broker retries elsewhere. 진행 중 요청 전부 중단 → 브로커가 다른 노드로 재시도.



323
324
325
326
327
328
329
330
331
 
# File 'lib/ticketing/connection.rb', line 323

def fail_all
  victims = @mutex.synchronize do
    all = (@acquire.values + @release.values).flatten
    @acquire.clear
    @release.clear
    all
  end
  victims.each { |resolver| resolver << TicketError.aborted }
end

#map(op) ⇒ Object 



303
 
# File 'lib/ticketing/connection.rb', line 303

def map(op) = op == Protocol::OP_ACQUIRE ? @acquire : @release

#register(op, key, resolver) ⇒ Object 



305
306
307
 
# File 'lib/ticketing/connection.rb', line 305

def register(op, key, resolver)
  @mutex.synchronize { (map(op)[key] ||= []) << resolver }
end

#resolve(op, key, reply) ⇒ Object 



309
310
311
312
313
314
315
316
317
318
319
 
# File 'lib/ticketing/connection.rb', line 309

def resolve(op, key, reply)
  resolver = @mutex.synchronize do
    pending = map(op)[key]
    next nil if pending.nil? || pending.empty?

    first = pending.shift
    map(op).delete(key) if pending.empty?
    first
  end
  resolver << reply unless resolver.nil?
end