Class: PgEventstore::Connection

Inherits:
Object
  • Object
show all
Defined in:
lib/pg_eventstore/connection.rb,
sig/pg_eventstore/connection.rbs

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(uri:, pool_size: 5, pool_timeout: 5) ⇒ Connection

Returns a new instance of Connection.

Parameters:

  • uri (String) —

    PostgreSQL connection URI. Example: "postgresql://postgres:postgres@localhost:5432/eventstore"

  • pool_size (Integer) (defaults to: 5) —

    Connection pool size

  • pool_timeout (Integer) (defaults to: 5) —

    Connection pool timeout in seconds

  • uri: (String)
  • pool_size: (Integer) (defaults to: 5)
  • pool_timeout: (Integer) (defaults to: 5) 


26
27
28
29
30
31
 
# File 'lib/pg_eventstore/connection.rb', line 26

def initialize(uri:, pool_size: 5, pool_timeout: 5)
  @uri = uri
  @pool_size = pool_size
  @pool_timeout = pool_timeout
  init_pool
end

Instance Attribute Details

#pool_size ⇒ Integer

Returns the value of attribute pool_size.

Returns:

  • (Integer) 


16
17
18
 
# File 'lib/pg_eventstore/connection.rb', line 16

def pool_size
  @pool_size
end

#pool_timeout ⇒ Integer

Returns the value of attribute pool_timeout.

Returns:

  • (Integer) 


19
20
21
 
# File 'lib/pg_eventstore/connection.rb', line 19

def pool_timeout
  @pool_timeout
end

#uri ⇒ String

Returns the value of attribute uri.

Returns:

  • (String) 


13
14
15
 
# File 'lib/pg_eventstore/connection.rb', line 13

def uri
  @uri
end

Instance Method Details

#discard_current_connection ⇒ void

This method returns an undefined value.



69
70
71
72
73
 
# File 'lib/pg_eventstore/connection.rb', line 69

def discard_current_connection
  @pool.discard_current_connection do |conn|
    conn.close unless conn.finished?
  end
end

#establish_connection ⇒ void

This method returns an undefined value.

Restore connection after shutdown by re-initializing connection pool



63
64
65
66
 
# File 'lib/pg_eventstore/connection.rb', line 63

def establish_connection
  @pool = nil
  init_pool
end

#init_pool ⇒ ConnectionPool

rubocop:disable Naming/MemoizedInstanceVariableName

Returns:

  • (ConnectionPool) 


79
80
81
82
83
84
85
86
 
# File 'lib/pg_eventstore/connection.rb', line 79

def init_pool
  @pool ||= ConnectionPool.new(size: pool_size, timeout: pool_timeout) do
    PgConnection.new(uri).tap do |conn|
      conn.type_map_for_results = PG::BasicTypeMapForResults.new(conn, registry: pg_type_registry)
      conn.type_map_for_queries = PG::BasicTypeMapForQueries.new(conn, registry: pg_type_registry)
    end
  end
end

#pg_type_registry ⇒ PG::BasicTypeRegistry

Returns:



90
91
92
93
94
95
96
 
# File 'lib/pg_eventstore/connection.rb', line 90

def pg_type_registry
  registry = PG::BasicTypeRegistry.new.register_default_types
  # 0 means that the pg value format is a text(1 for binary)
  registry.alias_type(0, 'uuid', 'text')
  registry.register_type 0, 'timestamp', PG::TextEncoder::TimestampUtc, PG::TextDecoder::TimestampUtc
  registry
end

#shutdown ⇒ void

This method returns an undefined value.



57
58
59
 
# File 'lib/pg_eventstore/connection.rb', line 57

def shutdown
  @pool.shutdown(&:close)
end

#with {|connection| ... } ⇒ Object

A shorthand from ConnectionPool#with.

Yields:

Yield Parameters:

  • connection (PgEventstore::PgConnection) —

    PostgreSQL connection instance

Yield Returns:

  • (Object)

Returns:

  • (Object) —

    a value of a given block



36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
 
# File 'lib/pg_eventstore/connection.rb', line 36

def with(&)
  should_retry = true
  @pool.with do |conn|
    yield conn
  rescue PG::ConnectionBad, PG::UnableToSend
    unless should_retry
      # Connection is broken. We should throw it away
      @pool.discard_current_connection
      raise
    end

    # Recover a connection after fork or when we lost a connection to PostgreSQL. We retry only once and without any
    # delay.
    conn.sync_reset

    should_retry = false
    retry
  end
end