Class: Droonga::Forwarder

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

Instance Method Summary collapse

Constructor Details

#initialize(loop, options = {}) ⇒ Forwarder

Returns a new instance of Forwarder.



28
29
30
31
32
 
# File 'lib/droonga/forwarder.rb', line 28

def initialize(loop, options={})
  @loop = loop
  @buffering = options[:buffering]
  @senders = {}
end

Instance Method Details

#forward(message, destination) ⇒ Object 



48
49
50
51
52
53
54
55
 
# File 'lib/droonga/forwarder.rb', line 48

def forward(message, destination)
  logger.trace("forward: start")
  command = destination["type"]
  receiver = destination["to"]
  arguments = destination["arguments"]
  output(receiver, message, command, arguments)
  logger.trace("forward: done")
end

#resume ⇒ Object 



57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
 
# File 'lib/droonga/forwarder.rb', line 57

def resume
  return unless Path.buffer.exist?
  Pathname.glob("#{Path.buffer}/*") do |path|
    next unless path.directory?

    destination = path.basename.to_s
    sender = @senders[destination]
    if sender
      sender.resume
      next
    end

    chunk_loader = BufferedTCPSocket::ChunkLoader.new(path)
    unless chunk_loader.have_any_chunk?
      #FileUtils.rm_rf(path.to_s) # TODO re-enable this
      next
    end

    components = destination.split(":")
    port = components.pop.to_i
    next if port.zero?
    host = components.join(":")

    sender = create_sender(host, port)
    sender.resume
    @senders[destination] = sender
  end
end

#shutdown ⇒ Object 



40
41
42
43
44
45
46
 
# File 'lib/droonga/forwarder.rb', line 40

def shutdown
  logger.trace("shutdown: start")
  @senders.each_value do |sender|
    sender.shutdown
  end
  logger.trace("shutdown: done")
end

#start ⇒ Object 



34
35
36
37
38
 
# File 'lib/droonga/forwarder.rb', line 34

def start
  logger.trace("start: start")
  resume
  logger.trace("start: done")
end