Added raise_event support to the Ruby and Python wrapped agent APIs.

git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@898727 13f79535-47bb-0310-9956-ffa450edef68

16 files changed, 509 insertions, 21 deletions

diff --git a/qpid/cpp/bindings/qmf/python/qmf.py b/qpid/cpp/bindings/qmf/python/qmf.py
index c00c777878..eec975c50f 100644
--- a/qpid/cpp/bindings/qmf/python/qmf.py
+++ b/qpid/cpp/bindings/qmf/python/qmf.py
@@ -34,6 +34,8 @@ from qmfengine import (TYPE_ABSTIME, TYPE_ARRAY, TYPE_BOOL, TYPE_DELTATIME,
                        TYPE_UINT8, TYPE_UUID)
 from qmfengine import (O_EQ, O_NE, O_LT, O_LE, O_GT, O_GE, O_RE_MATCH, O_RE_NOMATCH,
                        E_NOT, E_AND, E_OR, E_XOR)
+from qmfengine import (SEV_EMERG, SEV_ALERT, SEV_CRIT, SEV_ERROR, SEV_WARN, SEV_NOTICE,
+                       SEV_INFORM, SEV_DEBUG)
 
 
 def qmf_to_native(val):
@@ -351,9 +353,103 @@ class Session:
 
 
   ##==============================================================================
-  ## OBJECTS
+  ## OBJECTS and EVENTS
   ##==============================================================================
 
+class QmfEvent(object):
+    # attr_reader :impl, :event_class
+    def __init__(self, cls, kwargs={}):
+        self._allow_sets = True
+        if kwargs.has_key("broker"):
+            self._broker = kwargs["broker"]
+        else:
+            self._broker = None
+        if cls:
+            self.event_class = cls
+            self.impl = qmfengine.Event(self.event_class.impl)
+        elif kwargs.has_key("impl"):
+            self.impl = qmfengine.Event(kwargs["impl"])
+            self.event_class = SchemaEventClass(None, None, 0,
+                                                {"impl":self.impl.getClass()})
+        else:
+            raise Exception("Argument error: required parameter ('impl') not supplied")
+    
+    
+    def arguments(self):
+        list = []
+        for arg in self.event_class.arguments:
+            list.append([arg, self.get_attr(arg.name())])
+        return list
+
+
+    def get_attr(self, name):
+        val = self._value(name)
+        return qmf_to_native(val)
+    
+    
+    def set_attr(self, name, v):
+        val = self._value(name)
+        native_to_qmf(val, v)
+    
+    
+    def __getitem__(self, name):
+        return self.get_attr(name)
+    
+    
+    def __setitem__(self, name, value):
+        self.set_attr(name, value)
+    
+    
+    def __setattr__(self, name, value):
+        #
+        # Ignore the internal attributes, set them normally...
+        #
+        if (name[0] == '_' or
+            name == 'impl' or
+            name == 'event_class'):
+            return super.__setattr__(self, name, value)
+
+        if not self._allow_sets:
+            raise Exception("'Set' operations not permitted on this object")
+
+        #
+        # If the name matches an argument name, set the value of the argument.
+        #
+        # print "set name=%s" % str(name)
+        for arg in self.event_class.arguments:
+            if arg.name() == name:
+                return self.set_attr(name, value)
+
+        # unrecognized name?  should I raise an exception?
+        super.__setattr__(self, name, value)
+
+
+    def __getattr__(self, name, *args):
+        #
+        # If the name matches an argument name, return the value of the argument.
+        #
+        for arg in self.event_class.arguments:
+            if arg.name() == name:
+                return self.get_attr(name)
+
+        #
+        # This name means nothing to us, pass it up the line to the parent
+        # class's handler.
+        #
+        # print "__getattr__=%s" % str(name)
+        super.__getattr__(self, name)
+
+
+    def _value(self, name):
+        val = self.impl.getValue(name)
+        if not val:
+            raise Exception("Argument error: attribute named '%s' not defined for package %s, class %s" % 
+                            (name,
+                             self.event_class.impl.getClassKey().getPackageName(),
+                             self.event_class.impl.getClassKey().getClassName()))
+        return val
+
+
 class QmfObject(object):
     # attr_reader :impl, :object_class
     def __init__(self, cls, kwargs={}):
@@ -922,14 +1018,14 @@ class SchemaObjectClass:
 
 class SchemaEventClass:
     # attr_reader :impl :arguments
