Class: Fwd::Buffer

Inherits:
Object
  • Object
show all
Extended by:
Forwardable
Defined in:
lib/fwd/buffer.rb

Constant Summary collapse

MAX_SIZE =

64M

64 * 1024 * 1024

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(core) ⇒ Buffer

Constructor

Parameters:



10
11
12
13
14
15
16
17
18
19
 
# File 'lib/fwd/buffer.rb', line 10

def initialize(core)
  @core     = core
  @interval = (core.opts[:flush_interval] || 60).to_i
  @rate     = (core.opts[:flush_rate] || 10_000).to_i
  @limit    = (core.opts[:flush_limit] || 0).to_i
  @count    = 0

  reschedule!
  rotate!
end

Instance Attribute Details

#core ⇒ Object (readonly)

Returns the value of attribute core.



6
7
8
 
# File 'lib/fwd/buffer.rb', line 6

def core
  @core
end

#count ⇒ Object (readonly)

Returns the value of attribute count.



6
7
8
 
# File 'lib/fwd/buffer.rb', line 6

def count
  @count
end

#fd ⇒ Object (readonly)

Returns the value of attribute fd.



6
7
8
 
# File 'lib/fwd/buffer.rb', line 6

def fd
  @fd
end

#interval ⇒ Object (readonly)

Returns the value of attribute interval.



6
7
8
 
# File 'lib/fwd/buffer.rb', line 6

def interval
  @interval
end

#limit ⇒ Object (readonly)

Returns the value of attribute limit.



6
7
8
 
# File 'lib/fwd/buffer.rb', line 6

def limit
  @limit
end

#path ⇒ Object (readonly)

Returns the value of attribute path.



6
7
8
 
# File 'lib/fwd/buffer.rb', line 6

def path
  @path
end

#rate ⇒ Object (readonly)

Returns the value of attribute rate.



6
7
8
 
# File 'lib/fwd/buffer.rb', line 6

def rate
  @rate
end

#timer ⇒ Object (readonly)

Returns the value of attribute timer.



6
7
8
 
# File 'lib/fwd/buffer.rb', line 6

def timer
  @timer
end

Instance Method Details

#concat(data) ⇒ Object

Parameters:

  • data (String) —

    binary data



22
23
24
25
26
27
 
# File 'lib/fwd/buffer.rb', line 22

def concat(data)
  rotate! if rotate?
  @fd.write(data)
  @count += 1
  flush! if flush?
end

#flush! ⇒ Object

(Force) flush buffer



30
31
32
33
34
35
36
 
# File 'lib/fwd/buffer.rb', line 30

def flush!
  @count = 0
  rotate!
  core.flush!
ensure
  reschedule!
end

#flush? ⇒ Boolean

Returns true if flush is due.

Returns:

  • (Boolean) —

    true if flush is due



39
40
41
 
# File 'lib/fwd/buffer.rb', line 39

def flush?
  (@rate > 0 && @count >= @rate) || (@limit > 0 && @path.size >= @limit)
end

#rotate! ⇒ Object

(Force) rotate buffer file



44
45
46
47
48
 
# File 'lib/fwd/buffer.rb', line 44

def rotate!
  FileUtils.mv(@path.to_s, @path.to_s.sub(/\.open$/, ".closed")) if @path
  @fd, @path = new_file
rescue Errno::ENOENT
end

#rotate? ⇒ Boolean

Returns true if rotation is due.

Returns:

  • (Boolean) —

    true if rotation is due



51
52
53
 
# File 'lib/fwd/buffer.rb', line 51

def rotate?
  !@fd || @path.size > MAX_SIZE
end