Module: PgPipeline::TransactionOps

Defined in:
lib/pg_pipeline/transaction.rb

Class Method Summary collapse

Class Method Details

.rollback_quietly(tx) ⇒ Object 



82
83
84
85
86
87
88
89
90
91
 
# File 'lib/pg_pipeline/transaction.rb', line 82

def rollback_quietly(tx)
  return unless tx.open?

  conn = SessionOps.connection(tx)
  conn&.exec("ROLLBACK")
rescue PG::Error
  nil
ensure
  tx.__send__(:open=, false)
end

.rollback_to_savepoint(tx, ident) ⇒ Object 



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

def rollback_to_savepoint(tx, ident)
  conn = SessionOps.connection(tx)
  return unless conn

  conn.exec("ROLLBACK TO SAVEPOINT #{ident}")
  conn.exec("RELEASE SAVEPOINT #{ident}")
rescue PG::Error
  nil
end

.run(tx) ⇒ Object

Raises:



33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
 
# File 'lib/pg_pipeline/transaction.rb', line 33

def run(tx)
  SessionOps.ensure_active!(tx)
  raise Error, "transaction is already open" if tx.open?

  conn = SessionOps.connection(tx)
  conn.exec("BEGIN")
  tx.__send__(:open=, true)
  begin
    result = yield tx
    conn.exec("COMMIT")
    tx.__send__(:open=, false)
    result
  rescue Exception
    rollback_quietly(tx)
    raise
  end
end

.savepoint(tx, name) ⇒ Object

Raises:



51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
 
# File 'lib/pg_pipeline/transaction.rb', line 51

def savepoint(tx, name)
  SessionOps.ensure_active!(tx)
  raise Error, "savepoint requires an open transaction" unless tx.open?

  conn = SessionOps.connection(tx)
  seq = tx.__send__(:savepoint_seq) + 1
  tx.__send__(:savepoint_seq=, seq)
  point = name || "pgp_sp_#{seq}"
  ident = conn.quote_ident(point)

  conn.exec("SAVEPOINT #{ident}")
  begin
    result = yield tx
    conn.exec("RELEASE SAVEPOINT #{ident}")
    result
  rescue Exception
    rollback_to_savepoint(tx, ident)
    raise
  end
end