-    def __init__(self, package, name, kwargs={}):
+    def __init__(self, package, name, sev, kwargs={}):
       self.arguments = []
       if "impl" in kwargs:
           self.impl = kwargs["impl"]
           for i in range(self.impl.getArgumentCount()):
               self.arguments.append(SchemaArgument(nil, nil, {"impl":self.impl.getArgument(i)}))
       else:
-          self.impl = qmfengine.SchemaEventClass(package, name)
+          self.impl = qmfengine.SchemaEventClass(package, name, sev)
           if kwargs.has_key("desc"): self.impl.setDesc(kwargs["desc"])
     
     
@@ -1026,7 +1122,7 @@ class Console(Thread):
                 if kind == CLASS_OBJECT:
                     clist.append(SchemaObjectClass(None, None, {"impl":self.impl.getObjectClass(key)}))
                 elif kind == CLASS_EVENT:
-                    clist.append(SchemaEventClass(None, None, {"impl":self.impl.getEventClass(key)}))
+                    clist.append(SchemaEventClass(None, None, 0, {"impl":self.impl.getEventClass(key)}))
         return clist
     
     
@@ -1436,7 +1532,10 @@ class Agent(ConnectionHandler):
     def alloc_object_id(self, low = 0, high = 0):
         return ObjectId(self.impl.allocObjectId(low, high))
     
-    
+
+    def raise_event(self, event):
+        self.impl.raiseEvent(event.impl)
+
     def query_response(self, context, obj):
         self.impl.queryResponse(context, obj.impl)
     
diff --git a/qpid/cpp/bindings/qmf/qmfengine.i b/qpid/cpp/bindings/qmf/qmfengine.i
index 3477215254..eb350115a3 100644
--- a/qpid/cpp/bindings/qmf/qmfengine.i
+++ b/qpid/cpp/bindings/qmf/qmfengine.i
@@ -37,6 +37,7 @@
 %include <qmf/engine/Value.h>
 %include <qmf/engine/ObjectId.h>
 %include <qmf/engine/Object.h>
