Class: EventSource::Postgres::Get::SelectStatement

Inherits:
Object
  • Object
show all
Includes:
Log::Dependency
Defined in:
lib/event_source/postgres/get/select_statement.rb

Defined Under Namespace

Modules: Defaults

Class Method Summary collapse

Instance Method Summary collapse

Class Method Details

.build(stream_name, position: nil, batch_size: nil) ⇒ Object 



25
26
27
 
# File 'lib/event_source/postgres/get/select_statement.rb', line 25

def self.build(stream_name, position: nil, batch_size: nil)
  new(stream_name, position, batch_size)
end

Instance Method Details

#batch_size ⇒ Object 



13
14
15
 
# File 'lib/event_source/postgres/get/select_statement.rb', line 13

def batch_size
  @batch_size ||= Defaults.batch_size
end

#category_stream? ⇒ Boolean

Returns:

  • (Boolean) 


21
22
23
 
# File 'lib/event_source/postgres/get/select_statement.rb', line 21

def category_stream?
  is_category_stream ||= StreamName.category?(stream_name)
end

#position ⇒ Object 



9
10
11
 
# File 'lib/event_source/postgres/get/select_statement.rb', line 9

def position
  @position ||= Defaults.position
end

#position_field ⇒ Object 



68
69
70
71
72
73
74
 
# File 'lib/event_source/postgres/get/select_statement.rb', line 68

def position_field
  unless category_stream?
    'position'
  else
    'global_position'
  end
end

#sql ⇒ Object 



29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
 
# File 'lib/event_source/postgres/get/select_statement.rb', line 29

def sql
  logger.trace(tag: :sql) { "Composing select statement (Stream: #{stream_name}, Category: #{category_stream?}, Type: #{stream_type.inspect}, Position: #{position}, Batch Size: #{batch_size})" }

  statement = <<-SQL
    SELECT
      id::varchar,
      stream_name::varchar,
      position::int,
      type::varchar,
      global_position::bigint,
      data::varchar,
      metadata::varchar,
      time::timestamp
    FROM
      events
    WHERE
      #{where_clause_field} = '#{stream_name}' AND
      #{position_field} >= #{position}
    ORDER BY
      #{position_field} ASC
    LIMIT
      #{batch_size}
    ;
  SQL

  logger.debug(tag: :sql) { "Composed select statement (Stream: #{stream_name}, Category: #{category_stream?}, Type: #{stream_type.inspect}, Position: #{position}, Batch Size: #{batch_size})" }
  logger.debug(tags: [:data, :sql]) { "Statement: #{statement}" }

  statement
end

#stream_type ⇒ Object 



17
18
19
 
# File 'lib/event_source/postgres/get/select_statement.rb', line 17

def stream_type
  @stream_type ||= StreamName.get_type(stream_name)
end

#where_clause_field ⇒ Object 



60
61
62
63
64
65
66
 
# File 'lib/event_source/postgres/get/select_statement.rb', line 60

def where_clause_field
  unless category_stream?
    'stream_name'
  else
    'category(stream_name)'
  end
end