Class: Sparrow::Queues::Disk
- Inherits:
-
Object
- Object
- Sparrow::Queues::Disk
- Includes:
- Miscel
- Defined in:
- lib/sparrow/queues/disk.rb
Constant Summary collapse
- TRX_CMD_PUSH =
"\000".freeze
- TRX_CMD_POP =
"\001".freeze
- TRX_PUSH =
"\000%s%s".freeze
- TRX_POP =
"\001".freeze
Instance Attribute Summary collapse
-
#count_pop ⇒ Object
Returns the value of attribute count_pop.
-
#count_push ⇒ Object
Returns the value of attribute count_push.
-
#queue_name ⇒ Object
Returns the value of attribute queue_name.
-
#trxr ⇒ Object
Returns the value of attribute trxr.
-
#trxw ⇒ Object
Returns the value of attribute trxw.
Instance Method Summary collapse
- #clear ⇒ Object
- #count ⇒ Object
-
#initialize(queue_name) ⇒ Disk
constructor
A new instance of Disk.
- #pop ⇒ Object
- #push(value) ⇒ Object
Methods included from Miscel
#base_dir, #log_path, #logger, #options, #options=, #pid_dir
Constructor Details
#initialize(queue_name) ⇒ Disk
Returns a new instance of Disk.
19 20 21 22 23 24 |
# File 'lib/sparrow/queues/disk.rb', line 19 def initialize(queue_name) self.queue_name = queue_name self.count_pop = 0 self.count_push = 0 open_queue end |
Instance Attribute Details
#count_pop ⇒ Object
Returns the value of attribute count_pop.
16 17 18 |
# File 'lib/sparrow/queues/disk.rb', line 16 def count_pop @count_pop end |
#count_push ⇒ Object
Returns the value of attribute count_push.
17 18 19 |
# File 'lib/sparrow/queues/disk.rb', line 17 def count_push @count_push end |
#queue_name ⇒ Object
Returns the value of attribute queue_name.
13 14 15 |
# File 'lib/sparrow/queues/disk.rb', line 13 def queue_name @queue_name end |
#trxr ⇒ Object
Returns the value of attribute trxr.
14 15 16 |
# File 'lib/sparrow/queues/disk.rb', line 14 def trxr @trxr end |
#trxw ⇒ Object
Returns the value of attribute trxw.
15 16 17 |
# File 'lib/sparrow/queues/disk.rb', line 15 def trxw @trxw end |
Instance Method Details
#clear ⇒ Object
69 70 71 72 |
# File 'lib/sparrow/queues/disk.rb', line 69 def clear dirs = Dir.glob(queue_path) | Dir.glob(queue_path + '.*') FileUtils.rm_rf(dirs) unless dirs.empty? end |
#count ⇒ Object
74 75 76 |
# File 'lib/sparrow/queues/disk.rb', line 74 def count self.count_push - self.count_pop end |
#pop ⇒ Object
38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 |
# File 'lib/sparrow/queues/disk.rb', line 38 def pop while !trxr.eof? s_pos = trxr.pos cmd = trxr.read(1) if cmd != TRX_CMD_POP and cmd != TRX_CMD_PUSH logger.fatal 'Corrupt queue' return end raw_size = trxr.read(4) size = raw_size.unpack("I").first value = trxr.read(size) next if cmd == TRX_CMD_POP e_pos = trxr.pos trxr.seek(s_pos, IO::SEEK_SET) trxr.write(TRX_POP) # trxr.fsync trxr.pos = e_pos next unless value self.count_pop += 1 return value end if trxr.path == queue_path File.truncate(trxr.path, 0) else FileUtils.rm_rf trxr.path end open_reader nil end |
#push(value) ⇒ Object
26 27 28 29 30 31 32 33 34 35 36 |
# File 'lib/sparrow/queues/disk.rb', line 26 def push(value) value = value.to_s size = [value.size].pack("I") data = sprintf(TRX_PUSH, size, value) trxw.seek(0, IO::SEEK_END) trxw.write data # trxw.fsync rotate_queue if trxw.pos > max_log_size self.count_push += 1 value end |