+%include <qmf/engine/Event.h>
 
 
 %inline {
diff --git a/qpid/cpp/bindings/qmf/ruby/qmf.rb b/qpid/cpp/bindings/qmf/ruby/qmf.rb
index ea085d1be2..cc2aadc337 100644
--- a/qpid/cpp/bindings/qmf/ruby/qmf.rb
+++ b/qpid/cpp/bindings/qmf/ruby/qmf.rb
@@ -26,7 +26,8 @@ module Qmf
 
   # Pull all the TYPE_* constants into Qmf namespace.  Maybe there's an easier way?
   Qmfengine.constants.each do |c|
-    if c.index('TYPE_') == 0 or c.index('ACCESS_') == 0 or c.index('DIR_') == 0 or c.index('CLASS_') == 0
+    if c.index('TYPE_') == 0 or c.index('ACCESS_') == 0 or c.index('DIR_') == 0 or
+        c.index('CLASS_') == 0 or c.index('SEV_') == 0
       const_set(c, Qmfengine.const_get(c))
     end
   end
@@ -307,9 +308,90 @@ module Qmf
   end
 
   ##==============================================================================
-  ## OBJECTS
+  ## OBJECTS and EVENTS
   ##==============================================================================
 
+  class QmfEvent
+    attr_reader :impl, :event_class
+    def initialize(cls, kwargs={})
+      @broker = kwargs[:broker] if kwargs.include?(:broker)
+      @allow_sets = :true
+
+      if cls:
+        @event_class = cls
+        @impl = Qmfengine::Event.new(@event_class.impl)
+      elsif kwargs.include?(:impl)
+        @impl = Qmfengine::Event.new(kwargs[:impl])
+        @event_class = SchemaEventClass.new(nil, nil, :impl => @impl.getClass)
+      end
+    end
+
+    def arguments
+      list = []
+      @event_class.arguments.each do |arg|
+        list << [arg, get_attr(arg.name)]
+      end
+      return list
+    end
+
+    def get_attr(name)
+      val = value(name)
+      $util.qmf_to_native(val)
+    end
+
+    def set_attr(name, v)
+      val = value(name)
+      $util.native_to_qmf(val, v)
+    end
+
+    def [](name)
+      get_attr(name)
+    end
+
+    def []=(name, value)
+      set_attr(name, value)
+    end
+
+    def method_missing(name_in, *args)
+      #
+      # Convert the name to a string and determine if it represents an
+      # attribute assignment (i.e. "attr=")
+      #
+      name = name_in.to_s
+      attr_set = (name[name.length - 1] == 61)
+      name = name[0..name.length - 2] if attr_set
+      raise "Sets not permitted on this object" if attr_set && !@allow_sets
+
+      #
+      # If the name matches an argument name, set or return the value of the argument.
+      #
+      @event_class.arguments.each do |arg|
+        if arg.name == name
+          if attr_set
+            return set_attr(name, args[0])
+          else
+            return get_attr(name)
+          end
+        end
+      end
+
+      #
+      # This name means nothing to us, pass it up the line to the parent
+      # class's handler.
+      #
+      super.method_missing(name_in, args)
+    end
+
+    private
+    def value(name)
+      val = @impl.getValue(name.to_s)
+      if val.nil?
+        raise ArgumentError, "Attribute '#{name}' not defined for event #{@event_class.impl.getClassKey.getPackageName}:#{@object_class.impl.getClassKey.getClassName}"
+      end
+      return val
+    end
+  end
+
   class QmfObject
     include MonitorMixin
     attr_reader :impl, :object_class
@@ -845,7 +927,7 @@ module Qmf
 
   class SchemaEventClass
     attr_reader :impl, :arguments
-    def initialize(package, name, kwargs={})
+    def initialize(package, name, sev, kwargs={})
       @arguments = []
       if kwargs.include?(:impl)
         @impl = kwargs[:impl]
@@ -853,7 +935,7 @@ module Qmf
           @arguments << SchemaArgument.new(nil, nil, :impl => @impl.getArgument(i))
         end
       else
-        @impl = Qmfengine::SchemaEventClass.new(package, name)
+        @impl = Qmfengine::SchemaEventClass.new(package, name, sev)
         @impl.setDesc(kwargs[:desc]) if kwargs.include?(:desc)
       end
     end
@@ -1288,6 +1370,10 @@ module Qmf
       ObjectId.new(@impl.allocObjectId(low, high))
     end
 
+    def raise_event(event)
+      @impl.raiseEvent(event.impl)
+    end
+
     def query_response(context, object)
       @impl.queryResponse(context, object.impl)
     end
diff --git a/qpid/cpp/bindings/qmf/tests/agent_ruby.rb b/qpid/cpp/bindings/qmf/tests/agent_ruby.rb
index 0be779994a..86eefc59a6 100755
--- a/qpid/cpp/bindings/qmf/tests/agent_ruby.rb
+++ b/qpid/cpp/bindings/qmf/tests/agent_ruby.rb
@@ -23,7 +23,7 @@ require 'qmf'
 require 'socket'
 
 class Model
-  attr_reader :parent_class, :child_class
+  attr_reader :parent_class, :child_class, :event_class
 
   def initialize
     @parent_class = Qmf::SchemaObjectClass.new("org.apache.qpid.qmf", "parent")
@@ -79,11 +79,16 @@ class Model
 
     @child_class = Qmf::SchemaObjectClass.new("org.apache.qpid.qmf", "child")
     @child_class.add_property(Qmf::SchemaProperty.new("name", Qmf::TYPE_SSTR, :index => true))
+
+    @event_class = Qmf::SchemaEventClass.new("org.apache.qpid.qmf", "test_event", Qmf::SEV_INFORM)
+    @event_class.add_argument(Qmf::SchemaArgument.new("uint32val", Qmf::TYPE_UINT32))
+    @event_class.add_argument(Qmf::SchemaArgument.new("strval", Qmf::TYPE_LSTR))
   end
 
   def register(agent)
     agent.register_class(@parent_class)
     agent.register_class(@child_class)
+    agent.register_class(@event_class)
   end
 end
 
@@ -139,6 +144,11 @@ class App < Qmf::AgentHandler
                           'sstr' => "Short String",
                           'map'  => {'first' => 'FIRST', 'second' => 'SECOND'}}
 
+        event = Qmf::QmfEvent.new(@model.event_class)
+        event.uint32val = @parent.uint32val
+        event.strval = "Unused"
+        @agent.raise_event(event)
+
       elsif args['test'] == "small"
         @parent.uint64val = 4
         @parent.uint32val = 5
@@ -150,6 +160,11 @@ class App < Qmf::AgentHandler
         @parent.int16val = 10
         @parent.int8val  = 11
 
+        event = Qmf::QmfEvent.new(@model.event_class)
+        event.uint32val = @parent.uint32val
+        event.strval = "Unused"
+        @agent.raise_event(event)
+
       elsif args['test'] == "negative"
         @parent.uint64val = 0
         @parent.uint32val = 0
@@ -161,6 +176,11 @@ class App < Qmf::AgentHandler
         @parent.int16val = -1000
         @parent.int8val  = -100
 
+        event = Qmf::QmfEvent.new(@model.event_class)
+        event.uint32val = @parent.uint32val
+        event.strval = "Unused"
+        @agent.raise_event(event)
+
       else
         retCode = 1
         retText = "Invalid argument value for test"
