rjungemann · May 29, 2010 18:09 · May 29, 2010 · May 29, 2010 · May 29, 2010
diff --git a/client.rb b/client.rb
@@ -0,0 +1,13 @@
+STDOUT.sync = true
+
+require 'queue'
+
+start_time = Time.now.to_i
+msg = 0
+
+queue = Queue.new("testing")
+queue.subscribe do |obj|
+  msg += 1
+  seconds = Time.now.to_i - start_time
+  puts "%.3f ... " % [msg.to_f / seconds]
+end
diff --git a/server.rb b/server.rb
@@ -0,0 +1,9 @@
+require 'queue'
+
+queue = Queue.new("testing")
+queue.clear
+
+loop do
+  queue.push 1
+  puts queue.size
+end
diff --git a/queue.rb b/queue.rb
@@ -1,35 +1,47 @@
-Very efficient message queue backed by redis, using msgpack for serialization.
-
-
 require 'rubygems'
 require 'redis'
 require 'digest/md5'
-require 'msgpack'
+#require 'msgpack'
 
 $redis = Redis.new
 
+# Very efficient message queue.
+# Performance: ~ 2100 messages per second on i7 iMac
 class Queue
 
   def initialize(name)
-    @name = name
+    @queue_name = "queue:#{name}"    
+  end
+
+  def clear
+    $redis.del(@queue_name)
+  end
+
+  def size
+    $redis.llen(@queue_name)
   end
 
   def push(object)
     hash = Digest::MD5.hexdigest(object.to_s)
     $redis.set("msg:#{hash}", MessagePack.pack(object))
-    $redis.rpush("queue:#{@name}", hash)
+    $redis.rpush(@queue_name, hash)
   end
 
   def subscribe
     loop do            
-      hash = $redis.blpop("queue:#{@name}", 0)[1]
+      hash = $redis.blpop(@queue_name, 0)[1]
       object = $redis.get("msg:#{hash}")
-      begin
-        yield MessagePack.unpack(object)
-        $redis.del("msg:#{hash}")
-      rescue
-        $redis.lpush("queue:#{@name}", hash)
-      end      
+      if object
+        begin
+          yield MessagePack.unpack(object)
+          # Done, remove message from redis. 
+          $redis.del("msg:#{hash}")
+        rescue
+          # Error, add the message again to the end of the queue
+          $redis.rpush(@queue_name, hash)
+          raise
+        end      
+      end
     end    
   end
 end

diff --git a/queue.rb b/queue.rb
@@ -0,0 +1,36 @@
+Very efficient message queue backed by redis, using msgpack for serialization.
+
+
+require 'rubygems'
+require 'redis'
+require 'digest/md5'
+require 'msgpack'
+
+$redis = Redis.new
+
+class Queue
+
+  def initialize(name)
+    @name = name
+  end
+
+  def push(object)
+    hash = Digest::MD5.hexdigest(object.to_s)
+    $redis.set("msg:#{hash}", MessagePack.pack(object))
+    $redis.rpush("queue:#{@name}", hash)
+  end
+
+  def subscribe
+    loop do            
+      hash = $redis.blpop("queue:#{@name}", 0)[1]
+      object = $redis.get("msg:#{hash}")
+      begin
+        yield MessagePack.unpack(object)
+        $redis.del("msg:#{hash}")
+      rescue
+        $redis.lpush("queue:#{@name}", hash)
+      end      
+    end    
+  end
+end
+