Class: Daytona::EventDispatcher

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

Overview

Manages a Socket.IO connection and dispatches events to per-resource handlers. Generic — works for sandboxes, volumes, snapshots, runners, etc.

Constant Summary collapse

DISCONNECT_DELAY = 
30

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(api_url:, token:, organization_id: nil, source: nil, sdk_version: nil) ⇒ EventDispatcher

Returns a new instance of EventDispatcher.

Parameters:

  • api_url (String)
  • token (String)
  • organization_id (String, nil) (defaults to: nil) 


17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
 
# File 'lib/daytona/common/event_dispatcher.rb', line 17

def initialize(api_url:, token:, organization_id: nil, source: nil, sdk_version: nil)
  @api_url = api_url
  @token = token
  @organization_id = organization_id
  @source = source
  @sdk_version = sdk_version
  @client = nil
  @connected = false
  @failed = false
  @fail_error = nil
  @listeners = {}
  @registered_events = Set.new
  @mutex = Mutex.new
  @disconnect_timer = nil
  @disconnect_generation = 0
  @reconnect_thread = nil
  @connect_thread = nil
  @reconnecting = false
  @connecting = false
  @close_requested = false
  @closed = false
  @max_reconnects = 100
end

Instance Attribute Details

#fail_error ⇒ String? (readonly)

Returns:

  • (String, nil) 


153
154
155
 
# File 'lib/daytona/common/event_dispatcher.rb', line 153

def fail_error
  @fail_error
end

Instance Method Details

#connect ⇒ void

This method returns an undefined value.

Establish the Socket.IO connection.

Raises:

  • (StandardError) —

    on connection failure



64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
 
# File 'lib/daytona/common/event_dispatcher.rb', line 64

def connect
  @mutex.synchronize do
    return if @closed || @connected || @connecting

    @connecting = true
  end

  # Close any existing stale connection before creating a fresh one
  @client&.close rescue nil # rubocop:disable Style/RescueModifier

  client = SocketIOClient.new(
    api_url: @api_url,
    token: @token,
    organization_id: @organization_id,
    source: @source,
    sdk_version: @sdk_version,
    on_event: method(:handle_event),
    on_disconnect: method(:handle_disconnect)
  )

  @mutex.synchronize do
    @client = client
    @close_requested = false
  end

  client.connect

  @mutex.synchronize do
    @connected = true
    @failed = false
    @fail_error = nil
    schedule_delayed_disconnect_locked if @listeners.empty?
  end
rescue StandardError => e
  @mutex.synchronize do
    @failed = true
    @fail_error = "WebSocket connection failed: #{e.message}"
  end
ensure
  @mutex.synchronize { @connecting = false }
end

#connected? ⇒ Boolean

Returns:

  • (Boolean) 


143
144
145
 
# File 'lib/daytona/common/event_dispatcher.rb', line 143

def connected?
  @connected
end

#disconnect ⇒ Object

Disconnect and clean up.



156
157
158
159
160
161
162
 
# File 'lib/daytona/common/event_dispatcher.rb', line 156

def disconnect
  thread_state = @mutex.synchronize do
    begin_disconnect_locked(permanent: true, skip_thread: Thread.current)
  end

  finalize_disconnect(thread_state)
end

#ensure_connected ⇒ void

This method returns an undefined value.

Idempotent: ensure a connection attempt is in progress or already established. Non-blocking. Starts a background Thread to connect if not already connected and no attempt is currently running.



45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
 
# File 'lib/daytona/common/event_dispatcher.rb', line 45

def ensure_connected
  @mutex.synchronize do
    return if @closed || @connected || @connecting || @connect_thread&.alive?

    @connect_thread = Thread.new do
      connect
    rescue StandardError
      # Callers check connected? when they need it
    ensure
      @mutex.synchronize do
        @connect_thread = nil if @connect_thread == Thread.current
      end
    end
  end
end

#failed? ⇒ Boolean

Returns:

  • (Boolean) 


148
149
150
 
# File 'lib/daytona/common/event_dispatcher.rb', line 148

def failed?
  @failed
end

#subscribe(resource_id, events:) {|event_name, data| ... } ⇒ Proc

Subscribe to specific events for a resource.

Parameters:

  • resource_id (String) —

    The ID of the resource (e.g. sandbox ID, volume ID).

  • events (Array<String>) —

    List of Socket.IO event names to listen for.

Yields:

  • (event_name, data) —

    Called with raw event name and data hash.

Returns:

  • (Proc) —

    Unsubscribe function.



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
 
# File 'lib/daytona/common/event_dispatcher.rb', line 111

def subscribe(resource_id, events:, &handler)
  @mutex.synchronize do
    # No-op after disconnect
    return -> {} if @closed

    @disconnect_timer&.kill
    @disconnect_timer = nil
    @disconnect_generation += 1

    @listeners[resource_id] ||= []
    @listeners[resource_id] << handler
    # Mark events in filter under lock to avoid dropping first event
    events.each { |evt| @registered_events.add(evt) }
  end

  ensure_connected

  lambda {
    return if @close_requested || @closed

    should_schedule = false
    @mutex.synchronize do
      @listeners[resource_id]&.delete(handler)
      unsubscribe_resource_locked(resource_id) if @listeners[resource_id] && @listeners[resource_id].empty?
      should_schedule = @listeners.empty?

      schedule_delayed_disconnect_locked if should_schedule
    end
  }
end