@@ -169,9 +189,19 @@ class App < Qmf::AgentHandler
     elsif name == "set_short_string"
       @parent.sstrval = args['value']
 
+      event = Qmf::QmfEvent.new(@model.event_class)
+      event.uint32val = 0
+      event.strval = @parent.sstrval
+      @agent.raise_event(event)
+
     elsif name == "set_long_string"
       @parent.lstrval = args['value']
 
+      event = Qmf::QmfEvent.new(@model.event_class)
+      event.uint32val = 0
+      event.strval = @parent.lstrval
+      @agent.raise_event(event)
+
     elsif name == "create_child"
       oid = @agent.alloc_object_id(2)
       args['child_ref'] = oid
diff --git a/qpid/cpp/bindings/qmf/tests/python_agent.py b/qpid/cpp/bindings/qmf/tests/python_agent.py
index c42273d7b2..83531830b1 100644
--- a/qpid/cpp/bindings/qmf/tests/python_agent.py
+++ b/qpid/cpp/bindings/qmf/tests/python_agent.py
@@ -82,10 +82,15 @@ class Model:
         self.child_class = qmf.SchemaObjectClass("org.apache.qpid.qmf", "child")
         self.child_class.add_property(qmf.SchemaProperty("name", qmf.TYPE_SSTR, {"index":True}))
 
+        self.event_class = qmf.SchemaEventClass("org.apache.qpid.qmf", "test_event", qmf.SEV_NOTICE)
+        self.event_class.add_argument(qmf.SchemaArgument("uint32val", qmf.TYPE_UINT32))
+        self.event_class.add_argument(qmf.SchemaArgument("strval", qmf.TYPE_LSTR))
+
 
     def register(self, agent):
         agent.register_class(self.parent_class)
         agent.register_class(self.child_class)
+        agent.register_class(self.event_class)
 
 
 
@@ -151,6 +156,12 @@ class App(qmf.AgentHandler):
                                                  'sstr' : "Short String",
                                                  'map'  : {'first' : 'FIRST', 'second' : 'SECOND'}})
 
+                event = qmf.QmfEvent(self._model.event_class)
+                event.uint32val = self._parent.get_attr("uint32val")
+                event.strval = "Unused"
+
+                self._agent.raise_event(event)
+
                 ## Test the __getattr__ implementation:
                 ## @todo: remove once python_client implements this
                 ## form of property access
@@ -169,6 +180,11 @@ class App(qmf.AgentHandler):
                 self._parent.set_attr("int16val", 10)
                 self._parent.set_attr("int8val",  11)
 
+                event = qmf.QmfEvent(self._model.event_class)
+                event.uint32val = self._parent.uint32val
+                event.strval = "Unused"
+                self._agent.raise_event(event)
+
             elif args['test'] == "negative":
                 self._parent.set_attr("uint64val", 0)
                 self._parent.set_attr("uint32val", 0)
@@ -180,6 +196,11 @@ class App(qmf.AgentHandler):
                 self._parent.set_attr("int16val", -1000)
                 self._parent.set_attr("int8val",  -100)
 
+                event = qmf.QmfEvent(self._model.event_class)
+                event.uint32val = self._parent.uint32val
+                event.strval = "Unused"
+                self._agent.raise_event(event)
+
             else:
                 _retCode = 1
                 _retText = "Invalid argument value for test"
@@ -188,10 +209,20 @@ class App(qmf.AgentHandler):
 
         elif name == "set_short_string":
             self._parent.set_attr('sstrval', args['value'])
+            event = qmf.QmfEvent(self._model.event_class)
+            event.uint32val = 0
+            event.strval = self._parent.sstrval
+            self._agent.raise_event(event)
+
             self._agent.method_response(context, 0, "OK", args)
 
         elif name == "set_long_string":
             self._parent.set_attr('lstrval', args['value'])
+            event = qmf.QmfEvent(self._model.event_class)
+            event.uint32val = 0
+            event.strval = self._parent.lstrval
+            self._agent.raise_event(event)
+
             self._agent.method_response(context, 0, "OK", args)
 
         elif name == "create_child":
diff --git a/qpid/cpp/bindings/qmf/tests/python_console.py b/qpid/cpp/bindings/qmf/tests/python_console.py
index 6f366d1a3d..fe27c33c4b 100755
--- a/qpid/cpp/bindings/qmf/tests/python_console.py
+++ b/qpid/cpp/bindings/qmf/tests/python_console.py
@@ -23,6 +23,7 @@ from qpid.testlib import TestBase010
 from qpid.datatypes import Message
 from qpid.queue import Empty
 from time import sleep
