svn commit: r825492 - in /qpid/trunk/qpid/cpp/bindings/qmf: ruby/qmf.rb tests/ruby_console.rb

[tr...@apache.org]

Author: tross
Date: Thu Oct 15 13:24:10 2009
New Revision: 825492

URL: http://svn.apache.org/viewvc?rev=825492&view=rev
Log:
Added select for object queries (Ruby).
Added separate callback-thread to prevent deadlock when a user blocks on a callback.

Modified:
    qpid/trunk/qpid/cpp/bindings/qmf/ruby/qmf.rb
    qpid/trunk/qpid/cpp/bindings/qmf/tests/ruby_console.rb

Modified: qpid/trunk/qpid/cpp/bindings/qmf/ruby/qmf.rb
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/bindings/qmf/ruby/qmf.rb?rev=825492&r1=825491&r2=825492&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/bindings/qmf/ruby/qmf.rb (original)
+++ qpid/trunk/qpid/cpp/bindings/qmf/ruby/qmf.rb Thu Oct 15 13:24:10 2009
@@ -871,17 +871,22 @@
       @cv = new_cond
       @sync_count = nil
       @sync_result = nil
+      @select = []
+      @cb_cond = new_cond
+      @cb_thread = Thread.new do
+        run_cb_thread
+      end
     end
 
     def add_connection(conn)
       broker = Broker.new(self, conn)
-      @broker_list << broker
+      synchronize { @broker_list << broker }
       return broker
     end
 
     def del_connection(broker)
       broker.shutdown
-      @broker_list.delete(broker)
+      synchronize { @broker_list.delete(broker) }
     end
 
     def packages()
@@ -935,7 +940,7 @@
       if broker
         blist << broker
       else
-        blist = @broker_list
+        synchronize { blist = @broker_list }
       end
 
       agents = []
@@ -960,10 +965,16 @@
 
       query = Query.new(kwargs) if query.class == Hash
 
+      @select = []
+      kwargs.each do |k,v|
+        @select << [k, v] if k.is_a?(String)
+      end
+
       synchronize do
         @sync_count = 1
         @sync_result = []
-        broker = @broker_list[0]
+        broker = nil
+        synchronize { broker = @broker_list[0] }
         broker.send_query(query.impl, nil)
         unless @cv.wait(timeout) { @sync_count == 0 }
           raise "Timed out waiting for response"
@@ -985,10 +996,22 @@
       return objs.length > 0 ? objs[0] : nil
     end
 
+    # Check the object against select to check for a match
+    def select_match(object)
+      @select.each do |key, value|
+        object.properties.each do |prop, propval|
+          if key == prop.name && value != propval
+            return nil
+          end
+        end
+      end
+      return :true
+    end
+
     def _get_result(list, context)
       synchronize do
         list.each do |item|
-          @sync_result << item
+          @sync_result << item if select_match(item)
         end
         @sync_count -= 1
         @cv.signal
@@ -1004,7 +1027,20 @@
     def end_sync(sync)
     end
 
-    def do_console_events()
+    def run_cb_thread
+      while :true
+        synchronize { @cb_cond.wait(1) }
+        begin
+          count = do_console_events
+        end until count == 0
+      end
+    end
+
+    def start_console_events
+      synchronize { @cb_cond.signal }
+    end
+
+    def do_console_events
       count = 0
       valid = @impl.getEvent(@event)
       while valid
@@ -1151,10 +1187,10 @@
 
     def do_events()
       begin
-        ccnt = @console.do_console_events
+        @console.start_console_events
         bcnt = do_broker_events
         mcnt = do_broker_messages
-      end until ccnt == 0 and bcnt == 0 and mcnt == 0
+      end until bcnt == 0 and mcnt == 0
     end
 
     def conn_event_connected()

Modified: qpid/trunk/qpid/cpp/bindings/qmf/tests/ruby_console.rb
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/bindings/qmf/tests/ruby_console.rb?rev=825492&r1=825491&r2=825492&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/bindings/qmf/tests/ruby_console.rb (original)
+++ qpid/trunk/qpid/cpp/bindings/qmf/tests/ruby_console.rb Thu Oct 15 13:24:10 2009
@@ -143,6 +143,13 @@
       end
       puts "----"
 
+      elist = @qmfc.objects(:package => "org.apache.qpid.broker", :class => "exchange", 'durable' => true)
+      puts "---- Durable Exchanges ----"
+      elist.each do |e|
+        puts "Exchange: #{e.name}"
+      end
+      puts "----"
+
       qlist = @qmfc.objects(Qmf::Query.new(:package => "org.apache.qpid.broker",
                                                :class => "queue"))
       puts "---- Queues ----"



---------------------------------------------------------------------
Apache Qpid - AMQP Messaging Implementation
Project:      http://qpid.apache.org
Use/Interact: mailto:commits-subscribe@qpid.apache.org