Class: Arcp::Client

Inherits:
Object
  • Object
show all
Defined in:
lib/arcp/client.rb

Overview

ARCP client: opens a session over a transport, submits jobs, consumes events, and provides cursored job listings.

Examples:

Connect over an in-memory transport pair

server_t, client_t = Arcp::Transport::MemoryTransport.pair
Sync do
  client = Arcp::Client.open(
    transport: client_t,
    auth: { 'scheme' => 'bearer', 'token' => 'demo' }
  )
  handle = client.submit_job(agent: 'echo', input: { 'msg' => 'hi' })
  handle.subscribe(client: client).each { |ev| puts ev.kind }
  client.close
end

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(transport:, clock: Arcp::SystemClock.new) ⇒ Client

Returns a new instance of Client.



43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
 
# File 'lib/arcp/client.rb', line 43

def initialize(transport:, clock: Arcp::SystemClock.new)
  @transport = transport
  @clock = clock
  @session = nil
  @inbox = Async::Queue.new
  @pending = {}
  @job_streams = {}
  @job_results = {}
  @result_waiters = {}
  @reader_task = nil
  @heartbeat_task = nil
  @next_outbound_seq = 0
  @inbound_seq = 0
  @closed = false
  @mutex = Mutex.new
end

Instance Attribute Details

#session ⇒ Object (readonly)

Returns the value of attribute session.



33
34
35
 
# File 'lib/arcp/client.rb', line 33

def session
  @session
end

#transport ⇒ Object (readonly)

Returns the value of attribute transport.



33
34
35
 
# File 'lib/arcp/client.rb', line 33

def transport
  @transport
end

Class Method Details

.open(transport:, auth:, client_name: 'arcp-ruby', client_version: Arcp::VERSION, capabilities: nil, resume: nil, clock: Arcp::SystemClock.new) ⇒ Object 



35
36
37
38
39
40
41
 
# File 'lib/arcp/client.rb', line 35

def self.open(transport:, auth:, client_name: 'arcp-ruby', client_version: Arcp::VERSION,
              capabilities: nil, resume: nil, clock: Arcp::SystemClock.new)
  client = new(transport: transport, clock: clock)
  client.handshake!(auth: auth, client_name: client_name, client_version: client_version,
                    capabilities: capabilities, resume: resume)
  client
end

Instance Method Details

#ack(seq) ⇒ Object 



192
193
194
195
196
 
# File 'lib/arcp/client.rb', line 192

def ack(seq)
  require_feature!(Arcp::Session::Feature::ACK)
  send_envelope(type: Arcp::MessageTypes::SESSION_ACK,
                payload: Arcp::Session::Ack.new(last_processed_seq: seq).to_h)
end

#cancel_job(job_id:, reason: nil) ⇒ Object 



169
170
171
172
173
 
# File 'lib/arcp/client.rb', line 169

def cancel_job(job_id:, reason: nil)
  send_envelope(type: Arcp::MessageTypes::JOB_CANCEL,
                job_id: job_id,
                payload: Arcp::Job::Cancel.new(job_id: job_id, reason: reason).to_h)
end

#close(reason: nil) ⇒ Object 



211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
 
# File 'lib/arcp/client.rb', line 211

def close(reason: nil)
  return if @closed

  @closed = true
  begin
    send_envelope(type: Arcp::MessageTypes::SESSION_BYE,
                  payload: Arcp::Session::Bye.new(reason: reason).to_h)
  rescue StandardError
    nil
  end
  @heartbeat_task&.stop
  @reader_task&.stop
  @transport.close(reason: reason)
  drain_streams
  nil
end

#get_result(job_id:) ⇒ Object 



175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
 
# File 'lib/arcp/client.rb', line 175

def get_result(job_id:)
  env = @mutex.synchronize { @job_results[job_id] }
  if env.nil?
    queue = Async::Queue.new
    @mutex.synchronize { @result_waiters[job_id] = queue }
    env = queue.dequeue
  end
  case env.type
  when Arcp::MessageTypes::JOB_RESULT
    Arcp::Job::Result.from_h(env.payload)
  when Arcp::MessageTypes::JOB_ERROR
    raise Arcp::Job::JobError.from_h(env.payload).to_exception
  else
    raise Arcp::Errors::ProtocolViolation, "unexpected #{env.type}"
  end
end

#handshake!(auth:, client_name:, client_version:, capabilities: nil, resume: nil) ⇒ Object

Raises:



60
61
62
63
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
 
# File 'lib/arcp/client.rb', line 60

