Class: Melaya::Events

Inherits:
Object
  • Object
show all
Defined in:
lib/melaya/events.rb

Overview

Platform real-time events via Socket.IO v4 at /api/v1/events.

Implements the Engine.IO v4 handshake and long-poll transport (packets batched with the 0x1e record separator; server pings answered with a POSTed pong) and the Socket.IO v4 packet framing, mirroring the TS events.ts client.

Room semantics (matching the server):

- run:<runId>          — pipeline run events
- project:<project>    — all events in a project (runs + CRUD)
- hitl:user:<userId>   — HITL approval events (joined automatically by server)

Thread safety: all public methods are safe to call from any thread. Callbacks are invoked on a dedicated background reader thread — do not block in a callback.

NOTE: Requires Ruby standard library 'net/http' with persistent HTTP/1.1. For production use the connection is kept alive between polls.

Examples:

events = Melaya::Events.new(api_key: ENV["MELAYA_API_KEY"])

# Subscribe to a run
unsub = events.on_run_update("run-123") { |e| puts e["event_type"] }

# Subscribe to HITL approvals
events.on_hitl_approval { |e| puts "HITL: #{e["type"]}" }

# Clean up
unsub.call          # remove one subscription
events.leave_run("run-123")
events.close

Constant Summary collapse

EIO_OPEN =

Engine.IO v4 packet type characters

"0"
EIO_CLOSE = 
"1"
EIO_PING = 
"2"
EIO_PONG = 
"3"
EIO_MESSAGE = 
"4"
EIO_UPGRADE = 
"5"
EIO_NOOP = 
"6"
SIO_CONNECT =

Socket.IO v4 packet types (numeric, follows the EIO_MESSAGE prefix)

0
SIO_DISCONNECT = 
1
SIO_EVENT = 
2
SIO_ACK = 
3
SIO_ERROR = 
4
RECONNECT_BASE_DELAY =

seconds — first reconnect wait

2
RECONNECT_MAX_DELAY =

seconds — cap for exponential backoff

30

Instance Method Summary collapse

Constructor Details

#initialize(api_key:, base_url: HttpClient::DEFAULT_BASE_URL, verify_ssl: true) ⇒ Events

Returns a new instance of Events.

Parameters:

  • api_key (String) —

    mk_* key

  • base_url (String) (defaults to: HttpClient::DEFAULT_BASE_URL)
  • verify_ssl (Boolean) (defaults to: true)

Raises:

  • (ArgumentError) 


68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
 
# File 'lib/melaya/events.rb', line 68

def initialize(api_key:, base_url: HttpClient::DEFAULT_BASE_URL, verify_ssl: true)
  raise ArgumentError, "Melaya: TLS certificate verification cannot be disabled." unless verify_ssl
  @_tok       = api_key.freeze
  @base_url   = base_url.sub(/\/$/, "")
  @verify_ssl = true

  @sid            = nil
  @closed         = false
  @listeners      = Hash.new { |h, k| h[k] = [] }  # event → [callable]
  @joined_rooms   = Set.new                        # rejoin-on-reconnect list
  @room_listeners = Hash.new { |h, k| h[k] = [] }  # room  → [[event, callable]]
  @mutex          = Mutex.new

  _connect_async
end

Instance Method Details

#close ⇒ Object

Close the Socket.IO connection and release all listeners.



169
170
171
172
173
174
175
176
177
 
# File 'lib/melaya/events.rb', line 169

def close
  @mutex.synchronize do
    @closed = true
    @joined_rooms.clear
    @room_listeners.clear
    @listeners.clear
  end
  @poll_thread&.kill
end

#leave_project(project) ⇒ Object

Leave a project room, remove its listeners, and drop it from the rejoin-on-reconnect list.

Parameters:

  • project (String) 


164
165
166
 
# File 'lib/melaya/events.rb', line 164

def leave_project(project)
  leave_room("project:#{project}")
end

#leave_run(run_id) ⇒ Object

Leave a run room, remove its listeners, and drop it from the rejoin-on-reconnect list.

Parameters:

  • run_id (String) 


157
158
159
 
# File 'lib/melaya/events.rb', line 157

def leave_run(run_id)
  leave_room("run:#{run_id}")
end

#on_hitl_approval {|Hash| ... } ⇒ Proc

Subscribe to HITL approval invalidation events. The server joins authenticated sockets to the hitl:user: room on connect — no explicit join needed.

Yields:

  • (Hash)

Returns:

  • (Proc) 


123
124
125
 
# File 'lib/melaya/events.rb', line 123

def on_hitl_approval(&block)
  on("pushHitlApprovals", &block)
end

#on_init_phase(run_id) {|Hash| ... } ⇒ Proc

Subscribe to init-phase progress events for a run.

Parameters:

  • run_id (String)

Yields:

  • (Hash)

Returns:

  • (Proc) 


102
103
104
105
 
# File 'lib/melaya/events.rb', line 102

def on_init_phase(run_id, &block)
  join_room("run:#{run_id}")
  on("pushInitPhase", room: "run:#{run_id}", &run_filter(run_id, block))
end

#on_pipeline_created(project) {|Hash| ... } ⇒ Proc

Subscribe to pipeline-created events within a project room.

Parameters:

  • project (String)

Yields:

  • (Hash)

Returns:

  • (Proc) 


131
132
133
134
 
# File 'lib/melaya/events.rb', line 131

def on_pipeline_created(project, &block)
  join_room("project:#{project}")
  on("pipelineCreated", room: "project:#{project}", &project_filter(project, block))
end

#on_pipeline_deleted(project) {|Hash| ... } ⇒ Proc

Subscribe to pipeline-deleted events within a project room.

Parameters:

  • project (String)

Yields:

  • (Hash)

Returns:

  • (Proc) 


149
150
151
152
 
# File 'lib/melaya/events.rb', line 149

def on_pipeline_deleted(project, &block)
  join_room("project:#{project}")
  on("pipelineDeleted", room: "project:#{project}", &project_filter(project, block))
end

#on_pipeline_updated(project) {|Hash| ... } ⇒ Proc

Subscribe to pipeline-updated events within a project room.

Parameters:

  • project (String)

Yields:

  • (Hash)

Returns:

  • (Proc) 


140
141
142
143
 
# File 'lib/melaya/events.rb', line 140

def on_pipeline_updated(project, &block)
  join_room("project:#{project}")
  on("pipelineUpdated", room: "project:#{project}", &project_filter(project, block))
end

#on_project_event(project) {|Hash| ... } ⇒ Proc

Subscribe to all events in a project (runs + pipeline CRUD). When the payload carries a +projectId+/+project+, events for other projects are filtered out.

Parameters:

  • project (String)

Yields:

  • (Hash)

Returns:

  • (Proc) 


113
114
115
116
 
# File 'lib/melaya/events.rb', line 113

def on_project_event(project, &block)
  join_room("project:#{project}")
  on("pushEvent", room: "project:#{project}", &project_filter(project, block))
end

#on_run_update(run_id) {|Hash| ... } ⇒ Proc

Subscribe to run events for a specific pipeline run. Joins the run: Socket.IO room automatically. When the payload carries a runId, events for other runs are filtered out so multiple per-run subscriptions never receive cross-room events.

Parameters:

  • run_id (String)

Yields:

  • (Hash) —

    run push event

Returns:

  • (Proc) —

    call to unsubscribe



93
94
95
96
 
# File 'lib/melaya/events.rb', line 93

def on_run_update(run_id, &block)
  join_room("run:#{run_id}")
  on("pushEvent", room: "run:#{run_id}", &run_filter(run_id, block))
end