Module: PgPipeline::ClientOps

Defined in:
lib/pg_pipeline/client.rb

Class Method Summary collapse

Class Method Details

.abort!(client) ⇒ Object 



169
170
171
172
173
174
175
176
177
178
 
# File 'lib/pg_pipeline/client.rb', line 169

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 



158
159
160
161
162
163
164
165
166
167
 
# File 'lib/pg_pipeline/client.rb', line 158

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:



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

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:



184
185
186
187
188
 
# File 'lib/pg_pipeline/client.rb', line 184

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 



53
54
55
56
57
58
 
# File 'lib/pg_pipeline/client.rb', line 53

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

.pool(client) ⇒ Object 



180
181
182
 
# File 'lib/pg_pipeline/client.rb', line 180

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

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



81
82
83
84
85
86
 
# File 'lib/pg_pipeline/client.rb', line 81

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 



71
72
73
74
75
76
77
78
79
 
# File 'lib/pg_pipeline/client.rb', line 71

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.build(sql, params) }
  end
end

.query_prepared(client, statement, params) ⇒ Object 



88
89
90
91
92
93
94
 
# File 'lib/pg_pipeline/client.rb', line 88

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 



127
128
129
130
131
132
133
134
135
136
137
138
 
# File 'lib/pg_pipeline/client.rb', line 127

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) ⇒ Object

Raises:



60
61
62
63
64
65
66
67
68
69
 
# File 'lib/pg_pipeline/client.rb', line 60

def start(client)
  raise Error, "client already started" if started?(client)
  raise Error, "client start requires an active Fiber scheduler" unless Fiber.scheduler

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

.started?(client) ⇒ Boolean

Returns:

  • (Boolean) 


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

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

.stats(client) ⇒ Object 



153
154
155
156
 
# File 'lib/pg_pipeline/client.rb', line 153

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

.submit_with_failover(client) ⇒ Object 



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

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 



140
141
142
143
144
145
146
147
148
149
150
151
 
# File 'lib/pg_pipeline/client.rb', line 140

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 



96
97
98
99
100
101
 
# File 'lib/pg_pipeline/client.rb', line 96

def wait_for_request
  request = yield
  request.wait
ensure
  request.cancel! if request && !request.settled?
end