Class: Daytona::EventSubscriptionManager

Inherits:
Object
  • Object
show all
Defined in:
lib/daytona/common/event_subscription_manager.rb

Instance Method Summary collapse

Constructor Details

#initialize(dispatcher = nil) ⇒ EventSubscriptionManager

Returns a new instance of EventSubscriptionManager.



13
14
15
16
17
18
 
# File 'lib/daytona/common/event_subscription_manager.rb', line 13

def initialize(dispatcher = nil)
  @dispatcher = dispatcher
  @subscriptions = {}
  @mutex = Mutex.new
  @closed = false
end

Instance Method Details

#refresh(sub_id) ⇒ Object 



50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
 
# File 'lib/daytona/common/event_subscription_manager.rb', line 50

def refresh(sub_id)
  @mutex.synchronize do
    # Reject after shutdown to prevent use-after-close
    return false if @closed
  end

  timer_to_stop = nil

  @mutex.synchronize do
    return false unless @subscriptions.key?(sub_id)

    timer_to_stop = start_timer_locked(sub_id)
  end

  stop_thread(timer_to_stop)
  true
end

#shutdown ⇒ Object 



86
87
88
89
90
91
92
93
94
95
96
97
98
99
 
# File 'lib/daytona/common/event_subscription_manager.rb', line 86

def shutdown
  subscriptions = nil

  @mutex.synchronize do
    @closed = true
    subscriptions = @subscriptions.values
    @subscriptions = {}
  end

  subscriptions.each do |subscription|
    stop_thread(subscription[:timer])
    subscription[:unsubscribe].call
  end
end

#subscribe(resource_id:, handler:, events:) ⇒ Object 



20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
 
# File 'lib/daytona/common/event_subscription_manager.rb', line 20

def subscribe(resource_id:, handler:, events:)
  @mutex.synchronize do
    return nil if @closed || @dispatcher.nil?
  end

  unsubscribe = @dispatcher.subscribe(resource_id, events:, &handler)
  sub_id = SecureRandom.hex(16)
  timer_to_stop = nil
  rollback_unsubscribe = nil

  @mutex.synchronize do
    if @closed
      # Rollback dispatcher subscription on failure
      rollback_unsubscribe = unsubscribe
      next
    end

    @subscriptions[sub_id] = { unsubscribe:, timer: nil }
    timer_to_stop = start_timer_locked(sub_id)
  end

  if rollback_unsubscribe
    rollback_unsubscribe.call
    return nil
  end

  stop_thread(timer_to_stop)
  sub_id
end

#unsubscribe(sub_id) ⇒ Object 



68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
 
# File 'lib/daytona/common/event_subscription_manager.rb', line 68

def unsubscribe(sub_id)
  subscription = nil
  timer = nil

  @mutex.synchronize do
    subscription = @subscriptions.delete(sub_id)
    if subscription
      timer = subscription[:timer]
      subscription[:timer] = nil
    end
  end

  return unless subscription

  stop_thread(timer)
  subscription[:unsubscribe].call
end