+import qmf.console
 
 class QmfInteropTests(TestBase010):
 
@@ -166,6 +167,43 @@ class QmfInteropTests(TestBase010):
         for binding in bindings:
             self.assertEqual(binding.exchangeRef, mgmt_exchange.getObjectId())
 
+    def test_F_events(self):
+        class Handler(qmf.console.Console):
+            def __init__(self):
+                self.queue = []
+
+            def event(self, broker, event):
+                if event.getClassKey().getClassName() == "test_event":
+                    self.queue.append(event)
+
+        handler = Handler()
+        self.startQmf(handler)
+
+        parents = self.qmf.getObjects(_class="parent")
+        self.assertEqual(len(parents), 1)
+        parent = parents[0]
+
+        parent.set_numerics("big")
+        parent.set_numerics("small")
+        parent.set_numerics("negative")
+        parent.set_short_string("TEST")
+        parent.set_long_string("LONG_TEST")
+        parent.probe_userid()
+
+        queue = handler.queue
+        self.assertEqual(len(queue), 5)
+        self.assertEqual(queue[0].arguments["uint32val"], 0xA5A55A5A)
+        self.assertEqual(queue[0].arguments["strval"], "Unused")
+        self.assertEqual(queue[1].arguments["uint32val"], 5)
+        self.assertEqual(queue[1].arguments["strval"], "Unused")
+        self.assertEqual(queue[2].arguments["uint32val"], 0)
+        self.assertEqual(queue[2].arguments["strval"], "Unused")
+        self.assertEqual(queue[3].arguments["uint32val"], 0)
+        self.assertEqual(queue[3].arguments["strval"], "TEST")
+        self.assertEqual(queue[4].arguments["uint32val"], 0)
+        self.assertEqual(queue[4].arguments["strval"], "LONG_TEST")
+        
+
     def getProperty(self, msg, name):
         for h in msg.headers:
             if hasattr(h, name): return getattr(h, name)
diff --git a/qpid/cpp/include/qmf/engine/Event.h b/qpid/cpp/include/qmf/engine/Event.h
index 50ab5c1200..4e5200a0a8 100644
--- a/qpid/cpp/include/qmf/engine/Event.h
+++ b/qpid/cpp/include/qmf/engine/Event.h
@@ -23,7 +23,24 @@
 namespace qmf {
 namespace engine {
 
+    class SchemaEventClass;
+    class Value;
+    struct EventImpl;
+
     class Event {
+    public:
+        Event(const SchemaEventClass* type);
+        Event(const Event& from);
+        ~Event();
+
+        const SchemaEventClass* getClass() const;
+        Value* getValue(const char* key) const;
+
+    private:
+        friend struct EventImpl;
+        friend struct AgentImpl;
+        Event(EventImpl* impl);
+        EventImpl* impl;
     };
 }
 }
