Module: PgPipeline::ClientOps

Defined in:
lib/pg_pipeline/client.rb

Class Method Summary collapse

Class Method Details

.abort!(client) ⇒ Object 



176
177
178
179
180
181
182
183
184
185
 
# File 'lib/pg_pipeline/client.rb', line 176

def abort!(client)
  return unless started?(client)

  ensure_context!(client)
  begin
    pool(client).abort!
  ensure
    client.__send__(:started=, false) if pool(client).closing?
  end
end

.close(client) ⇒ Object 



165
166
167
168
169
170
171
172
173
174
 
# File 'lib/pg_pipeline/client.rb', line 165

def close(client)
  return unless started?(client)

  ensure_context!(client)
  begin
    pool(client).graceful_close
  ensure
    client.__send__(:started=, false) if pool(client).closing?
  end
end

.ensure_context!(client) ⇒ Object

Raises:



197
198
199
200
201
202
 
# File 'lib/pg_pipeline/client.rb', line 197

def ensure_context!(client)
  return if Thread.current.equal?(client.__send__(:owner_thread)) &&
            Fiber.scheduler.equal?(client.__send__(:scheduler))

  raise Error, "client is reactor-local and cannot be used from another thread/scheduler"
end

.ensure_started!(client) ⇒ Object

Raises:



191
192
193
194
195
 
# File 'lib/pg_pipeline/client.rb', line 191

def ensure_started!(client)
  raise Error, "client not started; call #start or use Client.open" unless started?(client)

  ensure_context!(client)
end

.open(connection_args, opts) ⇒ Object 



55
56
57
58
59
60
61
62
 
# File 'lib/pg_pipeline/client.rb', line 55

def open(connection_args, opts)
  client = Client.new(connection_args, **opts).start
  begin
    yield client
  ensure
    client.close
  end
end

.pool(client) ⇒ Object 



187
188
189
 
# File 'lib/pg_pipeline/client.rb', line 187

def pool(client)
  client.__send__(:pool)
end

.prepare(client, name, sql, param_types) ⇒ Object 



84
85
86
87
88
89
 
# File 'lib/pg_pipeline/client.rb', line 84

def prepare(client, name, sql, param_types)
  ensure_started!(client)
  sql = RequestOps.snapshot_sql(sql)
  SessionGuard.assert_multiplexable_normalized!(sql, mode: client.guard)
  pool(client).__send__(:prepare_statement, client, name, sql, param_types)
end

.query(client, sql, params) ⇒ Object 



74
75
76
77
78
79
80
81
82
 
# File 'lib/pg_pipeline/client.rb', line 74

def query(client, sql, params)
  ensure_started!(client)
  sql = RequestOps.snapshot_sql(sql)
  SessionGuard.assert_multiplexable_normalized!(sql, mode: client.guard)

  wait_for_request do
    submit_with_failover(client) { Request.new(sql: sql, params: params) }
  end
end

.query_prepared(client, statement, params) ⇒ Object 



91
92
93
94
95
96
97
 
# File 'lib/pg_pipeline/client.rb', line 91

def query_prepared(client, statement, params)
  ensure_started!(client)

  wait_for_request do
    submit_with_failover(client) { Request.prepared_query(statement, params: params) }
  end
end

.session(client) ⇒ Object 



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

def session(client)
  ensure_started!(client)

  pool(client).__send__(:with_pinned) do |conn|
    session = Session.new(conn)
    begin
      yield session
    ensure
      SessionOps.close!(session)
    end
  end
end

.start(client, parent) ⇒ Object

Raises:



64
65
66
67
68
69
70
71
72
 
# File 'lib/pg_pipeline/client.rb', line 64

def start(client, parent)
  raise Error, "client already started" if started?(client)

  pool(client).start(parent: parent)
  client.__send__(:owner_thread=, Thread.current)
  client.__send__(:scheduler=, Fiber.scheduler)
  client.__send__(:started=, true)
  client
end

.started?(client) ⇒ Boolean

Returns:

  • (Boolean) 


204
205
206
 
# File 'lib/pg_pipeline/client.rb', line 204

def started?(client)
  client.__send__(:started)
end

.stats(client) ⇒ Object 



160
161
162
163
 
# File 'lib/pg_pipeline/client.rb', line 160

def stats(client)
  ensure_started!(client)
  pool(client).stats
end

.submit_with_failover(client) ⇒ Object

NotDispatchedError is safe to retry on a fresh Request by contract. ShutdownError is retried only while the current Request is still pre-dispatch.



110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
 
# File 'lib/pg_pipeline/client.rb', line 110

def submit_with_failover(client)
  attempts = 0
  limit = [pool(client).pipeline_size, 1].max
  last_error = nil

  while attempts < limit
    request = yield
    begin
      pool(client).__send__(:pipeline_driver).submit(request)
      return request
    rescue NotDispatchedError => e
      last_error = e
      attempts += 1
    rescue ShutdownError => e
      last_error = e
      attempts += 1
      break unless request.state == :new && !request.settled?
    end
  end

  error = last_error || NotDispatchedError.new("no live pipeline connections; request was not dispatched")
  raise(error.is_a?(NotDispatchedError) ? error : NotDispatchedError.new(error.message))
end

.transaction(client) ⇒ Object 



147
148
149
150
151
152
153
154
155
156
157
158
 
# File 'lib/pg_pipeline/client.rb', line 147

def transaction(client)
  ensure_started!(client)

  pool(client).__send__(:with_pinned) do |conn|
    tx = Transaction.new(conn)
    begin
      TransactionOps.run(tx) { |transaction| yield transaction }
    ensure
      SessionOps.close!(tx)
    end
  end
end

.wait_for_request ⇒ Object 



99
100
101
102
103
104
105
106
 
# File 'lib/pg_pipeline/client.rb', line 99

def wait_for_request
  request = yield
  begin
    request.wait
  ensure
    request.cancel! unless request.settled?
  end
end