Module: PgPipeline::ClientOps

Defined in:
lib/pg_pipeline/client.rb

Class Method Summary collapse

Class Method Details

.abort!(client) ⇒ Object 



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

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 



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

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:



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

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:



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

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
 
# File 'lib/pg_pipeline/client.rb', line 55

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

.pool(client) ⇒ Object 



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

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

.prepare(client, name, sql, param_types, typed: false) ⇒ Object 



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

def prepare(client, name, sql, param_types, typed: false)
  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, typed: typed)
end

.query(client, sql, params) ⇒ Object 



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

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 



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

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

  wait_for_request do
    submit_with_failover(client) do
      request = Request.prepared_query(statement, params: params)
      pool(client).bind_type_map(request, statement) if statement.typed?
      request
    end
  end
end

.session(client) ⇒ Object 



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

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:



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

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) 


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

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

.stats(client) ⇒ Object 



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

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

.submit_with_failover(client) ⇒ Object 



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

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 



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

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 



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

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