Class: Restate::Client

Inherits:
Object
  • Object
show all
Defined in:
lib/restate/client.rb,
sig/restate.rbs

Overview

HTTP client for invoking Restate services and managing the Restate runtime from outside the Restate runtime.

Examples:

Via global config (recommended)

Restate.configure do |c|
  c.ingress_url = "http://localhost:8080"
  c.admin_url = "http://localhost:9070"
end
client = Restate.client
result = client.service(Greeter).greet("World")

Standalone

client = Restate::Client.new(ingress_url: "http://localhost:8080",
                             admin_url: "http://localhost:9070")

Service invocation

client.service("Greeter").greet("World")
client.object("Counter", "my-key").add(5)
client.workflow("UserSignup", "user42").run("user@example.com")

Admin operations

client.resolve_awakeable(awakeable_id, "result")
client.reject_awakeable(awakeable_id, "failed")
client.cancel_invocation(invocation_id)
client.create_deployment("http://localhost:9080")

Instance Method Summary collapse

Constructor Details

#initialize(ingress_url: 'http://localhost:8080', admin_url: 'http://localhost:9070', ingress_headers: {}, admin_headers: {}) ⇒ Client

Returns a new instance of Client.

Parameters:

  • ingress_url: (String) (defaults to: 'http://localhost:8080')
  • admin_url: (String) (defaults to: 'http://localhost:9070')
  • ingress_headers: (Hash[String, String]) (defaults to: {})
  • admin_headers: (Hash[String, String]) (defaults to: {}) 


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

def initialize(ingress_url: 'http://localhost:8080', admin_url: 'http://localhost:9070',
               ingress_headers: {}, admin_headers: {})
  @ingress_url = ingress_url.chomp('/')
  @admin_url = admin_url.chomp('/')
  @ingress_headers = ingress_headers
  @admin_headers = admin_headers
end

Instance Method Details

#cancel_invocation(invocation_id) ⇒ void

This method returns an undefined value.

Cancel a running invocation.

Parameters:

  • invocation_id (String) 


75
76
77
 
# File 'lib/restate/client.rb', line 75

def cancel_invocation(invocation_id)
  post_admin("/restate/invocations/#{invocation_id}/cancel", nil)
end

#execute_query(sql) ⇒ Array<Hash>

Execute a SQL query against Restate's introspection API (DataFusion). The admin API exposes system tables (sys_invocation, sys_journal, state, etc.) that can be queried with standard SQL.

Examples:

client.execute_query("SELECT id, status FROM sys_invocation LIMIT 10")

Parameters:

  • sql (String) —

    a SQL query string

Returns:

  • (Array<Hash>) —

    rows returned by the query



95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
 
# File 'lib/restate/client.rb', line 95

def execute_query(sql) # rubocop:disable Metrics/AbcSize,Metrics/MethodLength
  uri = URI("#{@admin_url}/query")
  request = Net::HTTP::Post.new(uri)
  request['Content-Type'] = 'application/json'
  request['Accept'] = 'application/json'
  @admin_headers.each { |k, v| request[k] = v }
  request.body = JSON.generate({ query: sql })
  response = Net::HTTP.start(uri.hostname, uri.port, # steep:ignore ArgumentTypeMismatch
                             use_ssl: uri.scheme == 'https',
                             open_timeout: 5,
                             read_timeout: 30) { |http| http.request(request) }
  Kernel.raise "Restate query error: #{response.code} #{response.body}" unless response.is_a?(Net::HTTPSuccess)
  body = response.body
  body && !body.empty? ? (JSON.parse(body)['rows'] || []) : []
end

#kill_invocation(invocation_id) ⇒ void

This method returns an undefined value.

Kill a running invocation (immediate termination, no cleanup).

Parameters:

  • invocation_id (String) 


80
81
82
 
# File 'lib/restate/client.rb', line 80

def kill_invocation(invocation_id)
  post_admin("/restate/invocations/#{invocation_id}/kill", nil)
end

#object(service, key) ⇒ ClientServiceProxy

Returns a proxy for calling a keyed virtual object.

Parameters:

  • service (Object)
  • key (String)

Returns:

  • (ClientServiceProxy) 


50
51
52
 
# File 'lib/restate/client.rb', line 50

def object(service, key)
  ClientServiceProxy.new(@ingress_url, resolve_name(service), key, @ingress_headers)
end

