Class: Droonga::ForwardBuffer

Inherits:
Object
  • Object
show all
Includes:
Loggable
Defined in:
lib/droonga/forward_buffer.rb

Constant Summary collapse

SUFFIX = 
".msgpack"

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(node_name) ⇒ ForwardBuffer

Returns a new instance of ForwardBuffer.



37
38
39
40
41
42
43
44
45
 
# File 'lib/droonga/forward_buffer.rb', line 37

def initialize(node_name)
  @on_forward = nil

  @packer = MessagePack::Packer.new
  @unpacker = MessagePack::Unpacker.new

  @target = node_name
  @process_messages_newer_than_timestamp = nil
end

Instance Attribute Details

#on_forward=(value) ⇒ Object

Sets the attribute on_forward

Parameters:

  • value —

    the value to set the attribute on_forward to.



35
36
37
 
# File 'lib/droonga/forward_buffer.rb', line 35

def on_forward=(value)
  @on_forward = value
end

Instance Method Details

#add(message, destination) ⇒ Object 



47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
 
# File 'lib/droonga/forward_buffer.rb', line 47

def add(message, destination)
  logger.trace("add: start")
  serf.set_have_unprocessed_messages_for(@target)
  buffered_message = {
    "message"     => message,
    "destination" => destination,
  }
  @packer.pack(buffered_message)
  file_path = create_buffered_message_path
  SafeFileWriter.write(file_path) do |output, file|
    output.puts(@packer.to_s)
  end
  @packer.clear
  logger.trace("add: done", :path => file_path)
end

#buffered_messages ⇒ Object 



84
85
86
87
88
 
# File 'lib/droonga/forward_buffer.rb', line 84

def buffered_messages
  Pathname.glob("#{data_directory}/*#{SUFFIX}").sort_by do |path|
    path
  end
end

#empty? ⇒ Boolean

Returns:

  • (Boolean) 


90
91
92
 
# File 'lib/droonga/forward_buffer.rb', line 90

def empty?
  data_directory.children.empty?
end

#process_messages_newer_than(timestamp) ⇒ Object 



94
95
96
 
# File 'lib/droonga/forward_buffer.rb', line 94

def process_messages_newer_than(timestamp)
  @process_messages_newer_than_timestamp = timestamp
end

#start_forward ⇒ Object 



63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
 
# File 'lib/droonga/forward_buffer.rb', line 63

def start_forward
  logger.trace("start_forward: start")
  n_forwarded_messages = 0
  buffered_messages.each do |buffered_message_path|
    forwarded = forward(buffered_message_path)
    n_forwarded_messages += 1 if forwarded
  end
  if n_forwarded_messages > 0 and
       @process_messages_newer_than_timestamp
    logger.info("#{n_forwarded_messages} new messages forwarded. " +
                  "The boundary is now cleared.")
    # Don't clear the boundary while forwarding.
    # Because buffered messages are not sorted by their "date",
    # so older messages can appear after newer one.
    # (ex. messages generated by Dispatcher)
    @process_messages_newer_than_timestamp = nil
  end
  serf.reset_have_unprocessed_messages_for(@target)
  logger.trace("start_forward: done")
end