Class: Mayu::EventStream::Log

Inherits:
Object
  • Object
show all
Extended by:
T::Sig
Defined in:
lib/mayu/event_stream.rb

Instance Method Summary collapse

Constructor Details

#initialize ⇒ Log

Returns a new instance of Log.



108
109
110
111
112
 
# File 'lib/mayu/event_stream.rb', line 108

def initialize
  @history = T.let([], T::Array[Message])
  @queue = T.let(Async::Queue.new, Async::Queue)
  @wrapper = T.let(Wrapper.new, Wrapper)
end

Instance Method Details

#ack(id) ⇒ Object 



126
127
128
129
130
 
# File 'lib/mayu/event_stream.rb', line 126

def ack(id)
  if index = @history.map(&:id).index(id)
    @history.slice!(0..index)
  end
end

#empty? ⇒ Boolean

Returns:

  • (Boolean) 


115
 
# File 'lib/mayu/event_stream.rb', line 115

def empty? = @queue.empty?

#pack(message) ⇒ Object 



147
148
149
150
151
152
 
# File 'lib/mayu/event_stream.rb', line 147

def pack(message)
  data = @wrapper.pack(message.to_a)
  # N = 32-bit unsigned, network (big-endian) byte order
  # a = arbitrary binary string (null padded, count is width)
  [data.bytesize, data].pack("N a*")
end

#pop ⇒ Object 



139
140
141
142
143
144
 
# File 'lib/mayu/event_stream.rb', line 139

def pop
  message = @queue.dequeue
  # There is no ack-functionality in the client so this will just grow anyways..
  # @history.push(message)
  message
end

#push(event, data = {}) ⇒ Object 



121
122
123
 
# File 'lib/mayu/event_stream.rb', line 121

def push(event, data = {})
  @queue.enqueue(Message.new(event, data))
end

#replay(last_id) ⇒ Object 



133
134
135
136
 
# File 'lib/mayu/event_stream.rb', line 133

def replay(last_id)
  ack(last_id)
  @history.dup
end

#size ⇒ Object 



118
 
# File 'lib/mayu/event_stream.rb', line 118

def size = @queue.size