#parse_response(response) ⇒ Object

Parameters:

  • response (Net::HTTPResponse)

Returns:

  • (Object) 


147
148
149
150
 
# File 'lib/restate/client.rb', line 147

def parse_response(response)
  body = response.body
  body && !body.empty? ? JSON.parse(body) : nil
end

#post_admin(path, body) ⇒ Object

rubocop:disable Metrics/AbcSize

Parameters:

  • path (String)
  • body (Object)

Returns:

  • (Object) 


134
135
136
137
138
139
140
141
142
143
144
145
 
# File 'lib/restate/client.rb', line 134

def post_admin(path, body) # rubocop:disable Metrics/AbcSize
  uri = URI("#{@admin_url}#{path}")
  request = Net::HTTP::Post.new(uri)
  request['Content-Type'] = 'application/json'
  @admin_headers.each { |k, v| request[k] = v }
  request.body = JSON.generate(body) if body
  response = Net::HTTP.start(uri.hostname, uri.port, # steep:ignore ArgumentTypeMismatch
                             use_ssl: uri.scheme == 'https',
                             read_timeout: 30) { |http| http.request(request) }
  Kernel.raise "Restate admin error: #{response.code} #{response.body}" unless response.is_a?(Net::HTTPSuccess)
  parse_response(response)
end

#post_ingress(path, body) ⇒ Object

rubocop:disable Metrics/AbcSize

Parameters:

  • path (String)
  • body (Object)

Returns:

  • (Object) 


121
122
123
124
125
126
127
128
129
130
131
132
 
# File 'lib/restate/client.rb', line 121

def post_ingress(path, body) # rubocop:disable Metrics/AbcSize
  uri = URI("#{@ingress_url}#{path}")
  request = Net::HTTP::Post.new(uri)
  request['Content-Type'] = 'application/json'
  @ingress_headers.each { |k, v| request[k] = v }
  request.body = JSON.generate(body) if body
  response = Net::HTTP.start(uri.hostname, uri.port, # steep:ignore ArgumentTypeMismatch
                             use_ssl: uri.scheme == 'https',
                             read_timeout: 30) { |http| http.request(request) }
  Kernel.raise "Restate ingress error: #{response.code} #{response.body}" unless response.is_a?(Net::HTTPSuccess)
  parse_response(response)
end

#reject_awakeable(awakeable_id, message, code: 500) ⇒ void

This method returns an undefined value.

Reject an awakeable from outside the Restate runtime.

Parameters:

  • awakeable_id (String)
  • message (String)
  • code: (Integer) (defaults to: 500) 


67
68
69
70
 
# File 'lib/restate/client.rb', line 67

def reject_awakeable(awakeable_id, message, code: 500)
  post_ingress("/restate/awakeables/#{awakeable_id}/reject",
               { 'message' => message, 'code' => code })
end

#resolve_awakeable(awakeable_id, payload) ⇒ void

This method returns an undefined value.

Resolve an awakeable from outside the Restate runtime.

Parameters:

  • awakeable_id (String)
  • payload (Object) 


62
63
64
 
# File 'lib/restate/client.rb', line 62

def resolve_awakeable(awakeable_id, payload)
  post_ingress("/restate/awakeables/#{awakeable_id}/resolve", payload)
end

#resolve_name(service) ⇒ String

Parameters:

  • service (Object)

Returns:

  • (String) 


113
114
115
116
117
118
119
 
# File 'lib/restate/client.rb', line 113

def resolve_name(service)
  if service.is_a?(Class) && service.respond_to?(:service_name)
    service.service_name # steep:ignore NoMethod
  else
    service.to_s
  end
end

#service(service) ⇒ ClientServiceProxy

Returns a proxy for calling a stateless service.

Parameters:

  • service (Object)

Returns:

  • (ClientServiceProxy) 


45
46
47
 
# File 'lib/restate/client.rb', line 45

def service(service)
  ClientServiceProxy.new(@ingress_url, resolve_name(service), nil, @ingress_headers)
end

#workflow(service, key) ⇒ ClientServiceProxy

Returns a proxy for calling a workflow.

Parameters:

  • service (Object)
  • key (String)

Returns:

  • (ClientServiceProxy) 


55
56
57
 
# File 'lib/restate/client.rb', line 55

def workflow(service, key)
  ClientServiceProxy.new(@ingress_url, resolve_name(service), key, @ingress_headers)
end