diff --git a/qpid/cpp/include/qmf/engine/Schema.h b/qpid/cpp/include/qmf/engine/Schema.h
index 9f5b444558..f53e84324a 100644
--- a/qpid/cpp/include/qmf/engine/Schema.h
+++ b/qpid/cpp/include/qmf/engine/Schema.h
@@ -29,6 +29,7 @@ namespace engine {
     enum Access { ACCESS_READ_CREATE = 1, ACCESS_READ_WRITE = 2, ACCESS_READ_ONLY = 3 };
     enum Direction { DIR_IN = 1, DIR_OUT = 2, DIR_IN_OUT = 3 };
     enum ClassKind { CLASS_OBJECT = 1, CLASS_EVENT = 2 };
+    enum Severity { SEV_EMERG = 0, SEV_ALERT = 1, SEV_CRIT = 2, SEV_ERROR = 3, SEV_WARN = 4, SEV_NOTICE = 5, SEV_INFORM = 6, SEV_DEBUG = 7 };
 
     struct SchemaArgumentImpl;
     struct SchemaMethodImpl;
@@ -186,13 +187,14 @@ namespace engine {
      */
     class SchemaEventClass {
     public:
-        SchemaEventClass(const char* package, const char* name);
+        SchemaEventClass(const char* package, const char* name, Severity severity);
         SchemaEventClass(const SchemaEventClass& from);
         ~SchemaEventClass();
         void addArgument(const SchemaArgument* argument);
         void setDesc(const char* desc);
 
         const SchemaClassKey* getClassKey() const;
+        Severity getSeverity() const;
         int getArgumentCount() const;
         const SchemaArgument* getArgument(int idx) const;
 
diff --git a/qpid/cpp/include/qmf/engine/Value.h b/qpid/cpp/include/qmf/engine/Value.h
index 8eae382caf..5b45061b78 100644
--- a/qpid/cpp/include/qmf/engine/Value.h
+++ b/qpid/cpp/include/qmf/engine/Value.h
@@ -110,6 +110,7 @@ namespace engine {
         friend struct ValueImpl;
         friend class  BrokerProxyImpl;
         friend struct ObjectImpl;
+        friend struct EventImpl;
         friend class  AgentImpl;
         Value(ValueImpl* impl);
         ValueImpl* impl;
diff --git a/qpid/cpp/src/qmf.mk b/qpid/cpp/src/qmf.mk
index 96a977f3cd..1e4c59b19e 100644
--- a/qpid/cpp/src/qmf.mk
+++ b/qpid/cpp/src/qmf.mk
@@ -73,6 +73,8 @@ libqmfengine_la_SOURCES =			\
   qmf/engine/ConnectionSettingsImpl.h		\
   qmf/engine/ConsoleImpl.cpp			\
   qmf/engine/ConsoleImpl.h			\
+  qmf/engine/EventImpl.cpp			\
+  qmf/engine/EventImpl.h			\
   qmf/engine/MessageImpl.cpp			\
   qmf/engine/MessageImpl.h			\
   qmf/engine/ObjectIdImpl.cpp			\
diff --git a/qpid/cpp/src/qmf/engine/Agent.cpp b/qpid/cpp/src/qmf/engine/Agent.cpp
index c5d1bff2e0..fe9b84c565 100644
--- a/qpid/cpp/src/qmf/engine/Agent.cpp
+++ b/qpid/cpp/src/qmf/engine/Agent.cpp
@@ -21,6 +21,7 @@
 #include "qmf/engine/MessageImpl.h"
 #include "qmf/engine/SchemaImpl.h"
 #include "qmf/engine/Typecode.h"
+#include "qmf/engine/EventImpl.h"
 #include "qmf/engine/ObjectImpl.h"
 #include "qmf/engine/ObjectIdImpl.h"
 #include "qmf/engine/QueryImpl.h"
@@ -476,9 +477,19 @@ const ObjectId* AgentImpl::allocObjectId(uint32_t persistIdLo, uint32_t persistI
     return allocObjectId(((uint64_t) persistIdHi) << 32 | (uint64_t) persistIdLo);
 }
 
-void AgentImpl::raiseEvent(Event&)
+void AgentImpl::raiseEvent(Event& event)
 {
     Mutex::ScopedLock _lock(lock);
+    Buffer buffer(outputBuffer, MA_BUFFER_SIZE);
+    Protocol::encodeHeader(buffer, Protocol::OP_EVENT_INDICATION);
+
+    event.impl->encodeSchemaKey(buffer);
+    buffer.putLongLong(uint64_t(Duration(now())));
+    event.impl->encode(buffer);
+    string key(event.impl->getRoutingKey(assignedBrokerBank, assignedAgentBank));
+
+    sendBufferLH(buffer, QMF_EXCHANGE, key);
+    QPID_LOG(trace, "SENT EventIndication");
 }
 
 AgentEventImpl::Ptr AgentImpl::eventDeclareQueue(const string& name)
diff --git a/qpid/cpp/src/qmf/engine/EventImpl.cpp b/qpid/cpp/src/qmf/engine/EventImpl.cpp
new file mode 100644
index 0000000000..6bdda9321e
--- /dev/null
+++ b/qpid/cpp/src/qmf/engine/EventImpl.cpp
@@ -0,0 +1,106 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * 
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include <qmf/engine/EventImpl.h>
+#include <qmf/engine/ValueImpl.h>
+
+using namespace std;
+using namespace qmf::engine;
+using qpid::framing::Buffer;
+
+EventImpl::EventImpl(const SchemaEventClass* type) : eventClass(type)
+{
+    int argCount = eventClass->getArgumentCount();
+    int idx;
+
+    for (idx = 0; idx < argCount; idx++) {
+        const SchemaArgument* arg = eventClass->getArgument(idx);
+        arguments[arg->getName()] = ValuePtr(new Value(arg->getType()));
+    }
+}
+
+
+EventImpl::EventImpl(const SchemaEventClass* type, Buffer&) :
+    eventClass(type)
+{
+}
+
+
+Event* EventImpl::factory(const SchemaEventClass* type, Buffer& buffer)
+{
+    EventImpl* impl(new EventImpl(type, buffer));
+    return new Event(impl);
+}
+
+
+Value* EventImpl::getValue(const char* key) const
+{
+    map<string, ValuePtr>::const_iterator iter;
+
+    iter = arguments.find(key);
+    if (iter != arguments.end())
+        return iter->second.get();
+
+    return 0;
+}
+
+
+void EventImpl::encodeSchemaKey(Buffer& buffer) const
+{
+    buffer.putShortString(eventClass->getClassKey()->getPackageName());
+    buffer.putShortString(eventClass->getClassKey()->getClassName());
+    buffer.putBin128(const_cast<uint8_t*>(eventClass->getClassKey()->getHash()));
+}
+
+
+void EventImpl::encode(Buffer& buffer) const
+{
+    buffer.putOctet((uint8_t) eventClass->getSeverity());
+
+    int argCount = eventClass->getArgumentCount();
+    for (int idx = 0; idx < argCount; idx++) {
+        const SchemaArgument* arg = eventClass->getArgument(idx);
+        ValuePtr value = arguments[arg->getName()];
+        value->impl->encode(buffer);
+    }
+}
+
+
+string EventImpl::getRoutingKey(uint32_t brokerBank, uint32_t agentBank) const
+{
+    stringstream key;
+
+    key << "console.event." << brokerBank << "." << agentBank << "." <<
+        eventClass->getClassKey()->getPackageName() << "." <<
+        eventClass->getClassKey()->getClassName();
+    return key.str();
+}
+
+
+//==================================================================
+// Wrappers
+//==================================================================
+
+Event::Event(const SchemaEventClass* type) : impl(new EventImpl(type)) {}
+Event::Event(EventImpl* i) : impl(i) {}
+Event::Event(const Event& from) : impl(new EventImpl(*(from.impl))) {}
+Event::~Event() { delete impl; }
+const SchemaEventClass* Event::getClass() const { return impl->getClass(); }
+Value* Event::getValue(const char* key) const { return impl->getValue(key); }
+
diff --git a/qpid/cpp/src/qmf/engine/EventImpl.h b/qpid/cpp/src/qmf/engine/EventImpl.h
new file mode 100644
index 0000000000..dfdf64e848
--- /dev/null
+++ b/qpid/cpp/src/qmf/engine/EventImpl.h
@@ -0,0 +1,53 @@
+#ifndef _QmfEngineEventImpl_
+#define _QmfEngineEventImpl_
+
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * 
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include <qmf/engine/Event.h>
+#include <qmf/engine/Schema.h>
+#include <qpid/framing/Buffer.h>
+#include <boost/shared_ptr.hpp>
+#include <map>
+
+namespace qmf {
+namespace engine {
+
+    struct EventImpl {
+        typedef boost::shared_ptr<Value> ValuePtr;
+        const SchemaEventClass* eventClass;
+        mutable std::map<std::string, ValuePtr> arguments;
+
+        EventImpl(const SchemaEventClass* type);
+        EventImpl(const SchemaEventClass* type, qpid::framing::Buffer& buffer);
+        static Event* factory(const SchemaEventClass* type, qpid::framing::Buffer& buffer);
+
+        const SchemaEventClass* getClass() const { return eventClass; }
+        Value* getValue(const char* key) const;
+
+        void encodeSchemaKey(qpid::framing::Buffer& buffer) const;
+        void encode(qpid::framing::Buffer& buffer) const;
+        std::string getRoutingKey(uint32_t brokerBank, uint32_t agentBank) const;
+    };
+
+}
+}
+
+#endif
+
diff --git a/qpid/cpp/src/qmf/engine/ResilientConnection.cpp b/qpid/cpp/src/qmf/engine/ResilientConnection.cpp
index 53524fdbd8..9c19e4d460 100644
--- a/qpid/cpp/src/qmf/engine/ResilientConnection.cpp
+++ b/qpid/cpp/src/qmf/engine/ResilientConnection.cpp
@@ -39,6 +39,8 @@
 #include <set>
 #include <boost/intrusive_ptr.hpp>
 #include <boost/noncopyable.hpp>
+#include <unistd.h>
+#include <fcntl.h>
 
 using namespace std;
 using namespace qmf::engine;
@@ -330,6 +332,10 @@ void ResilientConnectionImpl::unbind(SessionHandle handle,
 void ResilientConnectionImpl::setNotifyFd(int fd)
 {
     notifyFd = fd;
+    if (notifyFd > 0) {
+        int original = fcntl(notifyFd, F_GETFL);
+        fcntl(notifyFd, F_SETFL, O_NONBLOCK | original);
+    }
 }
 
 void ResilientConnectionImpl::run()
@@ -403,13 +409,16 @@ void ResilientConnectionImpl::EnqueueEvent(ResilientConnectionEvent::EventKind k
                                            const MessageImpl& message,
                                            const string& errorText)
 {
-    Mutex::ScopedLock _lock(lock);
-    ResilientConnectionEventImpl event(kind, message);
+    {
+        Mutex::ScopedLock _lock(lock);
+        ResilientConnectionEventImpl event(kind, message);
 
-    event.sessionContext = sessionContext;
-    event.errorText      = errorText;
+        event.sessionContext = sessionContext;
+        event.errorText      = errorText;
+
+        eventQueue.push_back(event);
+    }
 
-    eventQueue.push_back(event);
     if (notifyFd != -1)
     {
         int unused_ret;    //Suppress warnings about ignoring return value.
diff --git a/qpid/cpp/src/qmf/engine/SchemaImpl.cpp b/qpid/cpp/src/qmf/engine/SchemaImpl.cpp
index c37ec34890..249a08ba7f 100644
--- a/qpid/cpp/src/qmf/engine/SchemaImpl.cpp
+++ b/qpid/cpp/src/qmf/engine/SchemaImpl.cpp
@@ -459,7 +459,6 @@ SchemaEventClassImpl::SchemaEventClassImpl(Buffer& buffer) : hasHash(true), clas
     buffer.getShortString(package);
     buffer.getShortString(name);
     hash.decode(buffer);
-    buffer.putOctet(0); // No parent class
 
     uint16_t argCount = buffer.getShort();
 
@@ -598,13 +597,14 @@ const SchemaProperty* SchemaObjectClass::getProperty(int idx) const { return imp
 const SchemaStatistic* SchemaObjectClass::getStatistic(int idx) const { return impl->getStatistic(idx); }
 const SchemaMethod* SchemaObjectClass::getMethod(int idx) const { return impl->getMethod(idx); }
 
-SchemaEventClass::SchemaEventClass(const char* package, const char* name) : impl(new SchemaEventClassImpl(package, name)) {}
+SchemaEventClass::SchemaEventClass(const char* package, const char* name, Severity s) : impl(new SchemaEventClassImpl(package, name, s)) {}
 SchemaEventClass::SchemaEventClass(SchemaEventClassImpl* i) : impl(i) {}
 SchemaEventClass::SchemaEventClass(const SchemaEventClass& from) : impl(new SchemaEventClassImpl(*(from.impl))) {}
 SchemaEventClass::~SchemaEventClass() { delete impl; }
 void SchemaEventClass::addArgument(const SchemaArgument* argument) { impl->addArgument(argument); }
 void SchemaEventClass::setDesc(const char* desc) { impl->setDesc(desc); }
 const SchemaClassKey* SchemaEventClass::getClassKey() const { return impl->getClassKey(); }
+Severity SchemaEventClass::getSeverity() const { return impl->getSeverity(); }
 int SchemaEventClass::getArgumentCount() const { return impl->getArgumentCount(); }
 const SchemaArgument* SchemaEventClass::getArgument(int idx) const { return impl->getArgument(idx); }
 
diff --git a/qpid/cpp/src/qmf/engine/SchemaImpl.h b/qpid/cpp/src/qmf/engine/SchemaImpl.h
index af3a1d98e4..7be757ee8d 100644
--- a/qpid/cpp/src/qmf/engine/SchemaImpl.h
+++ b/qpid/cpp/src/qmf/engine/SchemaImpl.h
@@ -201,10 +201,11 @@ namespace engine {
         mutable bool hasHash;
         std::auto_ptr<SchemaClassKey> classKey;
         std::string description;
+        Severity severity;
         std::vector<const SchemaArgument*> arguments;
 
-        SchemaEventClassImpl(const char* p, const char* n) :
-            package(p), name(n), hasHash(false), classKey(SchemaClassKeyImpl::factory(package, name, hash)) {}
+    SchemaEventClassImpl(const char* p, const char* n, Severity sev) :
+        package(p), name(n), hasHash(false), classKey(SchemaClassKeyImpl::factory(package, name, hash)), severity(sev) {}
         SchemaEventClassImpl(qpid::framing::Buffer& buffer);
         static SchemaEventClass* factory(qpid::framing::Buffer& buffer);
 
@@ -213,6 +214,7 @@ namespace engine {
         void setDesc(const char* desc) { description = desc; }
 
         const SchemaClassKey* getClassKey() const;
+        Severity getSeverity() const { return severity; }
         int getArgumentCount() const { return arguments.size(); }
         const SchemaArgument* getArgument(int idx) const;
     };