You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by tr...@apache.org on 2008/10/07 23:47:36 UTC
svn commit: r702651 [1/2] - in /incubator/qpid/trunk/qpid:
cpp/examples/qmf-agent/ cpp/managementgen/ cpp/managementgen/qmf/
cpp/managementgen/qmf/templates/ cpp/src/qpid/acl/ cpp/src/qpid/agent/
cpp/src/qpid/broker/ cpp/src/qpid/framing/ cpp/src/qpid/...
Author: tross
Date: Tue Oct 7 14:47:35 2008
New Revision: 702651
URL: http://svn.apache.org/viewvc?rev=702651&view=rev
Log:
QPID-1327 - Event support for Management
Added:
incubator/qpid/trunk/qpid/cpp/managementgen/qmf/templates/Event.cpp
incubator/qpid/trunk/qpid/cpp/managementgen/qmf/templates/Event.h
incubator/qpid/trunk/qpid/cpp/src/qpid/management/ManagementEvent.h
Modified:
incubator/qpid/trunk/qpid/cpp/examples/qmf-agent/example.cpp
incubator/qpid/trunk/qpid/cpp/examples/qmf-agent/schema.xml
incubator/qpid/trunk/qpid/cpp/managementgen/qmf-gen
incubator/qpid/trunk/qpid/cpp/managementgen/qmf/generate.py
incubator/qpid/trunk/qpid/cpp/managementgen/qmf/management-types.xml
incubator/qpid/trunk/qpid/cpp/managementgen/qmf/schema.py
incubator/qpid/trunk/qpid/cpp/managementgen/qmf/templates/Class.cpp
incubator/qpid/trunk/qpid/cpp/managementgen/qmf/templates/Class.h
incubator/qpid/trunk/qpid/cpp/src/qpid/acl/Acl.cpp
incubator/qpid/trunk/qpid/cpp/src/qpid/acl/Acl.h
incubator/qpid/trunk/qpid/cpp/src/qpid/acl/management-schema.xml
incubator/qpid/trunk/qpid/cpp/src/qpid/agent/ManagementAgent.h
incubator/qpid/trunk/qpid/cpp/src/qpid/agent/ManagementAgentImpl.cpp
incubator/qpid/trunk/qpid/cpp/src/qpid/agent/ManagementAgentImpl.h
incubator/qpid/trunk/qpid/cpp/src/qpid/broker/Connection.cpp
incubator/qpid/trunk/qpid/cpp/src/qpid/broker/ConnectionState.h
incubator/qpid/trunk/qpid/cpp/src/qpid/broker/SessionAdapter.cpp
incubator/qpid/trunk/qpid/cpp/src/qpid/framing/Buffer.h
incubator/qpid/trunk/qpid/cpp/src/qpid/management/Manageable.h
incubator/qpid/trunk/qpid/cpp/src/qpid/management/ManagementBroker.cpp
incubator/qpid/trunk/qpid/cpp/src/qpid/management/ManagementBroker.h
incubator/qpid/trunk/qpid/cpp/src/qpid/management/ManagementObject.cpp
incubator/qpid/trunk/qpid/cpp/src/qpid/management/ManagementObject.h
incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/management/Names.java
incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/management/Protocol.java
incubator/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/management/domain/handler/impl/SchemaResponseMessageHandler.java
incubator/qpid/trunk/qpid/python/commands/qpid-config
incubator/qpid/trunk/qpid/python/commands/qpid-printevents
incubator/qpid/trunk/qpid/python/commands/qpid-route
incubator/qpid/trunk/qpid/python/qpid/management.py
incubator/qpid/trunk/qpid/python/qpid/managementdata.py
incubator/qpid/trunk/qpid/python/qpid/qmfconsole.py
incubator/qpid/trunk/qpid/python/tests_0-10/management.py
incubator/qpid/trunk/qpid/specs/management-schema.xml
Modified: incubator/qpid/trunk/qpid/cpp/examples/qmf-agent/example.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/examples/qmf-agent/example.cpp?rev=702651&r1=702650&r2=702651&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/examples/qmf-agent/example.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/examples/qmf-agent/example.cpp Tue Oct 7 14:47:35 2008
@@ -26,6 +26,7 @@
#include "qmf/org/apache/qpid/agent/example/Parent.h"
#include "qmf/org/apache/qpid/agent/example/Child.h"
#include "qmf/org/apache/qpid/agent/example/ArgsParentCreate_child.h"
+#include "qmf/org/apache/qpid/agent/example/EventChildCreated.h"
#include "qmf/org/apache/qpid/agent/example/Package.h"
#include <unistd.h>
@@ -129,7 +130,7 @@
children.push_back(child);
- mgmtObject->event_childCreated(ioArgs.i_name);
+ agent->raiseEvent(_qmf::EventChildCreated(ioArgs.i_name));
return STATUS_OK;
}
Modified: incubator/qpid/trunk/qpid/cpp/examples/qmf-agent/schema.xml
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/examples/qmf-agent/schema.xml?rev=702651&r1=702650&r2=702651&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/examples/qmf-agent/schema.xml (original)
+++ incubator/qpid/trunk/qpid/cpp/examples/qmf-agent/schema.xml Tue Oct 7 14:47:35 2008
@@ -37,14 +37,6 @@
<arg name="name" dir="I" type="sstr"/>
<arg name="childRef" dir="O" type="objId"/>
</method>
-
- <event name="childCreated">
- <arg name="name" type="sstr"/>
- </event>
-
- <event name="childDestroyed">
- <arg name="name" type="sstr"/>
- </event>
</class>
@@ -59,7 +51,14 @@
<statistic name="count" type="count64" unit="tick" desc="Counter that increases monotonically"/>
- <method name="delete"/>
+ <method name="delete"/>
</class>
+
+ <eventArguments>
+ <arg name="childName" type="sstr"/>
+ </eventArguments>
+
+ <event name="ChildCreated" args="childName"/>
+ <event name="ChildDestroyed" args="childName"/>
</schema>
Modified: incubator/qpid/trunk/qpid/cpp/managementgen/qmf-gen
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/managementgen/qmf-gen?rev=702651&r1=702650&r2=702651&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/managementgen/qmf-gen (original)
+++ incubator/qpid/trunk/qpid/cpp/managementgen/qmf-gen Tue Oct 7 14:47:35 2008
@@ -58,6 +58,8 @@
gen.makeClassFiles ("Class.h", package)
gen.makeClassFiles ("Class.cpp", package)
gen.makeMethodFiles ("Args.h", package)
+ gen.makeEventFiles ("Event.h", package)
+ gen.makeEventFiles ("Event.cpp", package)
gen.makePackageFile ("Package.h", package)
gen.makePackageFile ("Package.cpp", package)
Modified: incubator/qpid/trunk/qpid/cpp/managementgen/qmf/generate.py
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/managementgen/qmf/generate.py?rev=702651&r1=702650&r2=702651&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/managementgen/qmf/generate.py (original)
+++ incubator/qpid/trunk/qpid/cpp/managementgen/qmf/generate.py Tue Oct 7 14:47:35 2008
@@ -250,6 +250,14 @@
path = self.packagePath + _class.getNameCap () + extension
return path
+ def targetEventFile (self, event, templateFile):
+ dot = templateFile.find(".")
+ if dot == -1:
+ raise ValueError ("Invalid template file name %s" % templateFile)
+ extension = templateFile[dot:len (templateFile)]
+ path = self.packagePath + "Event" + event.getNameCap () + extension
+ return path
+
def targetMethodFile (self, method, templateFile):
""" Return the file name for a method file """
dot = templateFile.rfind(".")
@@ -293,6 +301,16 @@
stream = template.expand (_class)
self.writeIfChanged (stream, target, force)
+ def makeEventFiles (self, templateFile, schema, force=False):
+ """ Generate an expanded template per schema event """
+ events = schema.getEvents()
+ template = Template (self.input + templateFile, self)
+ self.templateFiles.append (templateFile)
+ for event in events:
+ target = self.targetEventFile(event, templateFile)
+ stream = template.expand(event)
+ self.writeIfChanged(stream, target, force)
+
def makeMethodFiles (self, templateFile, schema, force=False):
""" Generate an expanded template per method-with-arguments """
classes = schema.getClasses ()
Modified: incubator/qpid/trunk/qpid/cpp/managementgen/qmf/management-types.xml
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/managementgen/qmf/management-types.xml?rev=702651&r1=702650&r2=702651&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/managementgen/qmf/management-types.xml (original)
+++ incubator/qpid/trunk/qpid/cpp/managementgen/qmf/management-types.xml Tue Oct 7 14:47:35 2008
@@ -30,7 +30,7 @@
<type name="int64" base="S64" cpp="int64_t" encode="@.putInt64(#)" decode="# = @.getInt64()" accessor="direct" init="0"/>
<type name="bool" base="BOOL" cpp="uint8_t" encode="@.putOctet(#?1:0)" decode="# = @.getOctet()==1" accessor="direct" init="0"/>
<type name="sstr" base="SSTR" cpp="std::string" encode="@.putShortString(#)" decode="@.getShortString(#)" accessor="direct" init='""' byRef="y"/>
-<type name="lstr" base="LSTR" cpp="std::string" encode="@.putLongString(#)" decode="@.getLongString(#)" accessor="direct" init='""' byRef="y"/>
+<type name="lstr" base="LSTR" cpp="std::string" encode="@.putMediumString(#)" decode="@.getMediumString(#)" accessor="direct" init='""' byRef="y"/>
<type name="absTime" base="ABSTIME" cpp="int64_t" encode="@.putLongLong(#)" decode="# = @.getLongLong()" accessor="direct" init="0"/>
<type name="deltaTime" base="DELTATIME" cpp="uint64_t" encode="@.putLongLong(#)" decode="# = @.getLongLong()" accessor="direct" init="0"/>
<type name="float" base="FLOAT" cpp="float" encode="@.putFloat(#)" decode="# = @.getFloat()" accessor="direct" init="0."/>
Modified: incubator/qpid/trunk/qpid/cpp/managementgen/qmf/schema.py
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/managementgen/qmf/schema.py?rev=702651&r1=702650&r2=702651&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/managementgen/qmf/schema.py (original)
+++ incubator/qpid/trunk/qpid/cpp/managementgen/qmf/schema.py Tue Oct 7 14:47:35 2008
@@ -21,6 +21,32 @@
from cStringIO import StringIO
import md5
+class Hash:
+ """ Manage the hash of an XML sub-tree """
+ def __init__(self, node):
+ self.md5Sum = md5.new()
+ self._compute(node)
+
+ def addSubHash(self, hash):
+ """ Use this method to add the hash of a dependend-on XML fragment that is not in the sub-tree """
+ self.md5Sum.update(hash.getDigest())
+
+ def getDigest(self):
+ return self.md5Sum.digest()
+
+ def _compute(self, node):
+ attrs = node.attributes
+ self.md5Sum.update(node.nodeName)
+
+ for idx in range(attrs.length):
+ self.md5Sum.update(attrs.item(idx).nodeName)
+ self.md5Sum.update(attrs.item(idx).nodeValue)
+
+ for child in node.childNodes:
+ if child.nodeType == Node.ELEMENT_NODE:
+ self._compute(child)
+
+
#=====================================================================================
#
#=====================================================================================
@@ -525,6 +551,7 @@
self.maxLen = None
self.desc = None
self.default = None
+ self.hash = Hash(node)
attrs = node.attributes
for idx in range (attrs.length):
@@ -675,12 +702,12 @@
#
#=====================================================================================
class SchemaEvent:
- def __init__ (self, parent, node, typespec):
- self.parent = parent
- self.name = None
- self.desc = None
- self.args = []
- self.defaultSeverity = None
+ def __init__ (self, package, node, typespec, argset):
+ self.packageName = package
+ self.name = None
+ self.desc = None
+ self.args = []
+ self.hash = Hash(node)
attrs = node.attributes
for idx in range (attrs.length):
@@ -692,72 +719,96 @@
elif key == 'desc':
self.desc = val
- elif key == 'defaultSeverity':
- self.defaultSeverity = val
+ elif key == 'args':
+ list = val.replace(" ", "").split(",")
+ for item in list:
+ if item not in argset.args:
+ raise Exception("undefined argument '%s' in event" % item)
+ self.args.append(argset.args[item])
+ self.hash.addSubHash(argset.args[item].hash)
else:
raise ValueError ("Unknown attribute in event '%s'" % key)
- for child in node.childNodes:
- if child.nodeType == Node.ELEMENT_NODE:
- if child.nodeName == 'arg':
- arg = SchemaArg (child, typespec)
- self.args.append (arg)
- else:
- raise ValueError ("Unknown event tag '%s'" % child.nodeName)
-
def getName (self):
return self.name
+ def getNameCap(self):
+ return capitalize(self.name)
+
def getFullName (self):
- return capitalize(self.parent.getName()) + capitalize(self.name)
+ return capitalize(self.package + capitalize(self.name))
def getArgCount (self):
return len (self.args)
- def genMethodBody (self, stream, variables, classObject):
- stream.write("void ")
- classObject.genNameCap(stream, variables)
- stream.write("::event_%s(" % self.name)
- count = 0
+ def genArgCount (self, stream, variables):
+ stream.write("%d" % len(self.args))
+
+ def genArgDeclarations(self, stream, variables):
for arg in self.args:
- arg.genFormalParam(stream, variables)
- count += 1
- if count < len(self.args):
- stream.write(", ")
- stream.write(") {\n")
- stream.write(" ::qpid::sys::Mutex::ScopedLock mutex(getMutex());\n")
- stream.write(" Buffer* buf = startEventLH();\n")
- stream.write(" objectId.encode(*buf);\n")
- stream.write(" buf->putShortString(packageName);\n")
- stream.write(" buf->putShortString(className);\n")
- stream.write(" buf->putBin128(md5Sum);\n")
- stream.write(" buf->putShortString(\"%s\");\n" % self.name)
+ if arg.type.type.byRef:
+ ref = "&"
+ else:
+ ref = ""
+ stream.write(" const %s%s %s;\n" % (arg.type.type.cpp, ref, arg.name))
+
+ def genCloseNamespaces (self, stream, variables):
+ for item in self.packageName.split("."):
+ stream.write ("}")
+
+ def genConstructorArgs(self, stream, variables):
+ pre = ""
for arg in self.args:
- stream.write(" %s;\n" % arg.type.type.encode.replace("@", "(*buf)").replace("#", "_" + arg.name))
- stream.write(" finishEventLH(buf);\n")
- stream.write("}\n\n")
+ if arg.type.type.byRef:
+ ref = "&"
+ else:
+ ref = ""
+ stream.write("%sconst %s%s _%s" % (pre, arg.type.type.cpp, ref, arg.name))
+ pre = ",\n "
- def genMethodDecl (self, stream, variables):
- stream.write(" void event_%s(" % self.name)
- count = 0
+ def genConstructorInits(self, stream, variables):
+ pre = ""
for arg in self.args:
- arg.genFormalParam(stream, variables)
- count += 1
- if count < len(self.args):
- stream.write(", ")
- stream.write(");\n")
+ stream.write("%s%s(_%s)" % (pre, arg.name, arg.name))
+ pre = ",\n "
- def genSchema(self, stream, variables):
- stream.write (" ft = FieldTable ();\n")
- stream.write (" ft.setString (NAME, \"" + self.name + "\");\n")
- stream.write (" ft.setInt (ARGCOUNT, " + str (len (self.args)) + ");\n")
- if self.desc != None:
- stream.write (" ft.setString (DESC, \"" + self.desc + "\");\n")
- stream.write (" buf.put (ft);\n\n")
+ def genName(self, stream, variables):
+ stream.write(self.name)
+
+ def genNameCap(self, stream, variables):
+ stream.write(capitalize(self.name))
+
+ def genNamespace (self, stream, variables):
+ stream.write("::".join(self.packageName.split(".")))
+
+ def genNameLower(self, stream, variables):
+ stream.write(self.name.lower())
+
+ def genNameUpper(self, stream, variables):
+ stream.write(self.name.upper())
+
+ def genNamePackageLower(self, stream, variables):
+ stream.write(self.packageName.lower())
+
+ def genOpenNamespaces (self, stream, variables):
+ for item in self.packageName.split("."):
+ stream.write ("namespace %s {\n" % item)
+
+ def genArgEncodes(self, stream, variables):
for arg in self.args:
- arg.genSchema (stream, True)
+ stream.write(" " + arg.type.type.encode.replace("@", "buf").replace("#", arg.name) + ";\n")
+ def genArgSchema(self, stream, variables):
+ for arg in self.args:
+ arg.genSchema(stream, True)
+
+ def genSchemaMD5(self, stream, variables):
+ sum = self.hash.getDigest()
+ for idx in range (len (sum)):
+ if idx != 0:
+ stream.write (",")
+ stream.write (hex (ord (sum[idx])))
class SchemaClass:
@@ -768,9 +819,7 @@
self.methods = []
self.events = []
self.options = options
- self.md5Sum = md5.new ()
-
- self.hash (node)
+ self.hash = Hash(node)
attrs = node.attributes
self.name = makeValidCppSymbol(attrs['name'].nodeValue)
@@ -790,10 +839,6 @@
sub = SchemaMethod (self, child, typespec)
self.methods.append (sub)
- elif child.nodeName == 'event':
- sub = SchemaEvent (self, child, typespec)
- self.events.append (sub)
-
elif child.nodeName == 'group':
self.expandFragment (child, fragments)
@@ -820,24 +865,12 @@
result = result[0:pos] + "threadStats->" + result[pos:]
start = pos + 9 + len(next[1])
- def hash (self, node):
- attrs = node.attributes
- self.md5Sum.update (node.nodeName)
-
- for idx in range (attrs.length):
- self.md5Sum.update (attrs.item(idx).nodeName)
- self.md5Sum.update (attrs.item(idx).nodeValue)
-
- for child in node.childNodes:
- if child.nodeType == Node.ELEMENT_NODE:
- self.hash (child)
-
def expandFragment (self, node, fragments):
attrs = node.attributes
name = attrs['name'].nodeValue
for fragment in fragments:
if fragment.name == name:
- self.md5Sum.update (fragment.md5Sum.digest())
+ self.hash.addSubHash(fragment.hash)
for config in fragment.properties:
self.properties.append (config)
for inst in fragment.statistics:
@@ -937,27 +970,12 @@
inArgCount = inArgCount + 1
if methodCount == 0:
- stream.write ("string, Buffer&, Buffer& outBuf")
+ stream.write ("string&, Buffer&, Buffer& outBuf")
else:
if inArgCount == 0:
- stream.write ("string methodName, Buffer&, Buffer& outBuf")
+ stream.write ("string& methodName, Buffer&, Buffer& outBuf")
else:
- stream.write ("string methodName, Buffer& inBuf, Buffer& outBuf")
-
- def genEventCount (self, stream, variables):
- stream.write ("%d" % len (self.events))
-
- def genEventMethodBodies (self, stream, variables):
- for event in self.events:
- event.genMethodBody (stream, variables, self)
-
- def genEventMethodDecls (self, stream, variables):
- for event in self.events:
- event.genMethodDecl (stream, variables)
-
- def genEventSchema (self, stream, variables):
- for event in self.events:
- event.genSchema (stream, variables)
+ stream.write ("string& methodName, Buffer& inBuf, Buffer& outBuf")
def genHiLoStatResets (self, stream, variables):
for inst in self.statistics:
@@ -1124,7 +1142,7 @@
return
def genSchemaMD5 (self, stream, variables):
- sum = self.md5Sum.digest ()
+ sum = self.hash.getDigest()
for idx in range (len (sum)):
if idx != 0:
stream.write (",")
@@ -1149,6 +1167,20 @@
stat.genWrite (stream)
+class SchemaEventArgs:
+ def __init__(self, package, node, typespec, fragments, options):
+ self.packageName = package
+ self.options = options
+ self.args = {}
+
+ children = node.childNodes
+ for child in children:
+ if child.nodeType == Node.ELEMENT_NODE:
+ if child.nodeName == 'arg':
+ arg = SchemaArg(child, typespec)
+ self.args[arg.name] = arg
+ else:
+ raise Exception("Unknown tag '%s' in <eventArguments>" % child.nodeName)
class SchemaPackage:
def __init__ (self, typefile, schemafile, options):
@@ -1156,6 +1188,8 @@
self.classes = []
self.fragments = []
self.typespec = TypeSpec (typefile)
+ self.eventArgSet = None
+ self.events = []
dom = parse (schemafile)
document = dom.documentElement
@@ -1179,6 +1213,15 @@
self.fragments, options)
self.fragments.append (cls)
+ elif child.nodeName == 'eventArguments':
+ if self.eventArgSet:
+ raise Exception("Only one <eventArguments> may appear in a package")
+ self.eventArgSet = SchemaEventArgs(self.packageName, child, self.typespec, self.fragments, options)
+
+ elif child.nodeName == 'event':
+ event = SchemaEvent(self.packageName, child, self.typespec, self.eventArgSet)
+ self.events.append(event)
+
else:
raise ValueError ("Unknown schema tag '%s'" % child.nodeName)
@@ -1194,6 +1237,9 @@
def getClasses (self):
return self.classes
+ def getEvents(self):
+ return self.events
+
def genCloseNamespaces (self, stream, variables):
for item in self.packageName.split("."):
stream.write ("}")
@@ -1217,12 +1263,20 @@
stream.write ("#include \"")
_class.genNameCap (stream, variables)
stream.write (".h\"\n")
+ for _event in self.events:
+ stream.write ("#include \"Event")
+ _event.genNameCap(stream, variables)
+ stream.write (".h\"\n")
- def genClassRegisters (self, stream, variables):
+ def genClassRegisters(self, stream, variables):
for _class in self.classes:
- stream.write (" ")
- _class.genNameCap (stream, variables)
- stream.write ("::registerClass(agent);\n")
+ stream.write(" ")
+ _class.genNameCap(stream, variables)
+ stream.write("::registerSelf(agent);\n")
+ for _event in self.events:
+ stream.write(" Event")
+ _event.genNameCap(stream, variables)
+ stream.write("::registerSelf(agent);\n")
#=====================================================================================
Modified: incubator/qpid/trunk/qpid/cpp/managementgen/qmf/templates/Class.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/managementgen/qmf/templates/Class.cpp?rev=702651&r1=702650&r2=702651&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/managementgen/qmf/templates/Class.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/managementgen/qmf/templates/Class.cpp Tue Oct 7 14:47:35 2008
@@ -85,9 +85,9 @@
const string DEFAULT("default");
}
-void /*MGEN:Class.NameCap*/::registerClass(ManagementAgent* agent)
+void /*MGEN:Class.NameCap*/::registerSelf(ManagementAgent* agent)
{
- agent->RegisterClass(packageName, className, md5Sum, writeSchema);
+ agent->registerClass(packageName, className, md5Sum, writeSchema);
}
void /*MGEN:Class.NameCap*/::writeSchema (Buffer& buf)
@@ -95,13 +95,13 @@
FieldTable ft;
// Schema class header:
+ buf.putOctet (CLASS_KIND_TABLE);
buf.putShortString (packageName); // Package Name
buf.putShortString (className); // Class Name
buf.putBin128 (md5Sum); // Schema Hash
buf.putShort (/*MGEN:Class.ConfigCount*/); // Config Element Count
buf.putShort (/*MGEN:Class.InstCount*/); // Inst Element Count
buf.putShort (/*MGEN:Class.MethodCount*/); // Method Count
- buf.putShort (/*MGEN:Class.EventCount*/); // Event Count
// Properties
/*MGEN:Class.PropertySchema*/
@@ -109,8 +109,6 @@
/*MGEN:Class.StatisticSchema*/
// Methods
/*MGEN:Class.MethodSchema*/
- // Events
-/*MGEN:Class.EventSchema*/
}
/*MGEN:IF(Class.ExistPerThreadStats)*/
@@ -176,9 +174,8 @@
{
Manageable::status_t status = Manageable::STATUS_UNKNOWN_METHOD;
std::string text;
+
/*MGEN:Class.MethodHandlers*/
outBuf.putLong(status);
outBuf.putShortString(Manageable::StatusText(status, text));
}
-
-/*MGEN:Class.EventMethodBodies*/
Modified: incubator/qpid/trunk/qpid/cpp/managementgen/qmf/templates/Class.h
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/managementgen/qmf/templates/Class.h?rev=702651&r1=702650&r2=702651&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/managementgen/qmf/templates/Class.h (original)
+++ incubator/qpid/trunk/qpid/cpp/managementgen/qmf/templates/Class.h Tue Oct 7 14:47:35 2008
@@ -72,7 +72,7 @@
void writeProperties (::qpid::framing::Buffer& buf);
void writeStatistics (::qpid::framing::Buffer& buf,
bool skipHeaders = false);
- void doMethod (std::string methodName,
+ void doMethod (std::string& methodName,
::qpid::framing::Buffer& inBuf,
::qpid::framing::Buffer& outBuf);
writeSchemaCall_t getWriteSchemaCall(void) { return writeSchema; }
@@ -88,17 +88,15 @@
/*MGEN:Class.SetGeneralReferenceDeclaration*/
- static void registerClass (::qpid::management::ManagementAgent* agent);
- std::string& getPackageName (void) { return packageName; }
- std::string& getClassName (void) { return className; }
- uint8_t* getMd5Sum (void) { return md5Sum; }
+ static void registerSelf (::qpid::management::ManagementAgent* agent);
+ std::string& getPackageName (void) const { return packageName; }
+ std::string& getClassName (void) const { return className; }
+ uint8_t* getMd5Sum (void) const { return md5Sum; }
// Method IDs
/*MGEN:Class.MethodIdDeclarations*/
// Accessor Methods
/*MGEN:Class.AccessorMethods*/
- // Event Methods
-/*MGEN:Class.EventMethodDecls*/
};
}/*MGEN:Class.CloseNamespaces*/
Added: incubator/qpid/trunk/qpid/cpp/managementgen/qmf/templates/Event.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/managementgen/qmf/templates/Event.cpp?rev=702651&view=auto
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/managementgen/qmf/templates/Event.cpp (added)
+++ incubator/qpid/trunk/qpid/cpp/managementgen/qmf/templates/Event.cpp Tue Oct 7 14:47:35 2008
@@ -0,0 +1,77 @@
+/*MGEN:commentPrefix=//*/
+//
+// 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.
+//
+
+/*MGEN:Root.Disclaimer*/
+
+#include "qpid/log/Statement.h"
+#include "qpid/framing/FieldTable.h"
+#include "qpid/management/Manageable.h"
+#include "qpid/agent/ManagementAgent.h"
+#include "Event/*MGEN:Event.NameCap*/.h"
+
+using namespace qmf::/*MGEN:Event.Namespace*/;
+using namespace qpid::framing;
+using qpid::management::ManagementAgent;
+using qpid::management::Manageable;
+using qpid::management::ManagementObject;
+using qpid::management::Args;
+using std::string;
+
+string Event/*MGEN:Event.NameCap*/::packageName = string ("/*MGEN:Event.NamePackageLower*/");
+string Event/*MGEN:Event.NameCap*/::eventName = string ("/*MGEN:Event.Name*/");
+uint8_t Event/*MGEN:Event.NameCap*/::md5Sum[16] =
+ {/*MGEN:Event.SchemaMD5*/};
+
+Event/*MGEN:Event.NameCap*/::Event/*MGEN:Event.NameCap*/ (/*MGEN:Event.ConstructorArgs*/) :
+ /*MGEN:Event.ConstructorInits*/
+{}
+
+namespace {
+ const string NAME("name");
+ const string TYPE("type");
+ const string DESC("desc");
+ const string ARGCOUNT("argCount");
+ const string ARGS("args");
+}
+
+void Event/*MGEN:Event.NameCap*/::registerSelf(ManagementAgent* agent)
+{
+ agent->registerEvent(packageName, eventName, md5Sum, writeSchema);
+}
+
+void Event/*MGEN:Event.NameCap*/::writeSchema (Buffer& buf)
+{
+ FieldTable ft;
+
+ // Schema class header:
+ buf.putOctet (CLASS_KIND_EVENT);
+ buf.putShortString (packageName); // Package Name
+ buf.putShortString (eventName); // Event Name
+ buf.putBin128 (md5Sum); // Schema Hash
+ buf.putShort (/*MGEN:Event.ArgCount*/); // Argument Count
+
+ // Arguments
+/*MGEN:Event.ArgSchema*/
+}
+
+void Event/*MGEN:Event.NameCap*/::encode(::qpid::framing::Buffer& buf) const
+{
+/*MGEN:Event.ArgEncodes*/
+}
Added: incubator/qpid/trunk/qpid/cpp/managementgen/qmf/templates/Event.h
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/managementgen/qmf/templates/Event.h?rev=702651&view=auto
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/managementgen/qmf/templates/Event.h (added)
+++ incubator/qpid/trunk/qpid/cpp/managementgen/qmf/templates/Event.h Tue Oct 7 14:47:35 2008
@@ -0,0 +1,58 @@
+/*MGEN:commentPrefix=//*/
+#ifndef _MANAGEMENT_/*MGEN:Event.NameUpper*/_
+#define _MANAGEMENT_/*MGEN:Event.NameUpper*/_
+
+//
+// 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.
+//
+
+/*MGEN:Root.Disclaimer*/
+
+#include "qpid/management/ManagementEvent.h"
+#include "qpid/framing/FieldTable.h"
+#include "qpid/framing/Uuid.h"
+
+namespace qmf {
+/*MGEN:Event.OpenNamespaces*/
+
+class Event/*MGEN:Event.NameCap*/ : public ::qpid::management::ManagementEvent
+{
+ private:
+ static void writeSchema (::qpid::framing::Buffer& buf);
+ static std::string packageName;
+ static std::string eventName;
+ static uint8_t md5Sum[16];
+
+/*MGEN:Event.ArgDeclarations*/
+
+ public:
+ writeSchemaCall_t getWriteSchemaCall(void) { return writeSchema; }
+
+ Event/*MGEN:Event.NameCap*/(/*MGEN:Event.ConstructorArgs*/);
+ ~Event/*MGEN:Class.NameCap*/() {};
+
+ static void registerSelf(::qpid::management::ManagementAgent* agent);
+ std::string& getPackageName() const { return packageName; }
+ std::string& getEventName() const { return eventName; }
+ uint8_t* getMd5Sum() const { return md5Sum; }
+ void encode(::qpid::framing::Buffer& buffer) const;
+};
+
+}/*MGEN:Event.CloseNamespaces*/
+
+#endif /*!_MANAGEMENT_/*MGEN:Event.NameUpper*/_*/
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/acl/Acl.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/acl/Acl.cpp?rev=702651&r1=702650&r2=702651&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/acl/Acl.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/acl/Acl.cpp Tue Oct 7 14:47:35 2008
@@ -25,6 +25,10 @@
#include "qpid/shared_ptr.h"
#include "qpid/log/Logger.h"
#include "qmf/org/apache/qpid/acl/Package.h"
+#include "qmf/org/apache/qpid/acl/EventAllow.h"
+#include "qmf/org/apache/qpid/acl/EventDeny.h"
+#include "qmf/org/apache/qpid/acl/EventFileLoaded.h"
+#include "qmf/org/apache/qpid/acl/EventFileLoadFailed.h"
#include <map>
@@ -41,7 +45,7 @@
Acl::Acl (AclValues& av, broker::Broker& b): aclValues(av), broker(&b), transferAcl(false)
{
- ManagementAgent* agent = ManagementAgent::Singleton::getInstance();
+ agent = ManagementAgent::Singleton::getInstance();
if (agent != 0){
_qmf::Package packageInit(agent);
@@ -86,7 +90,11 @@
switch (aclreslt)
{
case ALLOWLOG:
- QPID_LOG(info, "ACL Allow id:" << id <<" action:" << AclHelper::getActionStr(action) << " ObjectType:" << AclHelper::getObjectTypeStr(objType) << " Name:" << name );
+ QPID_LOG(info, "ACL Allow id:" << id <<" action:" << AclHelper::getActionStr(action) <<
+ " ObjectType:" << AclHelper::getObjectTypeStr(objType) << " Name:" << name );
+ agent->raiseEvent(_qmf::EventAllow(id, AclHelper::getActionStr(action),
+ AclHelper::getObjectTypeStr(objType),
+ name, framing::FieldTable()));
case ALLOW:
return true;
case DENY:
@@ -94,13 +102,12 @@
return false;
case DENYLOG:
if (mgmtObject!=0) mgmtObject->inc_aclDenyCount();
- default:
- QPID_LOG(info, "ACL Deny id:" << id << " action:" << AclHelper::getActionStr(action) << " ObjectType:" << AclHelper::getObjectTypeStr(objType) << " Name:" << name);
- if (mgmtObject!=0){
- framing::FieldTable _params;
- mgmtObject->event_aclEvent(1, id, AclHelper::getActionStr(action),AclHelper::getObjectTypeStr(objType),name, _params);
- }
- return false;
+ default:
+ QPID_LOG(info, "ACL Deny id:" << id << " action:" << AclHelper::getActionStr(action) << " ObjectType:" << AclHelper::getObjectTypeStr(objType) << " Name:" << name);
+ agent->raiseEvent(_qmf::EventDeny(id, AclHelper::getActionStr(action),
+ AclHelper::getObjectTypeStr(objType),
+ name, framing::FieldTable()));
+ return false;
}
return false;
}
@@ -115,7 +122,7 @@
boost::shared_ptr<AclData> d(new AclData);
AclReader ar;
if (ar.read(aclFile, d)){
- mgmtObject->event_fileNotLoaded("","See log for file load reason failure");
+ agent->raiseEvent(_qmf::EventFileLoadFailed("", "See log for file load reason failure"));
return false;
}
@@ -127,7 +134,7 @@
sys::AbsTime now = sys::AbsTime::now();
int64_t ns = sys::Duration(now);
mgmtObject->set_lastAclLoad(ns);
- mgmtObject->event_fileLoaded("");
+ agent->raiseEvent(_qmf::EventFileLoaded(""));
}
return true;
}
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/acl/Acl.h
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/acl/Acl.h?rev=702651&r1=702650&r2=702651&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/acl/Acl.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/acl/Acl.h Tue Oct 7 14:47:35 2008
@@ -27,6 +27,7 @@
#include "qpid/RefCounted.h"
#include "qpid/broker/AclModule.h"
#include "qpid/management/Manageable.h"
+#include "qpid/agent/ManagementAgent.h"
#include "qmf/org/apache/qpid/acl/Acl.h"
#include <map>
@@ -57,7 +58,7 @@
bool transferAcl;
boost::shared_ptr<AclData> data;
qmf::org::apache::qpid::acl::Acl* mgmtObject; // mgnt owns lifecycle
-
+ qpid::management::ManagementAgent* agent;
public:
Acl (AclValues& av, broker::Broker& b);
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/acl/management-schema.xml
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/acl/management-schema.xml?rev=702651&r1=702650&r2=702651&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/acl/management-schema.xml (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/acl/management-schema.xml Tue Oct 7 14:47:35 2008
@@ -17,32 +17,28 @@
-->
<class name="acl">
- <property name="brokerRef" type="objId" references="qpid.Broker" access="RO" index="y" parentRef="y"/>
- <property name="policyFile" type="sstr" access="RO" desc="Name of the policy file"/>
- <property name="enforcingAcl" type="bool" access="RO" desc="Currently Enforcing ACL"/>
- <property name="transferAcl" type="bool" access="RO" desc="Any transfer ACL rules in force"/>
- <property name="lastAclLoad" type="absTime" access="RO" desc="Timestamp of last successful load of ACL"/>
- <statistic name="aclDenyCount" type="count64" unit="record" desc="Number of ACL requests denied"/>
+ <property name="brokerRef" type="objId" references="org.apache.qpid.broker:Broker" access="RO" index="y" parentRef="y"/>
+ <property name="policyFile" type="sstr" access="RO" desc="Name of the policy file"/>
+ <property name="enforcingAcl" type="bool" access="RO" desc="Currently Enforcing ACL"/>
+ <property name="transferAcl" type="bool" access="RO" desc="Any transfer ACL rules in force"/>
+ <property name="lastAclLoad" type="absTime" access="RO" desc="Timestamp of last successful load of ACL"/>
+ <statistic name="aclDenyCount" type="count64" unit="request" desc="Number of ACL requests denied"/>
<method name="reloadACLFile" desc="Reload the ACL file"/>
-
- <event name="aclEvent" defaultSeverity="info" desc="Event generated by the ACL policy">
- <arg name="denied" type="bool"/>
- <arg name="authId" type="sstr"/>
- <arg name="action" type="sstr"/>
- <arg name="objType" type="sstr"/>
- <arg name="name" type="sstr"/>
- <arg name="params" type="map"/>
- </event>
-
- <event name="fileLoaded" defaultSeverity="warning" desc="ACL file successfully loaded - New policy in effect">
- <arg name="authId" type="sstr" desc="Name of user who initiated the file load"/>
- </event>
-
- <event name="fileNotLoaded" defaultSeverity="error" desc="Replacement ACL file could not be loaded">
- <arg name="authId" type="sstr" desc="Name of user who initiated the file load"/>
- <arg name="reason" type="sstr" desc="Reason for failure"/>
- </event>
</class>
+ <eventArguments>
+ <arg name="action" type="sstr"/>
+ <arg name="arguments" type="map"/>
+ <arg name="objectName" type="sstr"/>
+ <arg name="objectType" type="sstr"/>
+ <arg name="reason" type="sstr"/>
+ <arg name="userId" type="sstr"/>
+ </eventArguments>
+
+ <event name="allow" args="userId, action, objectType, objectName, arguments"/>
+ <event name="deny" args="userId, action, objectType, objectName, arguments"/>
+ <event name="fileLoaded" args="userId"/>
+ <event name="fileLoadFailed" args="userId, reason"/>
+
</schema>
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/agent/ManagementAgent.h
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/agent/ManagementAgent.h?rev=702651&r1=702650&r2=702651&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/agent/ManagementAgent.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/agent/ManagementAgent.h Tue Oct 7 14:47:35 2008
@@ -21,6 +21,7 @@
//
#include "qpid/management/ManagementObject.h"
+#include "qpid/management/ManagementEvent.h"
#include "qpid/management/Manageable.h"
#include "qpid/sys/Mutex.h"
@@ -43,8 +44,8 @@
static ManagementAgent* agent;
};
- ManagementAgent () {}
- virtual ~ManagementAgent () {}
+ ManagementAgent() {}
+ virtual ~ManagementAgent() {}
virtual int getMaxThreads() = 0;
@@ -78,10 +79,16 @@
// package initializer generated by the management code generator.
//
virtual void
- RegisterClass (std::string packageName,
- std::string className,
- uint8_t* md5Sum,
- management::ManagementObject::writeSchemaCall_t schemaCall) = 0;
+ registerClass(std::string& packageName,
+ std::string& className,
+ uint8_t* md5Sum,
+ management::ManagementObject::writeSchemaCall_t schemaCall) = 0;
+
+ virtual void
+ registerEvent(std::string& packageName,
+ std::string& eventName,
+ uint8_t* md5Sum,
+ management::ManagementEvent::writeSchemaCall_t schemaCall) = 0;
// Add a management object to the agent. Once added, this object shall be visible
// in the greater management context.
@@ -97,8 +104,11 @@
// pointer. This allows the management agent to report the deletion of the object
// in an orderly way.
//
- virtual ObjectId addObject (ManagementObject* objectPtr,
- uint64_t persistId = 0) = 0;
+ virtual ObjectId addObject(ManagementObject* objectPtr, uint64_t persistId = 0) = 0;
+
+ //
+ //
+ virtual void raiseEvent(const ManagementEvent& event) = 0;
// If "useExternalThread" was set to true in init, this method must
// be called to provide a thread for any pending method calls that have arrived.
@@ -113,7 +123,7 @@
// to pollCallbacks are necessary to clear the backlog. If callLimit is zero,
// the return value will also be zero.
//
- virtual uint32_t pollCallbacks (uint32_t callLimit = 0) = 0;
+ virtual uint32_t pollCallbacks(uint32_t callLimit = 0) = 0;
// If "useExternalThread" was set to true in the constructor, this method provides
// a standard file descriptor that can be used in a select statement to signal that
@@ -121,14 +131,7 @@
// least one method call). When this fd is ready-for-read, pollCallbacks may be
// invoked. Calling pollCallbacks shall reset the ready-to-read state of the fd.
//
- virtual int getSignalFd (void) = 0;
-
-protected:
- friend class ManagementObject;
- virtual sys::Mutex& getMutex() = 0;
- virtual framing::Buffer* startEventLH() = 0;
- virtual void finishEventLH(framing::Buffer* buf) = 0;
-
+ virtual int getSignalFd() = 0;
};
}}
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/agent/ManagementAgentImpl.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/agent/ManagementAgentImpl.cpp?rev=702651&r1=702650&r2=702651&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/agent/ManagementAgentImpl.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/agent/ManagementAgentImpl.cpp Tue Oct 7 14:47:35 2008
@@ -75,12 +75,13 @@
return agent;
}
-const string ManagementAgentImpl::storeMagicNumber("MA01");
+const string ManagementAgentImpl::storeMagicNumber("MA02");
ManagementAgentImpl::ManagementAgentImpl() :
extThread(false), writeFd(-1), readFd(-1),
- clientWasAdded(true), requestedBank(0),
- assignedBank(0), brokerBank(0), bootSequence(0),
+ connected(false), lastFailure("never connected"),
+ clientWasAdded(true), requestedBrokerBank(0), requestedAgentBank(0),
+ assignedBrokerBank(0), assignedAgentBank(0), bootSequence(0),
connThreadBody(*this), connThread(connThreadBody),
pubThreadBody(*this), pubThread(pubThreadBody)
{
@@ -122,18 +123,24 @@
storeData(true);
}
-ManagementAgentImpl::~ManagementAgentImpl()
-{
+void ManagementAgentImpl::registerClass(std::string& packageName,
+ std::string& className,
+ uint8_t* md5Sum,
+ management::ManagementObject::writeSchemaCall_t schemaCall)
+{
+ Mutex::ScopedLock lock(agentLock);
+ PackageMap::iterator pIter = findOrAddPackage(packageName);
+ addClassLocal(ManagementItem::CLASS_KIND_TABLE, pIter, className, md5Sum, schemaCall);
}
-void ManagementAgentImpl::RegisterClass(std::string packageName,
- std::string className,
- uint8_t* md5Sum,
+void ManagementAgentImpl::registerEvent(std::string& packageName,
+ std::string& eventName,
+ uint8_t* md5Sum,
management::ManagementObject::writeSchemaCall_t schemaCall)
{
Mutex::ScopedLock lock(agentLock);
- PackageMap::iterator pIter = FindOrAddPackage(packageName);
- AddClassLocal(pIter, className, md5Sum, schemaCall);
+ PackageMap::iterator pIter = findOrAddPackage(packageName);
+ addClassLocal(ManagementItem::CLASS_KIND_EVENT, pIter, eventName, md5Sum, schemaCall);
}
ObjectId ManagementAgentImpl::addObject(ManagementObject* object,
@@ -151,6 +158,23 @@
return objectId;
}
+void ManagementAgentImpl::raiseEvent(const ManagementEvent& event)
+{
+ Mutex::ScopedLock lock(agentLock);
+ Buffer outBuffer(eventBuffer, MA_BUFFER_SIZE);
+ uint32_t outLen;
+
+ encodeHeader(outBuffer, 'e');
+ outBuffer.putShortString(event.getPackageName());
+ outBuffer.putShortString(event.getEventName());
+ outBuffer.putBin128(event.getMd5Sum());
+ outBuffer.putLongLong(uint64_t(Duration(now())));
+ event.encode(outBuffer);
+ outLen = MA_BUFFER_SIZE - outBuffer.available();
+ outBuffer.reset();
+ connThreadBody.sendBuffer(outBuffer, outLen, "qpid.management", "mgmt.event");
+}
+
uint32_t ManagementAgentImpl::pollCallbacks(uint32_t callLimit)
{
Mutex::ScopedLock lock(agentLock);
@@ -184,10 +208,12 @@
char rawbuffer[512];
Buffer buffer(rawbuffer, 512);
- EncodeHeader(buffer, 'A');
+ connected = true;
+ encodeHeader(buffer, 'A');
buffer.putShortString("RemoteAgent [C++]");
systemId.encode (buffer);
- buffer.putLong(requestedBank);
+ buffer.putLong(requestedBrokerBank);
+ buffer.putLong(requestedAgentBank);
uint32_t length = 512 - buffer.available();
buffer.reset();
connThreadBody.sendBuffer(buffer, length, "qpid.management", "broker");
@@ -197,10 +223,12 @@
{
if (!storeFile.empty()) {
ofstream outFile(storeFile.c_str());
- uint32_t bankToWrite = requested ? requestedBank : assignedBank;
+ uint32_t brokerBankToWrite = requested ? requestedBrokerBank : assignedBrokerBank;
+ uint32_t agentBankToWrite = requested ? requestedAgentBank : assignedAgentBank;
if (outFile.good()) {
- outFile << storeMagicNumber << " " << bankToWrite << " " << bootSequence << endl;
+ outFile << storeMagicNumber << " " << brokerBankToWrite << " " <<
+ agentBankToWrite << " " << bootSequence << endl;
outFile.close();
}
}
@@ -215,7 +243,8 @@
if (inFile.good()) {
inFile >> mn;
if (mn == storeMagicNumber) {
- inFile >> requestedBank;
+ inFile >> requestedBrokerBank;
+ inFile >> requestedAgentBank;
inFile >> bootSequence;
}
inFile.close();
@@ -229,7 +258,7 @@
Buffer outBuffer(outputBuffer, MA_BUFFER_SIZE);
uint32_t outLen;
- EncodeHeader(outBuffer, 'z', sequence);
+ encodeHeader(outBuffer, 'z', sequence);
outBuffer.putLong(code);
outBuffer.putShortString(text);
outLen = MA_BUFFER_SIZE - outBuffer.available();
@@ -241,20 +270,23 @@
{
Mutex::ScopedLock lock(agentLock);
- brokerBank = inBuffer.getLong();
- assignedBank = inBuffer.getLong();
- if (assignedBank != requestedBank) {
- if (requestedBank == 0)
- cout << "Initial object-id bank assigned: " << assignedBank << endl;
+ assignedBrokerBank = inBuffer.getLong();
+ assignedAgentBank = inBuffer.getLong();
+ if ((assignedBrokerBank != requestedBrokerBank) ||
+ (assignedAgentBank != requestedAgentBank)) {
+ if (requestedAgentBank == 0)
+ cout << "Initial object-id bank assigned: " << assignedBrokerBank << "." <<
+ assignedAgentBank << endl;
else
- cout << "Collision in object-id! New bank assigned: " << assignedBank << endl;
+ cout << "Collision in object-id! New bank assigned: " << assignedBrokerBank <<
+ "." << assignedAgentBank << endl;
storeData();
}
- attachment.setBanks(brokerBank, assignedBank);
+ attachment.setBanks(assignedBrokerBank, assignedAgentBank);
// Bind to qpid.management to receive commands
- connThreadBody.bindToBank(assignedBank);
+ connThreadBody.bindToBank(assignedBrokerBank, assignedAgentBank);
// Send package indications for all local packages
for (PackageMap::iterator pIter = packages.begin();
@@ -263,8 +295,8 @@
Buffer outBuffer(outputBuffer, MA_BUFFER_SIZE);
uint32_t outLen;
- EncodeHeader(outBuffer, 'p');
- EncodePackageIndication(outBuffer, pIter);
+ encodeHeader(outBuffer, 'p');
+ encodePackageIndication(outBuffer, pIter);
outLen = MA_BUFFER_SIZE - outBuffer.available();
outBuffer.reset();
connThreadBody.sendBuffer(outBuffer, outLen, "qpid.management", "broker");
@@ -273,8 +305,8 @@
ClassMap cMap = pIter->second;
for (ClassMap::iterator cIter = cMap.begin(); cIter != cMap.end(); cIter++) {
outBuffer.reset();
- EncodeHeader(outBuffer, 'q');
- EncodeClassIndication(outBuffer, pIter, cIter);
+ encodeHeader(outBuffer, 'q');
+ encodeClassIndication(outBuffer, pIter, cIter);
outLen = MA_BUFFER_SIZE - outBuffer.available();
outBuffer.reset();
connThreadBody.sendBuffer(outBuffer, outLen, "qpid.management", "broker");
@@ -294,14 +326,14 @@
PackageMap::iterator pIter = packages.find(packageName);
if (pIter != packages.end()) {
- ClassMap cMap = pIter->second;
+ ClassMap& cMap = pIter->second;
ClassMap::iterator cIter = cMap.find(key);
if (cIter != cMap.end()) {
- SchemaClass schema = cIter->second;
+ SchemaClass& schema = cIter->second;
Buffer outBuffer(outputBuffer, MA_BUFFER_SIZE);
uint32_t outLen;
- EncodeHeader(outBuffer, 's', sequence);
+ encodeHeader(outBuffer, 's', sequence);
schema.writeSchemaCall(outBuffer);
outLen = MA_BUFFER_SIZE - outBuffer.available();
outBuffer.reset();
@@ -331,7 +363,7 @@
inBuffer.getBin128(hash);
inBuffer.getShortString(methodName);
- EncodeHeader(outBuffer, 'm', sequence);
+ encodeHeader(outBuffer, 'm', sequence);
ManagementObjectMap::iterator iter = managementObjects.find(objId);
if (iter == managementObjects.end() || iter->second->isDeleted()) {
@@ -344,7 +376,14 @@
outBuffer.putShortString (Manageable::StatusText (Manageable::STATUS_INVALID_PARAMETER));
}
else
- iter->second->doMethod(methodName, inBuffer, outBuffer);
+ try {
+ outBuffer.record();
+ iter->second->doMethod(methodName, inBuffer, outBuffer);
+ } catch(std::exception& e) {
+ outBuffer.restore();
+ outBuffer.putLong(Manageable::STATUS_EXCEPTION);
+ outBuffer.putShortString(e.what());
+ }
}
outLen = MA_BUFFER_SIZE - outBuffer.available();
@@ -379,7 +418,7 @@
Buffer outBuffer(outputBuffer, MA_BUFFER_SIZE);
uint32_t outLen;
- EncodeHeader(outBuffer, 'g', sequence);
+ encodeHeader(outBuffer, 'g', sequence);
object->writeProperties(outBuffer);
object->writeStatistics(outBuffer, true);
outLen = MA_BUFFER_SIZE - outBuffer.available();
@@ -419,7 +458,7 @@
replyToKey = rt.getRoutingKey();
}
- if (CheckHeader(inBuffer, &opcode, &sequence))
+ if (checkHeader(inBuffer, &opcode, &sequence))
{
if (opcode == 'a') handleAttachResponse(inBuffer);
else if (opcode == 'S') handleSchemaRequest(inBuffer, sequence);
@@ -429,16 +468,16 @@
}
}
-void ManagementAgentImpl::EncodeHeader(Buffer& buf, uint8_t opcode, uint32_t seq)
+void ManagementAgentImpl::encodeHeader(Buffer& buf, uint8_t opcode, uint32_t seq)
{
buf.putOctet('A');
buf.putOctet('M');
- buf.putOctet('1');
+ buf.putOctet('2');
buf.putOctet(opcode);
buf.putLong (seq);
}
-bool ManagementAgentImpl::CheckHeader(Buffer& buf, uint8_t *opcode, uint32_t *seq)
+bool ManagementAgentImpl::checkHeader(Buffer& buf, uint8_t *opcode, uint32_t *seq)
{
if (buf.getSize() < 8)
return false;
@@ -450,10 +489,10 @@
*opcode = buf.getOctet();
*seq = buf.getLong();
- return h1 == 'A' && h2 == 'M' && h3 == '1';
+ return h1 == 'A' && h2 == 'M' && h3 == '2';
}
-ManagementAgentImpl::PackageMap::iterator ManagementAgentImpl::FindOrAddPackage(std::string name)
+ManagementAgentImpl::PackageMap::iterator ManagementAgentImpl::findOrAddPackage(const string& name)
{
PackageMap::iterator pIter = packages.find(name);
if (pIter != packages.end())
@@ -467,8 +506,8 @@
Buffer outBuffer(outputBuffer, MA_BUFFER_SIZE);
uint32_t outLen;
- EncodeHeader(outBuffer, 'p');
- EncodePackageIndication(outBuffer, result.first);
+ encodeHeader(outBuffer, 'p');
+ encodePackageIndication(outBuffer, result.first);
outLen = MA_BUFFER_SIZE - outBuffer.available();
outBuffer.reset();
connThreadBody.sendBuffer(outBuffer, outLen, "qpid.management", "mgmt.schema.package");
@@ -486,8 +525,9 @@
newManagementObjects.clear();
}
-void ManagementAgentImpl::AddClassLocal(PackageMap::iterator pIter,
- string className,
+void ManagementAgentImpl::addClassLocal(uint8_t classKind,
+ PackageMap::iterator pIter,
+ const string& className,
uint8_t* md5Sum,
management::ManagementObject::writeSchemaCall_t schemaCall)
{
@@ -502,32 +542,28 @@
return;
// No such class found, create a new class with local information.
- SchemaClass classInfo;
-
- classInfo.writeSchemaCall = schemaCall;
- cMap[key] = classInfo;
-
- // TODO: Publish a class-indication message
+ cMap.insert(std::pair<SchemaClassKey, SchemaClass>(key, SchemaClass(schemaCall, classKind)));
}
-void ManagementAgentImpl::EncodePackageIndication(Buffer& buf,
+void ManagementAgentImpl::encodePackageIndication(Buffer& buf,
PackageMap::iterator pIter)
{
buf.putShortString((*pIter).first);
}
-void ManagementAgentImpl::EncodeClassIndication(Buffer& buf,
+void ManagementAgentImpl::encodeClassIndication(Buffer& buf,
PackageMap::iterator pIter,
ClassMap::iterator cIter)
{
SchemaClassKey key = (*cIter).first;
+ buf.putOctet((*cIter).second.kind);
buf.putShortString((*pIter).first);
buf.putShortString(key.name);
- buf.putBin128 (key.hash);
+ buf.putBin128(key.hash);
}
-void ManagementAgentImpl::PeriodicProcessing()
+void ManagementAgentImpl::periodicProcessing()
{
#define BUFSIZE 65536
Mutex::ScopedLock lock(agentLock);
@@ -536,9 +572,12 @@
string routingKey;
std::list<ObjectId> deleteList;
+ if (!connected)
+ return;
+
{
Buffer msgBuffer(msgChars, BUFSIZE);
- EncodeHeader(msgBuffer, 'h');
+ encodeHeader(msgBuffer, 'h');
msgBuffer.putLongLong(uint64_t(Duration(now())));
contentSize = BUFSIZE - msgBuffer.available();
@@ -573,7 +612,7 @@
if (object->getConfigChanged() || object->isDeleted())
{
Buffer msgBuffer(msgChars, BUFSIZE);
- EncodeHeader(msgBuffer, 'c');
+ encodeHeader(msgBuffer, 'c');
object->writeProperties(msgBuffer);
contentSize = BUFSIZE - msgBuffer.available();
@@ -585,7 +624,7 @@
if (object->getInstChanged())
{
Buffer msgBuffer(msgChars, BUFSIZE);
- EncodeHeader(msgBuffer, 'i');
+ encodeHeader(msgBuffer, 'i');
object->writeStatistics(msgBuffer);
contentSize = BUFSIZE - msgBuffer.available();
@@ -664,8 +703,8 @@
void ManagementAgentImpl::ConnectionThread::sendBuffer(Buffer& buf,
uint32_t length,
- string exchange,
- string routingKey)
+ const string& exchange,
+ const string& routingKey)
{
{
Mutex::ScopedLock _lock(connLock);
@@ -683,10 +722,10 @@
session.messageTransfer(arg::content=msg, arg::destination=exchange);
}
-void ManagementAgentImpl::ConnectionThread::bindToBank(uint32_t agentBank)
+void ManagementAgentImpl::ConnectionThread::bindToBank(uint32_t brokerBank, uint32_t agentBank)
{
stringstream key;
- key << "agent." << agentBank;
+ key << "agent." << brokerBank << "." << agentBank;
session.exchangeBind(arg::exchange="qpid.management", arg::queue=queueName.str(),
arg::bindingKey=key.str());
}
@@ -695,28 +734,7 @@
void ManagementAgentImpl::PublishThread::run()
{
while (true) {
- ::sleep(5);
- agent.PeriodicProcessing();
+ ::sleep(agent.getInterval());
+ agent.periodicProcessing();
}
}
-
-Mutex& ManagementAgentImpl::getMutex()
-{
- return agentLock;
-}
-
-Buffer* ManagementAgentImpl::startEventLH()
-{
- Buffer* outBuffer(new Buffer(eventBuffer, MA_BUFFER_SIZE));
- EncodeHeader(*outBuffer, 'e');
- outBuffer->putLongLong(uint64_t(Duration(now())));
- return outBuffer;
-}
-
-void ManagementAgentImpl::finishEventLH(Buffer* outBuffer)
-{
- uint32_t outLen = MA_BUFFER_SIZE - outBuffer->available();
- outBuffer->reset();
- connThreadBody.sendBuffer(*outBuffer, outLen, "qpid.management", "mgmt.event");
- delete outBuffer;
-}
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/agent/ManagementAgentImpl.h
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/agent/ManagementAgentImpl.h?rev=702651&r1=702650&r2=702651&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/agent/ManagementAgentImpl.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/agent/ManagementAgentImpl.h Tue Oct 7 14:47:35 2008
@@ -43,24 +43,34 @@
public:
ManagementAgentImpl();
- virtual ~ManagementAgentImpl();
+ virtual ~ManagementAgentImpl() {};
+ //
+ // Methods from ManagementAgent
+ //
int getMaxThreads() { return 1; }
void init(std::string brokerHost = "localhost",
uint16_t brokerPort = 5672,
uint16_t intervalSeconds = 10,
bool useExternalThread = false,
std::string storeFile = "");
- void RegisterClass(std::string packageName,
- std::string className,
- uint8_t* md5Sum,
+ bool isConnected() { return connected; }
+ std::string& getLastFailure() { return lastFailure; }
+ void registerClass(std::string& packageName,
+ std::string& className,
+ uint8_t* md5Sum,
management::ManagementObject::writeSchemaCall_t schemaCall);
- ObjectId addObject (management::ManagementObject* objectPtr,
- uint64_t persistId = 0);
- uint32_t pollCallbacks (uint32_t callLimit = 0);
- int getSignalFd (void);
+ void registerEvent(std::string& packageName,
+ std::string& eventName,
+ uint8_t* md5Sum,
+ management::ManagementObject::writeSchemaCall_t schemaCall);
+ ObjectId addObject(management::ManagementObject* objectPtr, uint64_t persistId = 0);
+ void raiseEvent(const management::ManagementEvent& event);
+ uint32_t pollCallbacks(uint32_t callLimit = 0);
+ int getSignalFd();
- void PeriodicProcessing();
+ uint16_t getInterval() { return interval; }
+ void periodicProcessing();
private:
@@ -84,8 +94,10 @@
struct SchemaClass {
management::ManagementObject::writeSchemaCall_t writeSchemaCall;
+ uint8_t kind;
- SchemaClass () : writeSchemaCall(0) {}
+ SchemaClass(const management::ManagementObject::writeSchemaCall_t call,
+ const uint8_t _kind) : writeSchemaCall(call), kind(_kind) {}
};
struct QueuedMethod {
@@ -120,12 +132,16 @@
framing::Uuid systemId;
std::string host;
uint16_t port;
+ bool connected;
+ std::string lastFailure;
+
+ bool clientWasAdded;
+ uint32_t requestedBrokerBank;
+ uint32_t requestedAgentBank;
+ uint32_t assignedBrokerBank;
+ uint32_t assignedAgentBank;
+ uint16_t bootSequence;
- bool clientWasAdded;
- uint32_t requestedBank;
- uint32_t assignedBank;
- uint32_t brokerBank;
- uint16_t bootSequence;
# define MA_BUFFER_SIZE 65536
char outputBuffer[MA_BUFFER_SIZE];
char eventBuffer[MA_BUFFER_SIZE];
@@ -148,9 +164,9 @@
~ConnectionThread();
void sendBuffer(qpid::framing::Buffer& buf,
uint32_t length,
- std::string exchange,
- std::string routingKey);
- void bindToBank(uint32_t agentBank);
+ const std::string& exchange,
+ const std::string& routingKey);
+ void bindToBank(uint32_t brokerBank, uint32_t agentBank);
};
class PublishThread : public sys::Runnable
@@ -171,19 +187,20 @@
void startProtocol();
void storeData(bool requested=false);
void retrieveData();
- PackageMap::iterator FindOrAddPackage (std::string name);
+ PackageMap::iterator findOrAddPackage(const std::string& name);
void moveNewObjectsLH();
- void AddClassLocal (PackageMap::iterator pIter,
- std::string className,
+ void addClassLocal (uint8_t classKind,
+ PackageMap::iterator pIter,
+ const std::string& className,
uint8_t* md5Sum,
management::ManagementObject::writeSchemaCall_t schemaCall);
- void EncodePackageIndication (qpid::framing::Buffer& buf,
+ void encodePackageIndication (framing::Buffer& buf,
PackageMap::iterator pIter);
- void EncodeClassIndication (qpid::framing::Buffer& buf,
+ void encodeClassIndication (framing::Buffer& buf,
PackageMap::iterator pIter,
ClassMap::iterator cIter);
- void EncodeHeader (qpid::framing::Buffer& buf, uint8_t opcode, uint32_t seq = 0);
- bool CheckHeader (qpid::framing::Buffer& buf, uint8_t *opcode, uint32_t *seq);
+ void encodeHeader (framing::Buffer& buf, uint8_t opcode, uint32_t seq = 0);
+ bool checkHeader (framing::Buffer& buf, uint8_t *opcode, uint32_t *seq);
void sendCommandComplete (std::string replyToKey, uint32_t sequence,
uint32_t code = 0, std::string text = std::string("OK"));
void handleAttachResponse (qpid::framing::Buffer& inBuffer);
@@ -194,9 +211,6 @@
void handleGetQuery (qpid::framing::Buffer& inBuffer, uint32_t sequence, std::string replyTo);
void handleMethodRequest (qpid::framing::Buffer& inBuffer, uint32_t sequence, std::string replyTo);
void handleConsoleAddedIndication();
- sys::Mutex& getMutex();
- framing::Buffer* startEventLH();
- void finishEventLH(framing::Buffer* outBuffer);
};
}}
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/broker/Connection.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/broker/Connection.cpp?rev=702651&r1=702650&r2=702651&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/Connection.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/broker/Connection.cpp Tue Oct 7 14:47:35 2008
@@ -72,6 +72,7 @@
if (agent != 0)
mgmtObject = new _qmf::Connection(agent, this, parent, mgmtId, !isLink, false);
agent->addObject(mgmtObject);
+ ConnectionState::setUrl(mgmtId);
}
}
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/broker/ConnectionState.h
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/broker/ConnectionState.h?rev=702651&r1=702650&r2=702651&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/ConnectionState.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/broker/ConnectionState.h Tue Oct 7 14:47:35 2008
@@ -63,6 +63,9 @@
virtual void setUserId(const string& uid) { userId = uid; }
const string& getUserId() const { return userId; }
+ void setUrl(const string& _url) { url = _url; }
+ const string& getUrl() const { return url; }
+
void setFederationLink(bool b) { federationLink = b; }
bool isFederationLink() const { return federationLink; }
@@ -85,6 +88,7 @@
uint16_t heartbeat;
uint64_t stagingThreshold;
string userId;
+ string url;
bool federationLink;
};
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/broker/SessionAdapter.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/broker/SessionAdapter.cpp?rev=702651&r1=702650&r2=702651&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/SessionAdapter.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/broker/SessionAdapter.cpp Tue Oct 7 14:47:35 2008
@@ -26,6 +26,15 @@
#include "qpid/log/Statement.h"
#include "qpid/amqp_0_10/exceptions.h"
#include "qpid/framing/SequenceSet.h"
+#include "qpid/agent/ManagementAgent.h"
+#include "qmf/org/apache/qpid/broker/EventExchangeDeclare.h"
+#include "qmf/org/apache/qpid/broker/EventExchangeDelete.h"
+#include "qmf/org/apache/qpid/broker/EventQueueDeclare.h"
+#include "qmf/org/apache/qpid/broker/EventQueueDelete.h"
+#include "qmf/org/apache/qpid/broker/EventBind.h"
+#include "qmf/org/apache/qpid/broker/EventUnbind.h"
+#include "qmf/org/apache/qpid/broker/EventSubscribe.h"
+#include "qmf/org/apache/qpid/broker/EventUnsubscribe.h"
#include <boost/format.hpp>
#include <boost/cast.hpp>
#include <boost/bind.hpp>
@@ -36,6 +45,8 @@
using namespace qpid;
using namespace qpid::framing;
using namespace qpid::framing::dtx;
+using namespace qpid::management;
+namespace _qmf = qmf::org::apache::qpid::broker;
typedef std::vector<Queue::shared_ptr> QueueVector;
@@ -54,18 +65,17 @@
const string& alternateExchange,
bool passive, bool durable, bool /*autoDelete*/, const FieldTable& args){
- AclModule* acl = getBroker().getAcl();
- if (acl)
- {
+ AclModule* acl = getBroker().getAcl();
+ if (acl) {
std::map<acl::Property, std::string> params;
- params.insert(make_pair(acl::TYPE, type));
- params.insert(make_pair(acl::ALTERNATE, alternateExchange));
- params.insert(make_pair(acl::PASSIVE, std::string(passive ? "true" : "false") ));
- params.insert(make_pair(acl::DURABLE, std::string(durable ? "true" : "false")));
- if (!acl->authorise(getConnection().getUserId(),acl::CREATE,acl::EXCHANGE,exchange,¶ms) )
- throw NotAllowedException("ACL denied exhange declare request");
+ params.insert(make_pair(acl::TYPE, type));
+ params.insert(make_pair(acl::ALTERNATE, alternateExchange));
+ params.insert(make_pair(acl::PASSIVE, std::string(passive ? "true" : "false") ));
+ params.insert(make_pair(acl::DURABLE, std::string(durable ? "true" : "false")));
+ if (!acl->authorise(getConnection().getUserId(),acl::CREATE,acl::EXCHANGE,exchange,¶ms) )
+ throw NotAllowedException("ACL denied exhange declare request");
}
-
+
//TODO: implement autoDelete
Exchange::shared_ptr alternate;
if (!alternateExchange.empty()) {
@@ -90,6 +100,13 @@
checkType(response.first, type);
checkAlternate(response.first, alternate);
}
+
+ ManagementAgent* agent = ManagementAgent::Singleton::getInstance();
+ if (agent)
+ agent->raiseEvent(_qmf::EventExchangeDeclare(getConnection().getUrl(), getConnection().getUserId(), exchange, type,
+ alternateExchange, durable, false, args,
+ response.second ? "created" : "existing"));
+
}catch(UnknownExchangeTypeException& e){
throw CommandInvalidException(QPID_MSG("Exchange type not implemented: " << type));
}
@@ -106,38 +123,37 @@
void SessionAdapter::ExchangeHandlerImpl::checkAlternate(Exchange::shared_ptr exchange, Exchange::shared_ptr alternate)
{
if (alternate && alternate != exchange->getAlternate())
- throw NotAllowedException(
- QPID_MSG("Exchange declared with alternate-exchange "
- << exchange->getAlternate()->getName() << ", requested "
- << alternate->getName()));
+ throw NotAllowedException(QPID_MSG("Exchange declared with alternate-exchange "
+ << exchange->getAlternate()->getName() << ", requested "
+ << alternate->getName()));
}
-void SessionAdapter::ExchangeHandlerImpl::delete_(const string& name, bool /*ifUnused*/){
-
- AclModule* acl = getBroker().getAcl();
- if (acl)
- {
- if (!acl->authorise(getConnection().getUserId(),acl::DELETE,acl::EXCHANGE,name,NULL) )
- throw NotAllowedException("ACL denied exhange delete request");
+void SessionAdapter::ExchangeHandlerImpl::delete_(const string& name, bool /*ifUnused*/)
+{
+ AclModule* acl = getBroker().getAcl();
+ if (acl) {
+ if (!acl->authorise(getConnection().getUserId(),acl::DELETE,acl::EXCHANGE,name,NULL) )
+ throw NotAllowedException("ACL denied exhange delete request");
}
-
//TODO: implement unused
Exchange::shared_ptr exchange(getBroker().getExchanges().get(name));
if (exchange->inUseAsAlternate()) throw NotAllowedException(QPID_MSG("Exchange in use as alternate-exchange."));
if (exchange->isDurable()) getBroker().getStore().destroy(*exchange);
if (exchange->getAlternate()) exchange->getAlternate()->decAlternateUsers();
getBroker().getExchanges().destroy(name);
-}
+
+ ManagementAgent* agent = ManagementAgent::Singleton::getInstance();
+ if (agent)
+ agent->raiseEvent(_qmf::EventExchangeDelete(getConnection().getUrl(), getConnection().getUserId(), name));
+}
ExchangeQueryResult SessionAdapter::ExchangeHandlerImpl::query(const string& name)
{
-
- AclModule* acl = getBroker().getAcl();
- if (acl)
- {
- if (!acl->authorise(getConnection().getUserId(),acl::ACCESS,acl::EXCHANGE,name,NULL) )
- throw NotAllowedException("ACL denied exhange query request");
+ AclModule* acl = getBroker().getAcl();
+ if (acl) {
+ if (!acl->authorise(getConnection().getUserId(),acl::ACCESS,acl::EXCHANGE,name,NULL) )
+ throw NotAllowedException("ACL denied exhange query request");
}
try {
@@ -147,15 +163,15 @@
return ExchangeQueryResult("", false, true, FieldTable());
}
}
+
void SessionAdapter::ExchangeHandlerImpl::bind(const string& queueName,
const string& exchangeName, const string& routingKey,
- const FieldTable& arguments){
-
- AclModule* acl = getBroker().getAcl();
- if (acl)
- {
- if (!acl->authorise(getConnection().getUserId(),acl::BIND,acl::EXCHANGE,exchangeName,routingKey) )
- throw NotAllowedException("ACL denied exhange bind request");
+ const FieldTable& arguments)
+{
+ AclModule* acl = getBroker().getAcl();
+ if (acl) {
+ if (!acl->authorise(getConnection().getUserId(),acl::BIND,acl::EXCHANGE,exchangeName,routingKey) )
+ throw NotAllowedException("ACL denied exhange bind request");
}
Queue::shared_ptr queue = getQueue(queueName);
@@ -167,30 +183,29 @@
if (exchange->isDurable() && queue->isDurable()) {
getBroker().getStore().bind(*exchange, *queue, routingKey, arguments);
}
+
+ ManagementAgent* agent = ManagementAgent::Singleton::getInstance();
+ if (agent)
+ agent->raiseEvent(_qmf::EventBind(getConnection().getUrl(), getConnection().getUserId(), exchangeName, queueName, exchangeRoutingKey, arguments));
}
}else{
- throw NotFoundException(
- "Bind failed. No such exchange: " + exchangeName);
+ throw NotFoundException("Bind failed. No such exchange: " + exchangeName);
}
}
-void
-SessionAdapter::ExchangeHandlerImpl::unbind(const string& queueName,
- const string& exchangeName,
- const string& routingKey)
+void SessionAdapter::ExchangeHandlerImpl::unbind(const string& queueName,
+ const string& exchangeName,
+ const string& routingKey)
{
-
- AclModule* acl = getBroker().getAcl();
- if (acl)
- {
+ AclModule* acl = getBroker().getAcl();
+ if (acl) {
std::map<acl::Property, std::string> params;
- params.insert(make_pair(acl::QUEUENAME, queueName));
- params.insert(make_pair(acl::ROUTINGKEY, routingKey));
- if (!acl->authorise(getConnection().getUserId(),acl::UNBIND,acl::EXCHANGE,exchangeName,¶ms) )
- throw NotAllowedException("ACL denied exchange unbind request");
+ params.insert(make_pair(acl::QUEUENAME, queueName));
+ params.insert(make_pair(acl::ROUTINGKEY, routingKey));
+ if (!acl->authorise(getConnection().getUserId(),acl::UNBIND,acl::EXCHANGE,exchangeName,¶ms) )
+ throw NotAllowedException("ACL denied exchange unbind request");
}
-
Queue::shared_ptr queue = getQueue(queueName);
if (!queue.get()) throw NotFoundException("Unbind failed. No such exchange: " + exchangeName);
@@ -198,10 +213,14 @@
if (!exchange.get()) throw NotFoundException("Unbind failed. No such exchange: " + exchangeName);
//TODO: revise unbind to rely solely on binding key (not args)
- if (exchange->unbind(queue, routingKey, 0) && exchange->isDurable() && queue->isDurable()) {
- getBroker().getStore().unbind(*exchange, *queue, routingKey, FieldTable());
+ if (exchange->unbind(queue, routingKey, 0)) {
+ if (exchange->isDurable() && queue->isDurable())
+ getBroker().getStore().unbind(*exchange, *queue, routingKey, FieldTable());
+
+ ManagementAgent* agent = ManagementAgent::Singleton::getInstance();
+ if (agent)
+ agent->raiseEvent(_qmf::EventUnbind(getConnection().getUrl(), getConnection().getUserId(), exchangeName, queueName, routingKey));
}
-
}
ExchangeBoundResult SessionAdapter::ExchangeHandlerImpl::bound(const std::string& exchangeName,
@@ -209,16 +228,15 @@
const std::string& key,
const framing::FieldTable& args)
{
- AclModule* acl = getBroker().getAcl();
- if (acl)
- {
+ AclModule* acl = getBroker().getAcl();
+ if (acl) {
std::map<acl::Property, std::string> params;
- params.insert(make_pair(acl::QUEUENAME, queueName));
- params.insert(make_pair(acl::ROUTINGKEY, key));
- if (!acl->authorise(getConnection().getUserId(),acl::CREATE,acl::EXCHANGE,exchangeName,¶ms) )
- throw NotAllowedException("ACL denied exhange bound request");
+ params.insert(make_pair(acl::QUEUENAME, queueName));
+ params.insert(make_pair(acl::ROUTINGKEY, key));
+ if (!acl->authorise(getConnection().getUserId(),acl::CREATE,acl::EXCHANGE,exchangeName,¶ms) )
+ throw NotAllowedException("ACL denied exhange bound request");
}
-
+
Exchange::shared_ptr exchange;
try {
exchange = getBroker().getExchanges().get(exchangeName);
@@ -279,13 +297,12 @@
QueueQueryResult SessionAdapter::QueueHandlerImpl::query(const string& name)
{
- AclModule* acl = getBroker().getAcl();
- if (acl)
- {
- if (!acl->authorise(getConnection().getUserId(),acl::ACCESS,acl::QUEUE,name,NULL) )
- throw NotAllowedException("ACL denied queue query request");
+ AclModule* acl = getBroker().getAcl();
+ if (acl) {
+ if (!acl->authorise(getConnection().getUserId(),acl::ACCESS,acl::QUEUE,name,NULL) )
+ throw NotAllowedException("ACL denied queue query request");
}
-
+
Queue::shared_ptr queue = session.getBroker().getQueues().find(name);
if (queue) {
@@ -305,20 +322,19 @@
}
void SessionAdapter::QueueHandlerImpl::declare(const string& name, const string& alternateExchange,
- bool passive, bool durable, bool exclusive,
- bool autoDelete, const qpid::framing::FieldTable& arguments){
-
- AclModule* acl = getBroker().getAcl();
- if (acl)
- {
+ bool passive, bool durable, bool exclusive,
+ bool autoDelete, const qpid::framing::FieldTable& arguments)
+{
+ AclModule* acl = getBroker().getAcl();
+ if (acl) {
std::map<acl::Property, std::string> params;
- params.insert(make_pair(acl::ALTERNATE, alternateExchange));
- params.insert(make_pair(acl::PASSIVE, std::string(passive ? "true" : "false") ));
- params.insert(make_pair(acl::DURABLE, std::string(durable ? "true" : "false")));
- params.insert(make_pair(acl::EXCLUSIVE, std::string(exclusive ? "true" : "false")));
- params.insert(make_pair(acl::AUTODELETE, std::string(autoDelete ? "true" : "false")));
- if (!acl->authorise(getConnection().getUserId(),acl::CREATE,acl::QUEUE,name,¶ms) )
- throw NotAllowedException("ACL denied queue create request");
+ params.insert(make_pair(acl::ALTERNATE, alternateExchange));
+ params.insert(make_pair(acl::PASSIVE, std::string(passive ? "true" : "false") ));
+ params.insert(make_pair(acl::DURABLE, std::string(durable ? "true" : "false")));
+ params.insert(make_pair(acl::EXCLUSIVE, std::string(exclusive ? "true" : "false")));
+ params.insert(make_pair(acl::AUTODELETE, std::string(autoDelete ? "true" : "false")));
+ if (!acl->authorise(getConnection().getUserId(),acl::CREATE,acl::QUEUE,name,¶ms) )
+ throw NotAllowedException("ACL denied queue create request");
}
Exchange::shared_ptr alternate;
@@ -327,17 +343,16 @@
}
Queue::shared_ptr queue;
if (passive && !name.empty()) {
- queue = getQueue(name);
+ queue = getQueue(name);
//TODO: check alternate-exchange is as expected
} else {
- std::pair<Queue::shared_ptr, bool> queue_created =
- getBroker().getQueues().declare(
- name, durable,
- autoDelete,
- exclusive ? this : 0);
- queue = queue_created.first;
- assert(queue);
- if (queue_created.second) { // This is a new queue
+ std::pair<Queue::shared_ptr, bool> queue_created =
+ getBroker().getQueues().declare(name, durable,
+ autoDelete,
+ exclusive ? this : 0);
+ queue = queue_created.first;
+ assert(queue);
+ if (queue_created.second) { // This is a new queue
if (alternate) {
queue->setAlternateExchange(alternate);
alternate->incAlternateUsers();
@@ -346,44 +361,50 @@
//apply settings & create persistent record if required
queue_created.first->create(arguments);
- //add default binding:
- getBroker().getExchanges().getDefault()->bind(queue, name, 0);
+ //add default binding:
+ getBroker().getExchanges().getDefault()->bind(queue, name, 0);
queue->bound(getBroker().getExchanges().getDefault()->getName(), name, arguments);
//handle automatic cleanup:
- if (exclusive) {
- exclusiveQueues.push_back(queue);
- }
- } else {
+ if (exclusive) {
+ exclusiveQueues.push_back(queue);
+ }
+ } else {
if (exclusive && queue->setExclusiveOwner(this)) {
- exclusiveQueues.push_back(queue);
+ exclusiveQueues.push_back(queue);
}
}
+
+ ManagementAgent* agent = ManagementAgent::Singleton::getInstance();
+ if (agent)
+ agent->raiseEvent(_qmf::EventQueueDeclare(getConnection().getUrl(), getConnection().getUserId(),
+ name, durable, exclusive, autoDelete, arguments,
+ queue_created.second ? "created" : "existing"));
}
+
if (exclusive && !queue->isExclusiveOwner(this))
- throw ResourceLockedException(
- QPID_MSG("Cannot grant exclusive access to queue "
- << queue->getName()));
+ throw ResourceLockedException(QPID_MSG("Cannot grant exclusive access to queue "
+ << queue->getName()));
}
void SessionAdapter::QueueHandlerImpl::purge(const string& queue){
- AclModule* acl = getBroker().getAcl();
- if (acl)
- {
- if (!acl->authorise(getConnection().getUserId(),acl::PURGE,acl::QUEUE,queue,NULL) )
- throw NotAllowedException("ACL denied queue purge request");
+ AclModule* acl = getBroker().getAcl();
+ if (acl)
+ {
+ if (!acl->authorise(getConnection().getUserId(),acl::PURGE,acl::QUEUE,queue,NULL) )
+ throw NotAllowedException("ACL denied queue purge request");
}
getQueue(queue)->purge();
}
void SessionAdapter::QueueHandlerImpl::delete_(const string& queue, bool ifUnused, bool ifEmpty){
- AclModule* acl = getBroker().getAcl();
- if (acl)
- {
- if (!acl->authorise(getConnection().getUserId(),acl::DELETE,acl::QUEUE,queue,NULL) )
- throw NotAllowedException("ACL denied queue delete request");
+ AclModule* acl = getBroker().getAcl();
+ if (acl)
+ {
+ if (!acl->authorise(getConnection().getUserId(),acl::DELETE,acl::QUEUE,queue,NULL) )
+ throw NotAllowedException("ACL denied queue delete request");
}
ChannelException error(0, "");
@@ -401,6 +422,10 @@
q->destroy();
getBroker().getQueues().destroy(queue);
q->unbind(getBroker().getExchanges(), q);
+
+ ManagementAgent* agent = ManagementAgent::Singleton::getInstance();
+ if (agent)
+ agent->raiseEvent(_qmf::EventQueueDelete(getConnection().getUrl(), getConnection().getUserId(), queue));
}
}
@@ -441,12 +466,12 @@
const FieldTable& arguments)
{
- AclModule* acl = getBroker().getAcl();
- if (acl)
- {
- // add flags as needed
- if (!acl->authorise(getConnection().getUserId(),acl::CONSUME,acl::QUEUE,queueName,NULL) )
- throw NotAllowedException("ACL denied Queue subscribe request");
+ AclModule* acl = getBroker().getAcl();
+ if (acl)
+ {
+ // add flags as needed
+ if (!acl->authorise(getConnection().getUserId(),acl::CONSUME,acl::QUEUE,queueName,NULL) )
+ throw NotAllowedException("ACL denied Queue subscribe request");
}
Queue::shared_ptr queue = getQueue(queueName);
@@ -457,12 +482,21 @@
state.consume(MessageDelivery::getMessageDeliveryToken(destination, acceptMode, acquireMode),
tag, queue, false, //TODO get rid of no-local
acceptMode == 0, acquireMode == 0, exclusive, &arguments);
+
+ ManagementAgent* agent = ManagementAgent::Singleton::getInstance();
+ if (agent)
+ agent->raiseEvent(_qmf::EventSubscribe(getConnection().getUrl(), getConnection().getUserId(),
+ queueName, destination, exclusive, arguments));
}
void
SessionAdapter::MessageHandlerImpl::cancel(const string& destination )
{
state.cancel(destination);
+
+ ManagementAgent* agent = ManagementAgent::Singleton::getInstance();
+ if (agent)
+ agent->raiseEvent(_qmf::EventUnsubscribe(getConnection().getUrl(), getConnection().getUserId(), destination));
}
void
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/framing/Buffer.h
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/framing/Buffer.h?rev=702651&r1=702650&r2=702651&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/framing/Buffer.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/framing/Buffer.h Tue Oct 7 14:47:35 2008
@@ -28,7 +28,9 @@
namespace qpid {
namespace framing {
-struct OutOfBounds : qpid::Exception {};
+struct OutOfBounds : qpid::Exception {
+ OutOfBounds() : qpid::Exception(std::string("Out of Bounds")) {}
+};
class Content;
class FieldTable;
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/management/Manageable.h
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/management/Manageable.h?rev=702651&r1=702650&r2=702651&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/management/Manageable.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/management/Manageable.h Tue Oct 7 14:47:35 2008
@@ -45,6 +45,7 @@
static const status_t STATUS_INVALID_PARAMETER = 4;
static const status_t STATUS_FEATURE_NOT_IMPLEMENTED = 5;
static const status_t STATUS_FORBIDDEN = 6;
+ static const status_t STATUS_EXCEPTION = 7;
static const status_t STATUS_USER = 0x00010000;
// Every "Manageable" object must hold a reference to exactly one