def handshake!(auth:, client_name:, client_version:, capabilities: nil, resume: nil)
  caps = capabilities || Arcp::Session::CapabilitySet.local
  session_id = Arcp::Ids.session_id
  hello = Arcp::Session::Hello.new(
    client_name: client_name, client_version: client_version,
    auth: auth, capabilities: caps, resume: resume
  )
  env = Arcp::Envelope.build(
    type: Arcp::MessageTypes::SESSION_HELLO,
    session_id: session_id,
    payload: hello.to_h
  )
  @transport.send(env)

  welcome_env = @transport.receive
  raise Arcp::Errors::ProtocolViolation, 'transport closed before welcome' if welcome_env.nil?

  case welcome_env.type
  when Arcp::MessageTypes::SESSION_WELCOME
    welcome = Arcp::Session::Welcome.from_h(welcome_env.payload)
    effective = caps.intersect(welcome.capabilities)
    @session = Arcp::Session::Info.new(
      id: welcome_env.session_id,
      runtime_version: welcome.runtime_version,
      capabilities: effective,
      agents: welcome.capabilities.agents,
      heartbeat_interval_sec: welcome.heartbeat_interval_sec,
      resume_token: welcome.resume_token,
      resume_window_sec: welcome.resume_window_sec
    )
  when Arcp::MessageTypes::SESSION_ERROR
    err = Arcp::Session::SessionError.from_h(welcome_env.payload)
    raise Arcp::Errors.for(err.code, message: err.message, details: err.details || {})
  else
    raise Arcp::Errors::ProtocolViolation, "expected session.welcome, got #{welcome_env.type}"
  end

  start_reader!
  if @session.supports?(Arcp::Session::Feature::HEARTBEAT) && @session.heartbeat_interval_sec
    start_heartbeat!
  end
  @session
end

#list_jobs(status: nil, agent: nil, created_after: nil, limit: nil, cursor: nil) ⇒ Object 



104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
 
# File 'lib/arcp/client.rb', line 104

def list_jobs(status: nil, agent: nil, created_after: nil, limit: nil, cursor: nil)
  require_feature!(Arcp::Session::Feature::LIST_JOBS)

  Enumerator.new do |yielder|
    next_cursor = cursor
    loop do
      payload = Arcp::Session::ListJobs.new(
        filter: { 'status' => status, 'agent' => agent, 'created_after' => created_after }.compact,
        limit: limit,
        cursor: next_cursor
      ).to_h
      response = request(type: Arcp::MessageTypes::SESSION_LIST_JOBS,
                         expect: Arcp::MessageTypes::SESSION_JOBS,
                         payload: payload)
      jobs = Arcp::Session::JobsResponse.from_h(response.payload)
      jobs.jobs.each { |j| yielder << Arcp::Job::Summary.from_h(j) }
      next_cursor = jobs.next_cursor
      break if next_cursor.nil?
    end
  end.lazy
end

#send_envelope(type:, payload:, job_id: nil) ⇒ Object

Raises:



198
199
200
201
202
203
204
205
206
207
208
209
 
# File 'lib/arcp/client.rb', line 198

def send_envelope(type:, payload:, job_id: nil)
  raise Arcp::Errors::Internal, 'session not open' unless @session
  raise IOError, 'client closed' if @closed

  env = Arcp::Envelope.build(
    type: type, session_id: @session.id,
    trace_id: Arcp::Trace.current.trace_id,
    job_id: job_id, payload: payload
  )
  @transport.send(env)
  env
end

#submit_job(agent:, input: nil, lease_request: nil, lease_constraints: nil, idempotency_key: nil, max_runtime_sec: nil) ⇒ Object 



126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
 
# File 'lib/arcp/client.rb', line 126

def submit_job(agent:, input: nil, lease_request: nil, lease_constraints: nil,
               idempotency_key: nil, max_runtime_sec: nil)
  lease_constraints&.validate!

  submit = Arcp::Job::Submit.new(
    agent: agent, input: input,
    lease_request: lease_request, lease_constraints: lease_constraints,
    idempotency_key: idempotency_key, max_runtime_sec: max_runtime_sec
  )
  accepted_env = request(
    type: Arcp::MessageTypes::JOB_SUBMIT,
    expect: Arcp::MessageTypes::JOB_ACCEPTED,
    payload: submit.to_h
  )
  accepted = Arcp::Job::Accepted.from_h(accepted_env.payload)
  Arcp::Job::Handle.new(
    job_id: accepted.job_id, agent: accepted.agent,
    submitted_at: accepted.accepted_at,
    lease: accepted.lease,
    credentials: accepted.credentials
  )
end

#subscribe_job(job_id:, from_event_seq: nil, history: false) ⇒ Object 



149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
 
# File 'lib/arcp/client.rb', line 149

def subscribe_job(job_id:, from_event_seq: nil, history: false)
  queue = @mutex.synchronize { @job_streams[job_id] ||= Async::Queue.new }

  if @session.supports?(Arcp::Session::Feature::SUBSCRIBE) && from_event_seq
    send_envelope(type: Arcp::MessageTypes::JOB_SUBSCRIBE,
                  job_id: job_id,
                  payload: Arcp::Job::Subscribe.new(job_id: job_id, from_event_seq: from_event_seq,
                                                    history: history).to_h)
  end

  Enumerator.new do |yielder|
    loop do
      item = queue.dequeue
      break if item.nil? || item == :__arcp_end__

      yielder << item
    end
  end
end