Class: MessageStore::Postgres::Get::Stream::Last

Inherits:
Object
  • Object
show all
Includes:
Get::Stream::Last
Defined in:
lib/message_store/postgres/get/stream/last.rb

Defined Under Namespace

Modules: Deserialize, Time

Class Method Summary collapse

Instance Method Summary collapse

Class Method Details

.sql_command(stream_name) ⇒ Object 



47
48
49
50
51
 
# File 'lib/message_store/postgres/get/stream/last.rb', line 47

def self.sql_command(stream_name)
  parameters = '$1::varchar'

  "SELECT * FROM get_last_message(#{parameters});"
end

Instance Method Details

#call(stream_name) ⇒ Object 



14
15
16
17
18
19
20
21
22
23
24
25
26
27
 
# File 'lib/message_store/postgres/get/stream/last.rb', line 14

def call(stream_name)
  logger.trace(tag: :get) { "Getting last message data (Stream Name: #{stream_name})" }

  result = get_result(stream_name)

  return nil if result.nil?

  message_data = convert(result[0])

  logger.info(tag: :get) { "Finished getting message data (Stream Name: #{stream_name})" }
  logger.info(tags: [:data, :message_data]) { message_data.pretty_inspect }

  message_data
end

#configure(session: nil) ⇒ Object 



10
11
12
 
# File 'lib/message_store/postgres/get/stream/last.rb', line 10

def configure(session: nil)
  Session.configure(self, session: session)
end

#convert(record) ⇒ Object 



53
54
55
56
57
58
59
60
61
62
63
64
65
 
# File 'lib/message_store/postgres/get/stream/last.rb', line 53

def convert(record)
  logger.trace(tag: :get) { "Converting record to message data" }

  record['data'] = Deserialize.data(record['data'])
  record['metadata'] = Deserialize.metadata(record['metadata'])
  record['time'] = Time.utc_coerced(record['time'])

  message_data = MessageData::Read.build(record)

  logger.debug(tag: :get) { "Converted record to message data" }

  message_data
end

#get_result(stream_name) ⇒ Object 



29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
 
# File 'lib/message_store/postgres/get/stream/last.rb', line 29

def get_result(stream_name)
  logger.trace(tag: :get) { "Getting last record (Stream: #{stream_name})" }

  sql_command = self.class.sql_command(stream_name)

  params = [
    stream_name
  ]

  result = session.execute(sql_command, params)

  logger.debug(tag: :get) { "Finished getting result (Count: #{result.ntuples}, Stream: #{stream_name}" }

  return nil if result.ntuples == 0

  result
end