You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by gs...@apache.org on 2007/08/21 17:51:47 UTC
svn commit: r568174 [1/3] - in /incubator/qpid/trunk/qpid: cpp/
cpp/docs/api/ cpp/gentools/ cpp/rubygen/ cpp/rubygen/samples/
cpp/rubygen/templates/ cpp/src/ cpp/src/qpid/broker/ cpp/src/qpid/client/
cpp/src/qpid/framing/ cpp/src/tests/ python/mllib/ p...
Author: gsim
Date: Tue Aug 21 08:51:41 2007
New Revision: 568174
URL: http://svn.apache.org/viewvc?rev=568174&view=rev
Log:
Refresh of transitional xml to more closely reflect latest specification
Initial execution-result support (not yet handled on c++ client)
Generation is now all done through the ruby code (it is a little slower at present I'm afraid, will try to speed it up over the next weeks)
Added:
incubator/qpid/trunk/qpid/cpp/rubygen/templates/InvocationVisitor.rb
incubator/qpid/trunk/qpid/cpp/rubygen/templates/Operations.rb
- copied, changed from r565290, incubator/qpid/trunk/qpid/cpp/rubygen/samples/Operations.rb
incubator/qpid/trunk/qpid/cpp/rubygen/templates/structs.rb
incubator/qpid/trunk/qpid/cpp/src/qpid/framing/AMQP_HighestVersion.h (with props)
incubator/qpid/trunk/qpid/cpp/src/qpid/framing/StructHelper.h (with props)
Removed:
incubator/qpid/trunk/qpid/cpp/gentools/
incubator/qpid/trunk/qpid/cpp/rubygen/samples/
incubator/qpid/trunk/qpid/cpp/src/tests/Blob
Modified:
incubator/qpid/trunk/qpid/cpp/README
incubator/qpid/trunk/qpid/cpp/docs/api/ (props changed)
incubator/qpid/trunk/qpid/cpp/rubygen/amqpgen.rb
incubator/qpid/trunk/qpid/cpp/rubygen/cppgen.rb
incubator/qpid/trunk/qpid/cpp/rubygen/generate
incubator/qpid/trunk/qpid/cpp/rubygen/templates/Session.rb
incubator/qpid/trunk/qpid/cpp/src/Makefile.am
incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerAdapter.cpp
incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerAdapter.h
incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerChannel.cpp
incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerQueue.h
incubator/qpid/trunk/qpid/cpp/src/qpid/broker/DtxHandlerImpl.cpp
incubator/qpid/trunk/qpid/cpp/src/qpid/broker/DtxHandlerImpl.h
incubator/qpid/trunk/qpid/cpp/src/qpid/broker/MessageHandlerImpl.cpp
incubator/qpid/trunk/qpid/cpp/src/qpid/broker/MessageHandlerImpl.h
incubator/qpid/trunk/qpid/cpp/src/qpid/broker/SemanticHandler.cpp
incubator/qpid/trunk/qpid/cpp/src/qpid/broker/SemanticHandler.h
incubator/qpid/trunk/qpid/cpp/src/qpid/client/ (props changed)
incubator/qpid/trunk/qpid/cpp/src/qpid/client/ClientChannel.cpp
incubator/qpid/trunk/qpid/cpp/src/qpid/client/ClientChannel.h
incubator/qpid/trunk/qpid/cpp/src/qpid/client/Connection.h
incubator/qpid/trunk/qpid/cpp/src/qpid/client/ExecutionHandler.cpp
incubator/qpid/trunk/qpid/cpp/src/qpid/client/ExecutionHandler.h
incubator/qpid/trunk/qpid/cpp/src/qpid/framing/ (props changed)
incubator/qpid/trunk/qpid/cpp/src/qpid/framing/amqp_types_full.h
incubator/qpid/trunk/qpid/cpp/src/tests/Cluster_child.cpp
incubator/qpid/trunk/qpid/python/mllib/ (props changed)
incubator/qpid/trunk/qpid/python/qpid/peer.py
incubator/qpid/trunk/qpid/python/qpid/testlib.py
incubator/qpid/trunk/qpid/python/tests_0-10/alternate-exchange.py
incubator/qpid/trunk/qpid/python/tests_0-10/broker.py
incubator/qpid/trunk/qpid/python/tests_0-10/dtx.py
incubator/qpid/trunk/qpid/python/tests_0-10/example.py
incubator/qpid/trunk/qpid/python/tests_0-10/message.py
incubator/qpid/trunk/qpid/python/tests_0-10/query.py
incubator/qpid/trunk/qpid/python/tests_0-10/queue.py
incubator/qpid/trunk/qpid/python/tests_0-10/tx.py
incubator/qpid/trunk/qpid/specs/amqp-transitional.0-10.xml
Modified: incubator/qpid/trunk/qpid/cpp/README
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/README?rev=568174&r1=568173&r2=568174&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/README (original)
+++ incubator/qpid/trunk/qpid/cpp/README Tue Aug 21 08:51:41 2007
@@ -54,7 +54,7 @@
* libtool <http://www.gnu.org/software/libtool/> (1.5.22)
* doxygen <ftp://ftp.stack.nl/pub/users/dimitri/> (1.5.1)
* graphviz <http://www.graphviz.org/> (2.12)
- * JDK 5.0 <http://java.sun.com/j2se/1.5.0/> (1.5.0.11)
+ * ruby 1.8 <http://www.ruby-lang.org> (1.8.4)
=== Installing as root ===
Propchange: incubator/qpid/trunk/qpid/cpp/docs/api/
------------------------------------------------------------------------------
--- svn:ignore (original)
+++ svn:ignore Tue Aug 21 08:51:41 2007
@@ -7,3 +7,4 @@
developer.doxygen
user.doxygen
html.timestamp
+doxygen.log
Modified: incubator/qpid/trunk/qpid/cpp/rubygen/amqpgen.rb
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/rubygen/amqpgen.rb?rev=568174&r1=568173&r2=568174&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/rubygen/amqpgen.rb (original)
+++ incubator/qpid/trunk/qpid/cpp/rubygen/amqpgen.rb Tue Aug 21 08:51:41 2007
@@ -60,8 +60,7 @@
# Get AMQP type for a domain name.
def domain_type(name)
- domain=elements["/amqp/domain[@name='#{name}']"]
- (domain and domain.attributes["type"] or name)
+ @cache_domain_type ||= domain_type_impl(name)
end
# Get the AMQP type of this field.
@@ -70,6 +69,22 @@
dt=domain_type d if d
(dt or attributes["type"])
end
+
+ # determine whether this type is defined as a struct in the spec
+ def defined_as_struct()
+ @cache_defined_as_struct ||= defined_as_struct_impl()
+ end
+
+private
+ def domain_type_impl(name)
+ domain=elements["/amqp/domain[@name='#{name}']"]
+ (domain and domain.attributes["type"] or name)
+ end
+
+ def defined_as_struct_impl()
+ domain_name = attributes["domain"]
+ elements["/amqp/domain[@name='#{domain_name}']/struct"]
+ end
end
# AMQP method element
@@ -98,6 +113,18 @@
}
end
+ def has_result?()
+ @cache_has_result ||= elements["result/struct"]
+ end
+
+ def result_struct()
+ @cache_result_struct ||= has_result? ? AmqpStruct.new(elements["result/struct"], self) : nil
+ end
+
+ def is_server_method?()
+ @cache_is_server_method ||= elements["chassis[@name='server']"]
+ end
+
def request?() responds_to().empty?; end
def response?() not request?; end
end
@@ -116,7 +143,25 @@
def amqp_methods_on(chassis)
@cache_amqp_methods_on ||= { }
@cache_amqp_methods_on[chassis] ||= elements.collect("method/chassis[@name='#{chassis}']/..") { |m| AmqpMethod.new(m,self) }.sort_by_name
+ end
end
+
+
+# AMQP struct element.
+class AmqpStruct < AmqpElement
+ def initialize(xml,amqp) super; end
+
+ def type() attributes["type"]; end
+
+ def size() attributes["size"]; end
+
+ def result?() parent.name == "result"; end
+
+ def domain?() parent.name == "domain"; end
+
+ def fields()
+ @cache_fields ||= elements.collect("field") { |f| AmqpField.new(f,self); }
+ end
end
# AMQP root element.
@@ -152,9 +197,23 @@
@cache_amqp_classes ||= elements.collect("class") { |c|
AmqpClass.new(c,self) }.sort_by_name
end
+
+ def amqp_structs()
+ @cache_amqp_structs ||= amqp_result_structs + amqp_domain_structs
+ end
+
+ def amqp_domain_structs()
+ @cache_amqp_domain_structs ||= elements.collect("domain/struct") { |s| AmqpStruct.new(s, self) }
+ end
+
+ def amqp_result_structs()
+ @cache_amqp_result_structs ||= amqp_methods.collect { |m| m.result_struct }.compact
+ end
# Return all methods on all classes.
- def amqp_methods() amqp_classes.collect { |c| c.amqp_methods }.flatten; end
+ def amqp_methods()
+ @cache_amqp_methods ||= amqp_classes.collect { |c| c.amqp_methods }.flatten;
+ end
# Return all methods on chassis for all classes.
def amqp_methods_on(chassis)
Modified: incubator/qpid/trunk/qpid/cpp/rubygen/cppgen.rb
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/rubygen/cppgen.rb?rev=568174&r1=568173&r2=568174&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/rubygen/cppgen.rb (original)
+++ incubator/qpid/trunk/qpid/cpp/rubygen/cppgen.rb Tue Aug 21 08:51:41 2007
@@ -62,9 +62,11 @@
# Additional methods for AmqpField.
class AmqpField
def cppname() @cache_cppname ||= name.lcaps.cppsafe; end
- def cpptype() @cache_cpptype ||= amqp_root.param_type(field_type); end
+ def cpptype() @cache_cpptype ||= defined_as_struct ? "const " + field_type.caps + "&" : amqp_root.param_type(field_type); end
+ def cpp_member_type() @cache_cpp_member_type ||= defined_as_struct ? field_type.caps : amqp_root.member_type(field_type); end
def cppret_type() @cache_cpptype ||= amqp_root.return_type(field_type); end
def type_name () @type_name ||= cpptype+" "+cppname; end
+ def bit?() field_type == "bit" end
end
# Additional methods for AmqpMethod
@@ -82,6 +84,32 @@
end
end
+class AmqpStruct
+ def cppname()
+ @cache_cppname ||= cppname_impl()
+ end
+
+private
+
+ def cppname_impl()
+ #The name of the struct comes from the context it appears in:
+ #structs defined in a domain get their name from the domain
+ #element, structs defined in a result element get their name by
+ #appending 'Result' to the enclosing method name.
+ if (domain?)
+ parent.attributes["name"].caps
+ else
+ if (result?)
+ method = parent.parent
+ method.parent.attributes["name"].caps + method.attributes["name"].caps + "Result"
+ else
+ raise "Bad struct context: expected struct to be child of <domain> or <result>"
+ end
+ end
+
+ end
+end
+
# Additional methos for AmqpRoot
class AmqpRoot
# FIXME aconway 2007-06-20: fix u_int types, should be uint
@@ -97,6 +125,7 @@
"table"=>["FieldTable", "const FieldTable&", "const FieldTable&"],
"content"=>["Content", "const Content&", "const Content&"],
"rfc1982-long-set"=>["SequenceNumberSet", "const SequenceNumberSet&", "const SequenceNumberSet&"],
+ "long-struct"=>["string", "const string&"],
"uuid"=>["string", "const string&"] # FIXME should be: ["Uuid", "const Uuid&", "const Uuid&"]
}
Modified: incubator/qpid/trunk/qpid/cpp/rubygen/generate
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/rubygen/generate?rev=568174&r1=568173&r2=568174&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/rubygen/generate (original)
+++ incubator/qpid/trunk/qpid/cpp/rubygen/generate Tue Aug 21 08:51:41 2007
@@ -38,6 +38,10 @@
generator_files=Dir["**/*.rb"] << File.basename(__FILE__)
Dir.chdir dir
rgen_generator=generator_files.map{ |f| "$(rgen_dir)/#{f}" }
+ rgen_srcs=GenFiles.get.map{ |f| "#{Outdir}/#{f}" }
+ rgen_client_cpp = rgen_srcs.select { |f| f =~ /qpid\/client\/.+\.cpp$/ }
+ rgen_common_cpp = rgen_srcs.select { |f| f =~ /qpid\/framing\/.+\.cpp$/ }
+ rgen_h = rgen_srcs.select { |f| f =~ /.+\.h$/ }
File.open(makefile, 'w') { |out|
out << <<EOS
@@ -46,7 +50,13 @@
rgen_generator=#{rgen_generator.join(" \\\n ")}
-rgen_srcs=#{GenFiles.get.join(" \\\n ")}
+rgen_client_cpp=#{rgen_client_cpp.join(" \\\n ")}
+
+rgen_common_cpp=#{rgen_common_cpp.join(" \\\n ")}
+
+rgen_h=#{rgen_h.join(" \\\n ")}
+
+rgen_srcs=rgen_h rgen_client_cpp rgen_common_cpp
if GENERATE
$(rgen_srcs) $(srcdir)/#{File.basename makefile}: $(rgen_generator) $(specs)
Added: incubator/qpid/trunk/qpid/cpp/rubygen/templates/InvocationVisitor.rb
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/rubygen/templates/InvocationVisitor.rb?rev=568174&view=auto
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/rubygen/templates/InvocationVisitor.rb (added)
+++ incubator/qpid/trunk/qpid/cpp/rubygen/templates/InvocationVisitor.rb Tue Aug 21 08:51:41 2007
@@ -0,0 +1,105 @@
+#!/usr/bin/env ruby
+$: << ".." # Include .. in load path
+require 'cppgen'
+
+class InvocationVisitor < CppGen
+
+ def initialize(outdir, amqp)
+ super(outdir, amqp)
+ @namespace="qpid::framing"
+ @classname="InvocationVisitor"
+ @filename="qpid/framing/InvocationVisitor"
+ end
+
+ def invocation_args(m)
+ if (m.amqp_parent.name == "message" && (m.name == "transfer" || m.name == "append"))
+ "body"
+ else
+ m.param_names.collect {|p| "body.get" + p.caps + "()" }.join(",\n")
+ end
+ end
+
+ def null_visit(m)
+ genl "void InvocationVisitor::visit(const #{m.body_name}&){}"
+ end
+
+ def define_visit(m)
+ if (m.fields.size > 0)
+ body = "body"
+ else
+ body = "/*body*/"
+ end
+ gen <<EOS
+void InvocationVisitor::visit(const #{m.body_name}& #{body})
+{
+ AMQP_ServerOperations::#{m.amqp_parent.cppname}Handler* ptr(0);
+ if (invocable) {
+ ptr = dynamic_cast<AMQP_ServerOperations::#{m.amqp_parent.cppname}Handler*>(invocable);
+ } else {
+ ptr = ops->get#{m.amqp_parent.cppname}Handler();
+ }
+
+ if (ptr) {
+EOS
+ if (m.has_result?)
+ indent(2) { genl "encode<#{m.result_struct.cppname.caps}>(ptr->#{m.cppname}(#{invocation_args(m)}), result);" }
+ else
+ indent(2) { genl "ptr->#{m.cppname}(#{invocation_args(m)});" }
+ end
+
+ gen <<EOS
+ succeeded = true;
+ } else {
+ succeeded = false;
+ }
+}
+EOS
+ end
+
+ def generate()
+ h_file("#{@filename}") {
+ include "AMQP_ServerOperations.h"
+ include "MethodBodyConstVisitor.h"
+ include "qpid/framing/StructHelper.h"
+ namespace(@namespace) {
+ cpp_class("InvocationVisitor", ["public MethodBodyConstVisitor", "private StructHelper"]) {
+ indent {
+ genl "AMQP_ServerOperations* const ops;"
+ genl "Invocable* const invocable;"
+ genl "std::string result;"
+ genl "bool succeeded;"
+ }
+ genl "public:"
+ indent {
+ genl "InvocationVisitor(AMQP_ServerOperations* _ops) : ops(_ops), invocable(0) {}"
+ genl "InvocationVisitor(Invocable* _invocable) : ops(0), invocable(_invocable) {}"
+ genl "const std::string& getResult() const { return result; }"
+ genl "const bool hasResult() const { return !result.empty(); }"
+ genl "bool wasHandled() const { return succeeded; }"
+ genl "void clear();"
+ genl "virtual ~InvocationVisitor() {}"
+ }
+ @amqp.amqp_methods.each { |m| genl "void visit(const #{m.body_name}&);" }
+ }
+ }
+ }
+
+ cpp_file("#{@filename}") {
+ include "InvocationVisitor.h"
+ @amqp.amqp_methods.each { |m| include m.body_name }
+ namespace(@namespace) {
+ genl "void InvocationVisitor::clear() {"
+ indent {
+ genl "succeeded = false;"
+ genl "result.clear();"
+ }
+ genl "}"
+ genl
+ @amqp.amqp_methods.each { |m| m.is_server_method? ? define_visit(m) : null_visit(m) }
+ }
+ }
+ end
+end
+
+InvocationVisitor.new(Outdir, Amqp).generate();
+
Copied: incubator/qpid/trunk/qpid/cpp/rubygen/templates/Operations.rb (from r565290, incubator/qpid/trunk/qpid/cpp/rubygen/samples/Operations.rb)
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/rubygen/templates/Operations.rb?p2=incubator/qpid/trunk/qpid/cpp/rubygen/templates/Operations.rb&p1=incubator/qpid/trunk/qpid/cpp/rubygen/samples/Operations.rb&r1=565290&r2=568174&rev=568174&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/rubygen/samples/Operations.rb (original)
+++ incubator/qpid/trunk/qpid/cpp/rubygen/templates/Operations.rb Tue Aug 21 08:51:41 2007
@@ -13,46 +13,69 @@
end
def handler_method (m)
- gen "\nvirtual void #{m.cppname}("
- gen m.signature.join(",\n")
- gen ") = 0;\n"
+ if (m.has_result?)
+ return_type = "#{m.amqp_parent.name.caps}#{m.cppname.caps}Result"
+ else
+ return_type = "void"
+ end
+ if (m.amqp_parent.name == "message" && (m.name == "transfer" || m.name == "append"))
+ gen "\nvirtual #{return_type} #{m.cppname}(const framing::AMQMethodBody& context) = 0;\n"
+ else
+ gen "\nvirtual #{return_type} #{m.cppname}("
+ gen m.signature.join(",\n")
+ gen ") = 0;\n"
+ end
end
def handler_classname(c) c.name.caps+"Handler"; end
def handler_class(c)
- handlerclass=handler_classname c
- gen <<EOS
+ if (!c.amqp_methods_on(@chassis).empty?)
+ handlerclass=handler_classname c
+ gen <<EOS
// ==================== class #{handlerclass} ====================
-class #{handlerclass} : Invocable {
+class #{handlerclass} : public virtual Invocable {
// Constructors and destructors
public:
#{handlerclass}(){};
virtual ~#{handlerclass}() {}
// Protocol methods
EOS
- c.amqp_methods_on(@chassis).each { |m| handler_method(m) }
- gen <<EOS
+ c.amqp_methods_on(@chassis).each { |m| handler_method(m) }
+ gen <<EOS
}; // class #{handlerclass}
EOS
+ end
end
def handler_get(c)
- handlerclass=handler_classname c
- gen "virtual #{handlerclass}* get#{handlerclass}() = 0;\n"
+ if (!c.amqp_methods_on(@chassis).empty?)
+ handlerclass=handler_classname c
+ gen "virtual #{handlerclass}* get#{handlerclass}() = 0;\n"
+ end
end
def generate()
- h_file("#{@classname}.h") {
+ h_file("qpid/framing/#{@classname}.h") {
gen <<EOS
#include <sstream>
#include "qpid/framing/ProtocolVersion.h"
+#include "qpid/framing/amqp_structs.h"
namespace qpid {
namespace framing {
+class AMQMethodBody;
+
+class Invocable
+{
+protected:
+ Invocable() {}
+ virtual ~Invocable() {}
+};
+
class #{@classname} {
public:
@@ -60,8 +83,8 @@
virtual ProtocolVersion getVersion() const = 0;
- // Include framing constant declarations
- #include "AMQP_Constants.h"
+ // Include framing constant declarations - why?
+ //#include "qpid/framing/AMQP_Constants.h"
// Inner classes
EOS
@@ -74,7 +97,7 @@
indent { @amqp.amqp_classes.each { |c| handler_get(c) } }
gen <<EOS
}; /* class #{@classname} */
-}
+}}
EOS
}
end
Modified: incubator/qpid/trunk/qpid/cpp/rubygen/templates/Session.rb
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/rubygen/templates/Session.rb?rev=568174&r1=568173&r2=568174&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/rubygen/templates/Session.rb (original)
+++ incubator/qpid/trunk/qpid/cpp/rubygen/templates/Session.rb Tue Aug 21 08:51:41 2007
@@ -67,9 +67,9 @@
#include "qpid/framing/amqp_framing.h"
#include "qpid/framing/ProtocolVersion.h"
#include "qpid/framing/MethodContent.h"
-#include "ConnectionImpl.h"
-#include "Response.h"
-#include "SessionCore.h"
+#include "qpid/client/ConnectionImpl.h"
+#include "qpid/client/Response.h"
+#include "qpid/client/SessionCore.h"
namespace qpid {
namespace client {
Added: incubator/qpid/trunk/qpid/cpp/rubygen/templates/structs.rb
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/rubygen/templates/structs.rb?rev=568174&view=auto
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/rubygen/templates/structs.rb (added)
+++ incubator/qpid/trunk/qpid/cpp/rubygen/templates/structs.rb Tue Aug 21 08:51:41 2007
@@ -0,0 +1,345 @@
+#!/usr/bin/env ruby
+# Usage: output_directory xml_spec_file [xml_spec_file...]
+#
+$: << '..'
+require 'cppgen'
+
+class StructGen < CppGen
+
+ def initialize(outdir, amqp)
+ super(outdir, amqp)
+ end
+
+ EncodingMap={
+ "octet"=>"Octet",
+ "short"=>"Short",
+ "long"=>"Long",
+ "longlong"=>"LongLong",
+ "longstr"=>"LongString",
+ "shortstr"=>"ShortString",
+ "timestamp"=>"LongLong",
+ "uuid"=>"ShortString",#FIXME
+ "table"=>"FieldTable",
+ "content"=>"Content",
+ "long-struct"=>"LongString"
+ }
+ SizeMap={
+ "octet"=>"1",
+ "short"=>"2",
+ "long"=>"4",
+ "longlong"=>"8",
+ "timestamp"=>"8"
+ }
+
+ ValueTypes=["octet", "short", "long", "longlong", "timestamp"]
+
+ def printable_form(f)
+ if (f.cpptype == "u_int8_t")
+ return "(int) " + f.cppname
+ else
+ return f.cppname
+ end
+ end
+
+ def generate_encode(f, combined)
+ if (f.field_type == "bit")
+ genl "uint8_t #{f.cppname}_bits = #{f.cppname};"
+ count = 0
+ combined.each { |c| genl "#{f.cppname}_bits |= #{c.cppname} << #{count += 1};" }
+ genl "buffer.putOctet(#{f.cppname}_bits);"
+ else
+ encoded = EncodingMap[f.field_type]
+ if (encoded)
+ genl "buffer.put#{encoded}(#{f.cppname});"
+ else
+ genl "#{f.cppname}.encode(buffer);"
+ end
+ end
+ end
+
+ def generate_decode(f, combined)
+ if (f.field_type == "bit")
+ genl "uint8_t #{f.cppname}_bits = buffer.getOctet();"
+ genl "#{f.cppname} = 1 & #{f.cppname}_bits;"
+ count = 0
+ combined.each { |c| genl "#{c.cppname} = (1 << #{count += 1}) & #{f.cppname}_bits;" }
+ else
+ encoded = EncodingMap[f.field_type]
+ if (encoded)
+ if (ValueTypes.include?(f.field_type))
+ genl "#{f.cppname} = buffer.get#{encoded}();"
+ else
+ genl "buffer.get#{encoded}(#{f.cppname});"
+ end
+ else
+ genl "#{f.cppname}.decode(buffer);"
+ end
+ end
+ end
+
+ def generate_size(f, combined)
+ if (f.field_type == "bit")
+ names = ([f] + combined).collect {|g| g.cppname}
+ genl "+ 1 //#{names.join(", ")}"
+ else
+ size = SizeMap[f.field_type]
+ if (size)
+ genl "+ #{size} //#{f.cppname}"
+ elsif (f.cpp_member_type == "SequenceNumberSet")
+ genl "+ #{f.cppname}.encodedSize()"
+ else
+ encoded = EncodingMap[f.field_type]
+ gen "+ 4 " if encoded == "LongString"
+ gen "+ 1 " if encoded == "ShortString"
+ genl "+ #{f.cppname}.size()"
+ end
+ end
+ end
+
+ def process_fields(s)
+ last = nil
+ count = 0
+ bits = []
+ s.fields.each {
+ |f| if (last and last.bit? and f.bit? and count < 7)
+ count += 1
+ bits << f
+ else
+ if (last and last.bit?)
+ yield last, bits
+ count = 0
+ bits = []
+ end
+ if (not f.bit?)
+ yield f
+ end
+ last = f
+ end
+ }
+ if (last and last.bit?)
+ yield last, bits
+ end
+ end
+
+ def methodbody_extra_defs(s)
+ gen <<EOS
+ using AMQMethodBody::accept;
+ void accept(MethodBodyConstVisitor& v) const { v.visit(*this); }
+
+ inline ClassId amqpClassId() const { return CLASS_ID; }
+ inline MethodId amqpMethodId() const { return METHOD_ID; }
+EOS
+ if (s.is_server_method?)
+ gen <<EOS
+ void invoke(AMQP_ServerOperations& target)
+ {
+ target.get#{s.amqp_parent.cppname}Handler()->#{s.cppname}
+ (
+EOS
+ if (s.amqp_parent.name == "message" && (s.name == "transfer" || s.name == "append"))
+ indent(4) { genl "*this" }
+ else
+ indent(4) { genl s.param_names.join(",\n") }
+ end
+
+ genl <<EOS
+ );
+ }
+
+ bool invoke(Invocable* target)
+ {
+ AMQP_ServerOperations::#{s.amqp_parent.cppname}Handler* ptr
+ = dynamic_cast<AMQP_ServerOperations::#{s.amqp_parent.cppname}Handler*>(target);
+ if (ptr) {
+ ptr->#{s.cppname}(
+EOS
+ if (s.amqp_parent.name == "message" && (s.name == "transfer" || s.name == "append"))
+ indent(5) { genl "*this" }
+ else
+ indent(5) { genl s.param_names.join(",\n") }
+ end
+
+ gen <<EOS
+ );
+ return true;
+ } else {
+ return false;
+ }
+ }
+EOS
+ end
+ end
+
+ def define_constructor(name, s)
+ if (s.fields.size > 0)
+ genl "#{name}("
+ if (s.kind_of? AmqpMethod)
+ indent {gen "ProtocolVersion, "}
+ end
+ indent { gen s.fields.collect { |f| "#{f.cpptype} _#{f.cppname}" }.join(",\n") }
+ gen ")"
+ genl ": " if s.fields.size > 0
+ indent { gen s.fields.collect { |f| " #{f.cppname}(_#{f.cppname})" }.join(",\n") }
+ genl " {}"
+ end
+ if (s.kind_of? AmqpMethod)
+ genl "#{name}(ProtocolVersion=ProtocolVersion()) {}"
+ end
+ end
+
+ def define_accessors(f)
+ genl "void set#{f.name.caps}(#{f.cpptype} _#{f.cppname}) { #{f.cppname} = _#{f.cppname}; }"
+ genl "#{f.cpptype} get#{f.name.caps}() const { return #{f.cppname}; }"
+ end
+
+ def define_struct(s)
+ classname = s.cppname
+ inheritance = ""
+ if (s.kind_of? AmqpMethod)
+ classname = s.body_name
+ inheritance = ": public AMQMethodBody"
+ end
+
+ h_file("qpid/framing/#{classname}.h") {
+ if (s.kind_of? AmqpMethod)
+ gen <<EOS
+#include "qpid/framing/AMQMethodBody.h"
+#include "qpid/framing/AMQP_ServerOperations.h"
+#include "qpid/framing/MethodBodyConstVisitor.h"
+EOS
+ end
+
+ #need to include any nested struct definitions
+ s.fields.select {|f| f.defined_as_struct }.map {|f| include f.name.caps}
+
+ gen <<EOS
+
+#include <ostream>
+#include "qpid/framing/amqp_types_full.h"
+
+namespace qpid {
+namespace framing {
+
+class #{classname} #{inheritance} {
+EOS
+ indent { s.fields.each { |f| genl "#{f.cpp_member_type} #{f.cppname};" } }
+ genl "public:"
+ if (s.kind_of? AmqpMethod)
+ indent { genl "static const ClassId CLASS_ID = #{s.amqp_parent.index};" }
+ indent { genl "static const MethodId METHOD_ID = #{s.index};" }
+ end
+
+ if (s.kind_of? AmqpStruct)
+ if (s.type)
+ if (s.result?)
+ #as result structs have types that are only unique to the
+ #class, they have a class dependent qualifier added to them
+ #(this is inline with current python code but a formal
+ #solution is expected from the WG
+ indent { genl "static const uint16_t TYPE = #{s.type} + #{s.amqp_parent.amqp_parent.index} * 256;" }
+ else
+ indent { genl "static const uint16_t TYPE = #{s.type};" }
+ end
+ end
+ end
+
+ indent {
+ define_constructor(classname, s)
+ genl ""
+ s.fields.each { |f| define_accessors(f) }
+ }
+ if (s.kind_of? AmqpMethod)
+ methodbody_extra_defs(s)
+ end
+ if (s.kind_of? AmqpStruct)
+ indent {genl "friend std::ostream& operator<<(std::ostream&, const #{classname}&);" }
+ end
+
+ gen <<EOS
+ void encode(Buffer&) const;
+ void decode(Buffer&, uint32_t=0);
+ uint32_t size() const;
+ void print(std::ostream& out) const;
+}; /* class #{classname} */
+
+}}
+EOS
+ }
+ cpp_file("qpid/framing/#{classname}.cpp") {
+ if (s.fields.size > 0)
+ buffer = "buffer"
+ else
+ buffer = "/*buffer*/"
+ end
+ gen <<EOS
+#include "#{classname}.h"
+
+using namespace qpid::framing;
+
+void #{classname}::encode(Buffer& #{buffer}) const
+{
+EOS
+ indent { process_fields(s) { |f, combined| generate_encode(f, combined) } }
+ gen <<EOS
+}
+
+void #{classname}::decode(Buffer& #{buffer}, uint32_t /*size*/)
+{
+EOS
+ indent { process_fields(s) { |f, combined| generate_decode(f, combined) } }
+ gen <<EOS
+}
+
+uint32_t #{classname}::size() const
+{
+ return 0
+EOS
+ indent { process_fields(s) { |f, combined| generate_size(f, combined) } }
+ gen <<EOS
+ ;
+}
+
+void #{classname}::print(std::ostream& out) const
+{
+ out << "#{classname}: ";
+EOS
+ copy = Array.new(s.fields)
+ f = copy.shift
+
+ indent {
+ genl "out << \"#{f.name}=\" << #{printable_form(f)};" if f
+ copy.each { |f| genl "out << \"; #{f.name}=\" << #{printable_form(f)};" }
+ }
+ gen <<EOS
+}
+EOS
+
+ if (s.kind_of? AmqpStruct)
+ gen <<EOS
+namespace qpid{
+namespace framing{
+
+ std::ostream& operator<<(std::ostream& out, const #{classname}& s)
+ {
+ s.print(out);
+ return out;
+ }
+
+}
+}
+EOS
+ end
+
+}
+ end
+
+ def generate()
+ @amqp.amqp_structs.each { |s| define_struct(s) }
+ @amqp.amqp_methods.each { |m| define_struct(m) }
+ #generate a single include file containing the list of structs for convenience
+ h_file("qpid/framing/amqp_structs.h") { @amqp.amqp_structs.each { |s| genl "#include \"#{s.cppname}.h\"" } }
+ end
+end
+
+StructGen.new(ARGV[0], Amqp).generate()
+
Modified: incubator/qpid/trunk/qpid/cpp/src/Makefile.am
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/Makefile.am?rev=568174&r1=568173&r2=568174&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/Makefile.am (original)
+++ incubator/qpid/trunk/qpid/cpp/src/Makefile.am Tue Aug 21 08:51:41 2007
@@ -14,37 +14,23 @@
# AMQP_XML is defined in ../configure.ac
specs=@AMQP_XML@ $(top_srcdir)/xml/cluster.xml
-EXTRA_DIST += generate.mk generate.sh $(generated_cpp) $(generated_h) $(rgen_srcs)
+EXTRA_DIST += $(rgen_h)
-if GENERATE
-
-# Java code generator.
-# Must generate into a separate gen directory because otherwise
-# there's no way to figure out which files are generated.
-
-gentools_dir=$(top_srcdir)/gentools
-
-$(srcdir)/generate.mk $(generated_cpp) $(generated_h): generate.timestamp
-generate.timestamp: generate.sh $(specs) $(generator)
- env gentools_dir=$(gentools_dir) specs="$(specs)" $(srcdir)/generate.sh
- touch $@
-# Empty rule in case a generator file is renamed/removed.
-$(generator):
+if GENERATE
maintainer-clean-local:
rm -rf gen
# Ruby generator.
rgen_dir=$(top_srcdir)/rubygen
-rgen_cmd=ruby -I $(rgen_dir) $(rgen_dir)/generate . $(specs) all $(srcdir)/rubygen.mk
+rgen_cmd=ruby -I $(rgen_dir) $(rgen_dir)/generate $(srcdir)/gen $(specs) all $(srcdir)/rubygen.mk
endif # GENERATE
-include $(srcdir)/generate.mk
include $(srcdir)/rubygen.mk
-DISTCLEANFILES=generate.mk rubygen.mk
+DISTCLEANFILES=rubygen.mk
# Code generated by C++
noinst_PROGRAMS=generate_MethodHolderMaxSize_h
@@ -110,6 +96,7 @@
$(LIB_CLOCK_GETTIME)
libqpidcommon_la_SOURCES = \
+ $(rgen_common_cpp) \
$(platform_src) \
qpid/framing/AMQBody.cpp \
qpid/framing/AMQMethodBody.cpp \
@@ -137,13 +124,10 @@
qpid/framing/FrameHandler.h \
qpid/framing/HandlerUpdater.h \
qpid/framing/Blob.h \
- qpid/framing/AMQP_ClientProxy.cpp \
- qpid/framing/AMQP_ServerProxy.cpp \
qpid/framing/variant.h \
- gen/qpid/framing/AMQP_HighestVersion.h \
+ qpid/framing/AMQP_HighestVersion.h \
qpid/framing/Blob.cpp \
qpid/framing/MethodHolder.h qpid/framing/MethodHolder.cpp \
- qpid/framing/MethodHolder_construct.cpp \
qpid/framing/MethodHolderMaxSize.h \
qpid/Exception.cpp \
qpid/Plugin.h \
@@ -217,6 +201,7 @@
libqpidclient_la_LIBADD = libqpidcommon.la
libqpidclient_la_SOURCES = \
+ $(rgen_client_cpp) \
qpid/client/ClientConnection.cpp \
qpid/client/ClientChannel.cpp \
qpid/client/ClientExchange.cpp \
@@ -233,12 +218,12 @@
qpid/client/FutureResponse.cpp \
qpid/client/FutureFactory.cpp \
qpid/client/ReceivedContent.cpp \
- qpid/client/Session.cpp \
qpid/client/SessionCore.cpp \
qpid/client/StateManager.cpp
nobase_include_HEADERS = \
+ $(rgen_h) \
$(platform_hdr) \
qpid/broker/AccumulatedAck.h \
qpid/broker/BrokerChannel.h \
@@ -329,7 +314,6 @@
qpid/client/FutureFactory.h \
qpid/client/ReceivedContent.h \
qpid/client/Response.h \
- qpid/client/Session.h \
qpid/client/SessionCore.h \
qpid/client/StateManager.h \
qpid/framing/AMQBody.h \
@@ -357,6 +341,7 @@
qpid/framing/SerializeHandler.h \
qpid/framing/SequenceNumber.h \
qpid/framing/SequenceNumberSet.h \
+ qpid/framing/StructHelper.h \
qpid/framing/Value.h \
qpid/framing/Visitor.h \
qpid/framing/Uuid.h \
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerAdapter.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerAdapter.cpp?rev=568174&r1=568173&r2=568174&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerAdapter.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerAdapter.cpp Tue Aug 21 08:51:41 2007
@@ -137,17 +137,17 @@
broker.getExchanges().destroy(name);
}
-void BrokerAdapter::ExchangeHandlerImpl::query(u_int16_t /*ticket*/, const string& name)
+ExchangeQueryResult BrokerAdapter::ExchangeHandlerImpl::query(u_int16_t /*ticket*/, const string& name)
{
try {
Exchange::shared_ptr exchange(broker.getExchanges().get(name));
- client.queryOk(exchange->getType(), exchange->isDurable(), false, exchange->getArgs());
+ return ExchangeQueryResult(exchange->getType(), exchange->isDurable(), false, exchange->getArgs());
} catch (const ChannelException& e) {
- client.queryOk("", false, true, FieldTable());
+ return ExchangeQueryResult("", false, true, FieldTable());
}
}
-void BrokerAdapter::BindingHandlerImpl::query(u_int16_t /*ticket*/,
+BindingQueryResult BrokerAdapter::BindingHandlerImpl::query(u_int16_t /*ticket*/,
const std::string& exchangeName,
const std::string& queueName,
const std::string& key,
@@ -164,24 +164,40 @@
}
if (!exchange) {
- client.queryOk(true, false, false, false, false);
+ return BindingQueryResult(true, false, false, false, false);
} else if (!queueName.empty() && !queue) {
- client.queryOk(false, true, false, false, false);
+ return BindingQueryResult(false, true, false, false, false);
} else if (exchange->isBound(queue, key.empty() ? 0 : &key, args.count() > 0 ? &args : &args)) {
- client.queryOk(false, false, false, false, false);
+ return BindingQueryResult(false, false, false, false, false);
} else {
//need to test each specified option individually
bool queueMatched = queueName.empty() || exchange->isBound(queue, 0, 0);
bool keyMatched = key.empty() || exchange->isBound(Queue::shared_ptr(), &key, 0);
bool argsMatched = args.count() == 0 || exchange->isBound(Queue::shared_ptr(), 0, &args);
- client.queryOk(false, false, !queueMatched, !keyMatched, !argsMatched);
+ return BindingQueryResult(false, false, !queueMatched, !keyMatched, !argsMatched);
}
}
+QueueQueryResult BrokerAdapter::QueueHandlerImpl::query(const string& name)
+{
+ Queue::shared_ptr queue = getQueue(name);
+ Exchange::shared_ptr alternateExchange = queue->getAlternateExchange();
+
+ return QueueQueryResult(queue->getName(),
+ alternateExchange ? alternateExchange->getName() : "",
+ queue->isDurable(),
+ queue->hasExclusiveOwner(),
+ queue->isAutoDelete(),
+ queue->getSettings(),
+ queue->getMessageCount(),
+ queue->getConsumerCount());
+
+}
+
void BrokerAdapter::QueueHandlerImpl::declare(uint16_t /*ticket*/, const string& name, const string& alternateExchange,
bool passive, bool durable, bool exclusive,
- bool autoDelete, bool nowait, const qpid::framing::FieldTable& arguments){
+ bool autoDelete, const qpid::framing::FieldTable& arguments){
Exchange::shared_ptr alternate;
if (!alternateExchange.empty()) {
alternate = broker.getExchanges().get(alternateExchange);
@@ -223,11 +239,6 @@
405,
format("Cannot grant exclusive access to queue '%s'")
% queue->getName());
- if (!nowait) {
- string queueName = queue->getName();
- client.declareOk(
- queueName, queue->getMessageCount(), queue->getConsumerCount());
- }
}
void BrokerAdapter::QueueHandlerImpl::bind(uint16_t /*ticket*/, const string& queueName,
@@ -269,17 +280,13 @@
}
-void BrokerAdapter::QueueHandlerImpl::purge(uint16_t /*ticket*/, const string& queueName, bool nowait){
-
- Queue::shared_ptr queue = getQueue(queueName);
- int count = queue->purge();
- if(!nowait) client.purgeOk( count);
+void BrokerAdapter::QueueHandlerImpl::purge(uint16_t /*ticket*/, const string& queue){
+ getQueue(queue)->purge();
}
void BrokerAdapter::QueueHandlerImpl::delete_(uint16_t /*ticket*/, const string& queue,
- bool ifUnused, bool ifEmpty, bool nowait){
+ bool ifUnused, bool ifEmpty){
ChannelException error(0, "");
- int count(0);
Queue::shared_ptr q = getQueue(queue);
if(ifEmpty && q->getMessageCount() > 0){
throw ChannelException(406, "Queue not empty.");
@@ -291,14 +298,10 @@
QueueVector::iterator i = find(connection.exclusiveQueues.begin(), connection.exclusiveQueues.end(), q);
if(i < connection.exclusiveQueues.end()) connection.exclusiveQueues.erase(i);
}
- count = q->getMessageCount();
q->destroy();
broker.getQueues().destroy(queue);
q->unbind(broker.getExchanges(), q);
}
-
- if(!nowait)
- client.deleteOk(count);
}
@@ -333,10 +336,8 @@
queue->requestDispatch();
}
-void BrokerAdapter::BasicHandlerImpl::cancel(const string& consumerTag, bool nowait){
+void BrokerAdapter::BasicHandlerImpl::cancel(const string& consumerTag){
channel.cancel(consumerTag);
-
- if(!nowait) client.cancelOk(consumerTag);
}
void BrokerAdapter::BasicHandlerImpl::publish(uint16_t /*ticket*/,
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerAdapter.h
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerAdapter.h?rev=568174&r1=568173&r2=568174&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerAdapter.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerAdapter.h Tue Aug 21 08:51:41 2007
@@ -121,7 +121,7 @@
const qpid::framing::FieldTable& arguments);
void delete_(uint16_t ticket,
const std::string& exchange, bool ifUnused);
- void query(u_int16_t ticket, const string& name);
+ framing::ExchangeQueryResult query(u_int16_t ticket, const string& name);
private:
void checkType(Exchange::shared_ptr exchange, const std::string& type);
void checkAlternate(Exchange::shared_ptr exchange, Exchange::shared_ptr alternate);
@@ -134,11 +134,11 @@
public:
BindingHandlerImpl(BrokerAdapter& parent) : HandlerImplType(parent) {}
- void query(u_int16_t ticket,
- const std::string& exchange,
- const std::string& queue,
- const std::string& routingKey,
- const framing::FieldTable& arguments);
+ framing::BindingQueryResult query(u_int16_t ticket,
+ const std::string& exchange,
+ const std::string& queue,
+ const std::string& routingKey,
+ const framing::FieldTable& arguments);
};
class QueueHandlerImpl :
@@ -151,7 +151,7 @@
void declare(uint16_t ticket, const std::string& queue,
const std::string& alternateExchange,
bool passive, bool durable, bool exclusive,
- bool autoDelete, bool nowait,
+ bool autoDelete,
const qpid::framing::FieldTable& arguments);
void bind(uint16_t ticket, const std::string& queue,
const std::string& exchange, const std::string& routingKey,
@@ -161,11 +161,10 @@
const std::string& exchange,
const std::string& routingKey,
const qpid::framing::FieldTable& arguments );
- void purge(uint16_t ticket, const std::string& queue,
- bool nowait);
+ framing::QueueQueryResult query(const string& queue);
+ void purge(uint16_t ticket, const std::string& queue);
void delete_(uint16_t ticket, const std::string& queue,
- bool ifUnused, bool ifEmpty,
- bool nowait);
+ bool ifUnused, bool ifEmpty);
};
class BasicHandlerImpl :
@@ -179,18 +178,15 @@
void qos(uint32_t prefetchSize,
uint16_t prefetchCount, bool global);
- void consume(
- uint16_t ticket, const std::string& queue,
- const std::string& consumerTag, bool noLocal, bool noAck,
- bool exclusive, bool nowait,
- const qpid::framing::FieldTable& fields);
- void cancel(const std::string& consumerTag,
- bool nowait);
+ void consume(uint16_t ticket, const std::string& queue,
+ const std::string& consumerTag,
+ bool noLocal, bool noAck, bool exclusive, bool nowait,
+ const qpid::framing::FieldTable& fields);
+ void cancel(const std::string& consumerTag);
void publish(uint16_t ticket,
const std::string& exchange, const std::string& routingKey,
bool rejectUnroutable, bool immediate);
- void get(uint16_t ticket, const std::string& queue,
- bool noAck);
+ void get(uint16_t ticket, const std::string& queue, bool noAck);
void ack(uint64_t deliveryTag, bool multiple);
void reject(uint64_t deliveryTag, bool requeue);
void recover(bool requeue);
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerChannel.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerChannel.cpp?rev=568174&r1=568173&r2=568174&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerChannel.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerChannel.cpp Tue Aug 21 08:51:41 2007
@@ -374,7 +374,7 @@
//just acked single element (move end past it)
++end;
}
-
+
for_each(start, end, boost::bind(&Channel::acknowledged, this, _1));
if (txBuffer.get()) {
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerQueue.h
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerQueue.h?rev=568174&r1=568173&r2=568174&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerQueue.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/broker/BrokerQueue.h Tue Aug 21 08:51:41 2007
@@ -145,8 +145,10 @@
inline const string& getName() const { return name; }
inline const bool isExclusiveOwner(const ConnectionToken* const o) const { return o == owner; }
inline bool hasExclusiveConsumer() const { return exclusive; }
+ inline bool hasExclusiveOwner() const { return owner != 0; }
inline bool isDurable() const { return store != 0; }
-
+ inline const framing::FieldTable& getSettings() const { return settings; }
+ inline bool isAutoDelete() const { return autodelete; }
bool canAutoDelete() const;
bool enqueue(TransactionContext* ctxt, Message::shared_ptr& msg);
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/broker/DtxHandlerImpl.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/broker/DtxHandlerImpl.cpp?rev=568174&r1=568173&r2=568174&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/DtxHandlerImpl.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/broker/DtxHandlerImpl.cpp Tue Aug 21 08:51:41 2007
@@ -22,19 +22,9 @@
#include "BrokerChannel.h"
using namespace qpid::broker;
-using qpid::framing::AMQP_ClientProxy;
-using qpid::framing::Buffer;
-using qpid::framing::FieldTable;
+using namespace qpid::framing;
using std::string;
-DtxHandlerImpl::DtxHandlerImpl(CoreRefs& parent) :
- CoreRefs(parent),
- dClient(AMQP_ClientProxy::DtxDemarcation::get(proxy)),
- cClient(AMQP_ClientProxy::DtxCoordination::get(proxy))
-
-{
-}
-
const int XA_RBROLLBACK(1);
const int XA_RBTIMEOUT(2);
const int XA_HEURHAZ(3);
@@ -44,6 +34,7 @@
const int XA_RDONLY(7);
const int XA_OK(8);
+DtxHandlerImpl::DtxHandlerImpl(CoreRefs& parent) : CoreRefs(parent) {}
// DtxDemarcationHandler:
@@ -53,10 +44,10 @@
channel.selectDtx();
}
-void DtxHandlerImpl::end(u_int16_t /*ticket*/,
- const string& xid,
- bool fail,
- bool suspend)
+DtxDemarcationEndResult DtxHandlerImpl::end(u_int16_t /*ticket*/,
+ const string& xid,
+ bool fail,
+ bool suspend)
{
try {
if (fail) {
@@ -64,7 +55,7 @@
if (suspend) {
throw ConnectionException(503, "End and suspend cannot both be set.");
} else {
- dClient.endOk(XA_RBROLLBACK);
+ return DtxDemarcationEndResult(XA_RBROLLBACK);
}
} else {
if (suspend) {
@@ -72,14 +63,14 @@
} else {
channel.endDtx(xid, false);
}
- dClient.endOk(XA_OK);
+ return DtxDemarcationEndResult(XA_OK);
}
} catch (const DtxTimeoutException& e) {
- dClient.endOk(XA_RBTIMEOUT);
+ return DtxDemarcationEndResult(XA_RBTIMEOUT);
}
}
-void DtxHandlerImpl::start(u_int16_t /*ticket*/,
+DtxDemarcationStartResult DtxHandlerImpl::start(u_int16_t /*ticket*/,
const string& xid,
bool join,
bool resume)
@@ -93,52 +84,52 @@
} else {
channel.startDtx(xid, broker.getDtxManager(), join);
}
- dClient.startOk(XA_OK);
+ return DtxDemarcationStartResult(XA_OK);
} catch (const DtxTimeoutException& e) {
- dClient.startOk(XA_RBTIMEOUT);
+ return DtxDemarcationStartResult(XA_RBTIMEOUT);
}
}
// DtxCoordinationHandler:
-void DtxHandlerImpl::prepare(u_int16_t /*ticket*/,
+DtxCoordinationPrepareResult DtxHandlerImpl::prepare(u_int16_t /*ticket*/,
const string& xid)
{
try {
bool ok = broker.getDtxManager().prepare(xid);
- cClient.prepareOk(ok ? XA_OK : XA_RBROLLBACK);
+ return DtxCoordinationPrepareResult(ok ? XA_OK : XA_RBROLLBACK);
} catch (const DtxTimeoutException& e) {
- cClient.prepareOk(XA_RBTIMEOUT);
+ return DtxCoordinationPrepareResult(XA_RBTIMEOUT);
}
}
-void DtxHandlerImpl::commit(u_int16_t /*ticket*/,
+DtxCoordinationCommitResult DtxHandlerImpl::commit(u_int16_t /*ticket*/,
const string& xid,
bool onePhase)
{
try {
bool ok = broker.getDtxManager().commit(xid, onePhase);
- cClient.commitOk(ok ? XA_OK : XA_RBROLLBACK);
+ return DtxCoordinationCommitResult(ok ? XA_OK : XA_RBROLLBACK);
} catch (const DtxTimeoutException& e) {
- cClient.commitOk(XA_RBTIMEOUT);
+ return DtxCoordinationCommitResult(XA_RBTIMEOUT);
}
}
-void DtxHandlerImpl::rollback(u_int16_t /*ticket*/,
+DtxCoordinationRollbackResult DtxHandlerImpl::rollback(u_int16_t /*ticket*/,
const string& xid )
{
try {
broker.getDtxManager().rollback(xid);
- cClient.rollbackOk(XA_OK);
+ return DtxCoordinationRollbackResult(XA_OK);
} catch (const DtxTimeoutException& e) {
- cClient.rollbackOk(XA_RBTIMEOUT);
+ return DtxCoordinationRollbackResult(XA_RBTIMEOUT);
}
}
-void DtxHandlerImpl::recover(u_int16_t /*ticket*/,
- bool /*startscan*/,
- bool /*endscan*/ )
+DtxCoordinationRecoverResult DtxHandlerImpl::recover(u_int16_t /*ticket*/,
+ bool /*startscan*/,
+ bool /*endscan*/ )
{
//TODO: what do startscan and endscan actually mean?
@@ -169,7 +160,7 @@
FieldTable response;
response.setString("xids", data);
- cClient.recoverOk(response);
+ return DtxCoordinationRecoverResult(response);
}
void DtxHandlerImpl::forget(u_int16_t /*ticket*/,
@@ -179,10 +170,10 @@
throw ConnectionException(503, boost::format("Forget is invalid. Branch with xid %1% not heuristically completed!") % xid);
}
-void DtxHandlerImpl::getTimeout(const string& xid)
+DtxCoordinationGetTimeoutResult DtxHandlerImpl::getTimeout(const string& xid)
{
uint32_t timeout = broker.getDtxManager().getTimeout(xid);
- cClient.getTimeoutOk(timeout);
+ return DtxCoordinationGetTimeoutResult(timeout);
}
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/broker/DtxHandlerImpl.h
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/broker/DtxHandlerImpl.h?rev=568174&r1=568173&r2=568174&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/DtxHandlerImpl.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/broker/DtxHandlerImpl.h Tue Aug 21 08:51:41 2007
@@ -31,34 +31,34 @@
public framing::AMQP_ServerOperations::DtxCoordinationHandler,
public framing::AMQP_ServerOperations::DtxDemarcationHandler
{
- framing::AMQP_ClientProxy::DtxDemarcation dClient;
- framing::AMQP_ClientProxy::DtxCoordination cClient;
public:
DtxHandlerImpl(CoreRefs& parent);
// DtxCoordinationHandler:
- void commit(u_int16_t ticket, const std::string& xid, bool onePhase);
+ framing::DtxCoordinationCommitResult commit(u_int16_t ticket, const std::string& xid, bool onePhase);
void forget(u_int16_t ticket, const std::string& xid);
- void getTimeout(const std::string& xid);
+ framing::DtxCoordinationGetTimeoutResult getTimeout(const std::string& xid);
- void prepare(u_int16_t ticket, const std::string& xid);
+ framing::DtxCoordinationPrepareResult prepare(u_int16_t ticket, const std::string& xid);
- void recover(u_int16_t ticket, bool startscan, bool endscan);
+ framing::DtxCoordinationRecoverResult recover(u_int16_t ticket, bool startscan, bool endscan);
- void rollback(u_int16_t ticket, const std::string& xid);
+ framing::DtxCoordinationRollbackResult rollback(u_int16_t ticket, const std::string& xid);
void setTimeout(u_int16_t ticket, const std::string& xid, u_int32_t timeout);
// DtxDemarcationHandler:
-
- void end(u_int16_t ticket, const std::string& xid, bool fail, bool suspend);
-
+
+ framing::DtxDemarcationEndResult end(u_int16_t ticket, const std::string& xid, bool fail, bool suspend);
+
void select();
+
+ framing::DtxDemarcationStartResult start(u_int16_t ticket, const std::string& xid, bool join, bool resume);
+
- void start(u_int16_t ticket, const std::string& xid, bool join, bool resume);
};
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/broker/MessageHandlerImpl.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/broker/MessageHandlerImpl.cpp?rev=568174&r1=568173&r2=568174&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/MessageHandlerImpl.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/broker/MessageHandlerImpl.cpp Tue Aug 21 08:51:41 2007
@@ -87,19 +87,21 @@
}
void
-MessageHandlerImpl::consume(uint16_t /*ticket*/,
+MessageHandlerImpl::subscribe(uint16_t /*ticket*/,
const string& queueName,
const string& destination,
bool noLocal,
- bool noAck,
+ u_int8_t confirmMode,
+ u_int8_t /*acquireMode*/,//TODO: implement acquire modes
bool exclusive,
const framing::FieldTable& filter )
{
Queue::shared_ptr queue = getQueue(queueName);
if(!destination.empty() && channel.exists(destination))
throw ConnectionException(530, "Consumer tags must be unique");
+
string tag = destination;
- channel.consume(MessageMessage::getToken(destination), tag, queue, noLocal, !noAck, exclusive, &filter);
+ channel.consume(MessageMessage::getToken(destination), tag, queue, noLocal, confirmMode == 1, exclusive, &filter);
// Dispatch messages as there is now a consumer.
queue->requestDispatch();
}
@@ -153,8 +155,9 @@
}
void
-MessageHandlerImpl::reject(uint16_t /*code*/, const string& /*text*/ )
+MessageHandlerImpl::reject(const SequenceNumberSet& /*transfers*/, uint16_t /*code*/, const string& /*text*/ )
{
+ //TODO: implement
}
void
@@ -210,5 +213,14 @@
channel.stop(destination);
}
+void MessageHandlerImpl::acquire(const SequenceNumberSet& /*transfers*/, u_int8_t /*mode*/)
+{
+ throw ConnectionException(540, "Not yet implemented");
+}
+
+void MessageHandlerImpl::release(const SequenceNumberSet& /*transfers*/)
+{
+ throw ConnectionException(540, "Not yet implemented");
+}
}} // namespace qpid::broker
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/broker/MessageHandlerImpl.h
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/broker/MessageHandlerImpl.h?rev=568174&r1=568173&r2=568174&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/MessageHandlerImpl.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/broker/MessageHandlerImpl.h Tue Aug 21 08:51:41 2007
@@ -49,14 +49,6 @@
void close(const std::string& reference );
- void consume(uint16_t ticket,
- const std::string& queue,
- const std::string& destination,
- bool noLocal,
- bool noAck,
- bool exclusive,
- const framing::FieldTable& filter );
-
void empty();
void get(uint16_t ticket,
@@ -76,8 +68,9 @@
void recover(bool requeue );
- void reject(uint16_t code,
- const std::string& text );
+ void reject(const framing::SequenceNumberSet& transfers,
+ uint16_t code,
+ const std::string& text );
void resume(const std::string& reference,
const std::string& identifier );
@@ -91,6 +84,19 @@
void flush(const std::string& destination);
void stop(const std::string& destination);
+
+ void acquire(const framing::SequenceNumberSet& transfers, u_int8_t mode);
+
+ void release(const framing::SequenceNumberSet& transfers);
+
+ void subscribe(u_int16_t ticket,
+ const string& queue,
+ const string& destination,
+ bool noLocal,
+ u_int8_t confirmMode,
+ u_int8_t acquireMode,
+ bool exclusive,
+ const framing::FieldTable& filter);
private:
ReferenceRegistry references;
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/broker/SemanticHandler.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/broker/SemanticHandler.cpp?rev=568174&r1=568173&r2=568174&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/SemanticHandler.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/broker/SemanticHandler.cpp Tue Aug 21 08:51:41 2007
@@ -22,8 +22,10 @@
#include "SemanticHandler.h"
#include "BrokerAdapter.h"
#include "qpid/framing/ChannelAdapter.h"
-#include "qpid/framing/ExecutionCompleteBody.h"
#include "qpid/framing/ChannelCloseOkBody.h"
+#include "qpid/framing/ExecutionCompleteBody.h"
+#include "qpid/framing/ExecutionResultBody.h"
+#include "qpid/framing/InvocationVisitor.h"
using namespace qpid::broker;
using namespace qpid::framing;
@@ -66,6 +68,11 @@
{
try {
if (!method->invoke(this)) {
+ //temporary hack until channel management is moved to its own handler:
+ if (method->amqpClassId() != ChannelOpenBody::CLASS_ID) {
+ ++(incoming.lwm);
+ }
+
//else do the usual:
handleL4(method);
//(if the frameset is complete) we can move the execution-mark
@@ -73,7 +80,9 @@
//temporary hack until channel management is moved to its own handler:
if (method->amqpClassId() != ChannelOpenBody::CLASS_ID) {
- ++(incoming.hwm);
+ //TODO: need to account for async store opreations
+ //when this command is a message publication
+ ++(incoming.hwm);
}
//note: need to be more sophisticated than this if we execute
@@ -85,7 +94,7 @@
}
}
-void SemanticHandler::complete(uint32_t cumulative, SequenceNumberSet range)
+void SemanticHandler::complete(uint32_t cumulative, const SequenceNumberSet& range)
{
//record:
SequenceNumber mark(cumulative);
@@ -98,7 +107,7 @@
if (range.size() % 2) { //must be even number
throw ConnectionException(530, "Received odd number of elements in ranged mark");
} else {
- for (SequenceNumberSet::iterator i = range.begin(); i != range.end(); i++) {
+ for (SequenceNumberSet::const_iterator i = range.begin(); i != range.end(); i++) {
channel.ackRange((uint64_t) i->getValue(), (uint64_t) (++i)->getValue());
}
}
@@ -113,6 +122,25 @@
ChannelAdapter::send(ExecutionCompleteBody(getVersion(), incoming.hwm.getValue(), SequenceNumberSet()));
}
}
+void SemanticHandler::sync()
+{
+ //for now, just treat as flush; will need to get more clever when we deal with async publication
+ flush();
+}
+
+void SemanticHandler::noop()
+{
+ //Do nothing...
+ //
+ //is this an L3 control? or is it an L4 command?
+ //if the former, of what use is it?
+ //if the latter it may contain a synch request... but its odd to have it in this class
+}
+
+void SemanticHandler::result(uint32_t /*command*/, const std::string& /*data*/)
+{
+ //never actually sent by client at present
+}
void SemanticHandler::handleL4(framing::AMQMethodBody* method)
{
@@ -124,7 +152,13 @@
throw ConnectionException(504, out.str());
}
} else {
- method->invoke(*adapter);
+ InvocationVisitor v(adapter.get());
+ method->accept(v);
+ if (!v.wasHandled()) {
+ throw ConnectionException(540, "Not implemented");
+ } else if (v.hasResult()) {
+ ChannelAdapter::send(ExecutionResultBody(getVersion(), incoming.lwm.getValue(), v.getResult()));
+ }
}
}catch(const ChannelException& e){
adapter->getProxy().getChannel().close(
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/broker/SemanticHandler.h
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/broker/SemanticHandler.h?rev=568174&r1=568173&r2=568174&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/broker/SemanticHandler.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/broker/SemanticHandler.h Tue Aug 21 08:51:41 2007
@@ -79,8 +79,11 @@
void handle(framing::AMQFrame& frame);
//execution class method handlers:
- void complete(uint32_t cumulativeExecutionMark, framing::SequenceNumberSet range);
+ void complete(uint32_t cumulativeExecutionMark, const framing::SequenceNumberSet& range);
void flush();
+ void noop();
+ void result(uint32_t command, const std::string& data);
+ void sync();
};
}}
Propchange: incubator/qpid/trunk/qpid/cpp/src/qpid/client/
------------------------------------------------------------------------------
--- svn:ignore (original)
+++ svn:ignore Tue Aug 21 08:51:41 2007
@@ -3,5 +3,3 @@
Makefile
Makefile.in
.dirstamp
-Session.cpp
-Session.h
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/client/ClientChannel.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/client/ClientChannel.cpp?rev=568174&r1=568173&r2=568174&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/client/ClientChannel.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/client/ClientChannel.cpp Tue Aug 21 08:51:41 2007
@@ -20,6 +20,7 @@
*/
#include "qpid/log/Statement.h"
#include <iostream>
+#include <sstream>
#include "ClientChannel.h"
#include "qpid/sys/Monitor.h"
#include "ClientMessage.h"
@@ -54,7 +55,8 @@
};
Channel::Channel(bool _transactional, u_int16_t _prefetch) :
- prefetch(_prefetch), transactional(_transactional), running(false)
+ prefetch(_prefetch), transactional(_transactional), running(false),
+ uniqueId(true)/*could eventually be the session id*/, nameCounter(0)
{
}
@@ -103,20 +105,22 @@
}
void Channel::declareQueue(Queue& queue, bool synch){
+ if (queue.getName().empty()) {
+ stringstream uniqueName;
+ uniqueName << uniqueId << "-queue-" << ++nameCounter;
+ queue.setName(uniqueName.str());
+ }
+
FieldTable args;
ScopedSync s(*session, synch);
- Response r = session->queueDeclare(0, queue.getName(), empty, false/*passive*/, queue.isDurable(),
- queue.isExclusive(), queue.isAutoDelete(), !synch, args);
+ session->queueDeclare(0, queue.getName(), empty, false/*passive*/, queue.isDurable(),
+ queue.isExclusive(), queue.isAutoDelete(), args);
- if(synch) {
- if(queue.getName().length() == 0)
- queue.setName(r.as<QueueDeclareOkBody>().getQueue());
- }
}
void Channel::deleteQueue(Queue& queue, bool ifunused, bool ifempty, bool synch){
ScopedSync s(*session, synch);
- session->queueDelete(0, queue.getName(), ifunused, ifempty, !synch);
+ session->queueDelete(0, queue.getName(), ifunused, ifempty);
}
void Channel::bind(const Exchange& exchange, const Queue& queue, const std::string& key, const FieldTable& args, bool synch){
@@ -168,7 +172,7 @@
consumers.erase(i);
}
ScopedSync s(*session, synch);
- session->basicCancel(tag, !synch);
+ session->basicCancel(tag);
}
bool Channel::get(Message& msg, const Queue& queue, AckMode ackMode) {
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/client/ClientChannel.h
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/client/ClientChannel.h?rev=568174&r1=568173&r2=568174&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/client/ClientChannel.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/client/ClientChannel.h Tue Aug 21 08:51:41 2007
@@ -24,11 +24,12 @@
#include <memory>
#include <boost/scoped_ptr.hpp>
#include "qpid/framing/amqp_framing.h"
+#include "qpid/framing/Uuid.h"
#include "ClientExchange.h"
#include "ClientMessage.h"
#include "ClientQueue.h"
#include "ConnectionImpl.h"
-#include "Session.h"
+#include "qpid/client/Session.h"
#include "qpid/Exception.h"
#include "qpid/sys/Mutex.h"
#include "qpid/sys/Runnable.h"
@@ -83,6 +84,8 @@
SessionCore::shared_ptr sessionCore;
framing::ChannelId channelId;
BlockingQueue<ReceivedContent::shared_ptr> gets;
+ framing::Uuid uniqueId;
+ uint32_t nameCounter;
void stop();
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/client/Connection.h
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/client/Connection.h?rev=568174&r1=568173&r2=568174&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/client/Connection.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/client/Connection.h Tue Aug 21 08:51:41 2007
@@ -26,7 +26,7 @@
#include "qpid/QpidError.h"
#include "ClientChannel.h"
#include "ConnectionImpl.h"
-#include "Session.h"
+#include "qpid/client/Session.h"
#include "qpid/framing/AMQP_HighestVersion.h"
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/client/ExecutionHandler.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/client/ExecutionHandler.cpp?rev=568174&r1=568173&r2=568174&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/client/ExecutionHandler.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/client/ExecutionHandler.cpp Tue Aug 21 08:51:41 2007
@@ -78,7 +78,7 @@
}
}
-void ExecutionHandler::complete(uint32_t cumulative, SequenceNumberSet range)
+void ExecutionHandler::complete(uint32_t cumulative, const SequenceNumberSet& range)
{
SequenceNumber mark(cumulative);
if (outgoing.lwm < mark) {
@@ -99,6 +99,26 @@
{
//send completion
incoming.lwm = incoming.hwm;
+}
+
+void ExecutionHandler::noop()
+{
+ //do nothing
+}
+
+void ExecutionHandler::result(uint32_t /*command*/, const std::string& /*data*/)
+{
+ //TODO: need to signal the result to the appropriate listener
+}
+
+void ExecutionHandler::sync()
+{
+ //TODO: implement (the application is in charge of completion of
+ //some commands, so need to track completion for them).
+
+ //This shouldn't ever need to be called by the server (in my
+ //opinion) as the server never needs to synchronise with the
+ //clients execution
}
void ExecutionHandler::sendFlush()
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/client/ExecutionHandler.h
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/client/ExecutionHandler.h?rev=568174&r1=568173&r2=568174&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/client/ExecutionHandler.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/client/ExecutionHandler.h Tue Aug 21 08:51:41 2007
@@ -45,8 +45,11 @@
framing::ProtocolVersion version;
uint64_t maxFrameSize;
- void complete(uint32_t mark, framing::SequenceNumberSet range);
+ void complete(uint32_t mark, const framing::SequenceNumberSet& range);
void flush();
+ void noop();
+ void result(uint32_t command, const std::string& data);
+ void sync();
public:
BlockingQueue<ReceivedContent::shared_ptr> received;
Propchange: incubator/qpid/trunk/qpid/cpp/src/qpid/framing/
------------------------------------------------------------------------------
--- svn:ignore (original)
+++ svn:ignore Tue Aug 21 08:51:41 2007
@@ -1,15 +1,7 @@
.deps
.libs
.dirstamp
-frame_body_lists.h
method_variants.h.gch.deps
method_variants.h.gch
method_variants.h
-AMQP_ClientProxy.cpp
-AMQP_ClientProxy.h
-AMQP_ServerProxy.cpp
-AMQP_ServerProxy.h
MethodHolderMaxSize.h
-MethodHolder_construct.cpp
-MethodBodyConstVisitor.h
-all_method_bodies.h
Added: incubator/qpid/trunk/qpid/cpp/src/qpid/framing/AMQP_HighestVersion.h
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/framing/AMQP_HighestVersion.h?rev=568174&view=auto
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/framing/AMQP_HighestVersion.h (added)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/framing/AMQP_HighestVersion.h Tue Aug 21 08:51:41 2007
@@ -0,0 +1,40 @@
+/*
+ *
+ * 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.
+ *
+ */
+
+/*
+ * This file used to be auto-generated by Qpid Gentools v.0.1
+ * its here temporarily until we get a full solution to multi-version support
+ */
+#ifndef qpid_framing_highestProtocolVersion__
+#define qpid_framing_highestProtocolVersion__
+
+#include "qpid/framing/ProtocolVersion.h"
+
+
+namespace qpid {
+namespace framing {
+
+static ProtocolVersion highestProtocolVersion(0, 10);
+
+} /* namespace framing */
+} /* namespace qpid */
+
+#endif
Propchange: incubator/qpid/trunk/qpid/cpp/src/qpid/framing/AMQP_HighestVersion.h
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: incubator/qpid/trunk/qpid/cpp/src/qpid/framing/AMQP_HighestVersion.h
------------------------------------------------------------------------------
svn:keywords = Rev Date
Added: incubator/qpid/trunk/qpid/cpp/src/qpid/framing/StructHelper.h
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/framing/StructHelper.h?rev=568174&view=auto
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/framing/StructHelper.h (added)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/framing/StructHelper.h Tue Aug 21 08:51:41 2007
@@ -0,0 +1,56 @@
+/*
+ *
+ * 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.
+ *
+ */
+#ifndef _StructHelper_
+#define _StructHelper_
+
+#include "qpid/Exception.h"
+
+namespace qpid {
+namespace framing {
+
+class StructHelper
+{
+public:
+
+ template <class T> void encode(const T t, std::string& data) {
+ uint32_t size = t.size() + 2/*type*/;
+ Buffer buffer(size);
+ buffer.putShort(T::TYPE);
+ t.encode(buffer);
+ buffer.flip();
+ buffer.getRawData(data, size);
+ }
+
+ template <class T> void decode(T t, std::string& data) {
+ Buffer buffer(data.length());
+ buffer.putRawData(data);
+ buffer.flip();
+ uint16_t type = buffer.getShort();
+ if (type == T::TYPE) {
+ t.decode(buffer);
+ } else {
+ throw Exception("Type code does not match");
+ }
+ }
+};
+
+}}
+#endif
Propchange: incubator/qpid/trunk/qpid/cpp/src/qpid/framing/StructHelper.h
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: incubator/qpid/trunk/qpid/cpp/src/qpid/framing/StructHelper.h
------------------------------------------------------------------------------
svn:keywords = Rev Date
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/framing/amqp_types_full.h
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/framing/amqp_types_full.h?rev=568174&r1=568173&r2=568174&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/framing/amqp_types_full.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/framing/amqp_types_full.h Tue Aug 21 08:51:41 2007
@@ -32,5 +32,6 @@
#include "amqp_types.h"
#include "FramingContent.h"
#include "FieldTable.h"
+#include "SequenceNumberSet.h"
#endif /*!_framing_amqp_types_decl_h*/
Modified: incubator/qpid/trunk/qpid/cpp/src/tests/Cluster_child.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/tests/Cluster_child.cpp?rev=568174&r1=568173&r2=568174&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/tests/Cluster_child.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/tests/Cluster_child.cpp Tue Aug 21 08:51:41 2007
@@ -32,7 +32,7 @@
static const ProtocolVersion VER;
-/** Chlid part of Cluster::clusterTwo test */
+/** Child part of Cluster::clusterTwo test */
void clusterTwo() {
TestCluster cluster("clusterTwo", "amqp:child:2");
AMQFrame frame;
Propchange: incubator/qpid/trunk/qpid/python/mllib/
------------------------------------------------------------------------------
--- svn:ignore (added)
+++ svn:ignore Tue Aug 21 08:51:41 2007
@@ -0,0 +1 @@
+*.class
Modified: incubator/qpid/trunk/qpid/python/qpid/peer.py
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/python/qpid/peer.py?rev=568174&r1=568173&r2=568174&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/python/qpid/peer.py (original)
+++ incubator/qpid/trunk/qpid/python/qpid/peer.py Tue Aug 21 08:51:41 2007
@@ -208,6 +208,8 @@
self.responses.close()
self.completion.close()
self.incoming_completion.reset()
+ for f in self.futures.values():
+ f.put_response(self, reason)
def write(self, frame, content = None):
if self.closed:
@@ -324,7 +326,10 @@
raise ValueError(resp)
elif frame.method.result:
if self.synchronous:
- return future.get_response(timeout=10)
+ fr = future.get_response(timeout=10)
+ if self.closed:
+ raise Closed(self.reason)
+ return fr
else:
return future
elif self.synchronous and not frame.method.response \
@@ -373,7 +378,10 @@
def get_response(self, timeout=None):
self.completed.wait(timeout)
- return self.response
+ if self.completed.isSet():
+ return self.response
+ else:
+ return None
def is_complete(self):
return self.completed.isSet()
Modified: incubator/qpid/trunk/qpid/python/qpid/testlib.py
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/python/qpid/testlib.py?rev=568174&r1=568173&r2=568174&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/python/qpid/testlib.py (original)
+++ incubator/qpid/trunk/qpid/python/qpid/testlib.py Tue Aug 21 08:51:41 2007
@@ -228,7 +228,7 @@
def queue_declare(self, channel=None, *args, **keys):
channel = channel or self.channel
reply = channel.queue_declare(*args, **keys)
- self.queues.append((channel, reply.queue))
+ self.queues.append((channel, keys["queue"]))
return reply
def exchange_declare(self, channel=None, ticket=0, exchange='',
@@ -254,7 +254,7 @@
if not "uniqueTag" in dir(self): self.uniqueTag = 1
else: self.uniqueTag += 1
consumer_tag = "tag" + str(self.uniqueTag)
- self.channel.message_consume(queue=queueName, destination=consumer_tag, no_ack=True)
+ self.channel.message_subscribe(queue=queueName, destination=consumer_tag)
return self.client.queue(consumer_tag)
def assertEmpty(self, queue):
Modified: incubator/qpid/trunk/qpid/python/tests_0-10/alternate-exchange.py
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/python/tests_0-10/alternate-exchange.py?rev=568174&r1=568173&r2=568174&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/python/tests_0-10/alternate-exchange.py (original)
+++ incubator/qpid/trunk/qpid/python/tests_0-10/alternate-exchange.py Tue Aug 21 08:51:41 2007
@@ -39,13 +39,13 @@
#declare, bind (to the alternate exchange) and consume from a queue for 'returned' messages
channel.queue_declare(queue="returns", exclusive=True)
channel.queue_bind(queue="returns", exchange="secondary")
- channel.message_consume(destination="a", queue="returns")
+ channel.message_subscribe(destination="a", queue="returns")
returned = self.client.queue("a")
#declare, bind (to the primary exchange) and consume from a queue for 'processed' messages
channel.queue_declare(queue="processed", exclusive=True)
channel.queue_bind(queue="processed", exchange="primary", routing_key="my-key")
- channel.message_consume(destination="b", queue="processed")
+ channel.message_subscribe(destination="b", queue="processed")
processed = self.client.queue("b")
#publish to the primary exchange
@@ -73,7 +73,7 @@
channel.exchange_declare(exchange="dlq", type="fanout")
channel.queue_declare(queue="deleted", exclusive=True)
channel.queue_bind(exchange="dlq", queue="deleted")
- channel.message_consume(destination="dlq", queue="deleted")
+ channel.message_subscribe(destination="dlq", queue="deleted")
dlq = self.client.queue("dlq")
#create a queue using the dlq as its alternate exchange:
@@ -103,7 +103,7 @@
channel.exchange_declare(exchange="dlq", type="fanout")
channel.queue_declare(queue="immediate", exclusive=True)
channel.queue_bind(exchange="dlq", queue="immediate")
- channel.message_consume(destination="dlq", queue="immediate")
+ channel.message_subscribe(destination="dlq", queue="immediate")
dlq = self.client.queue("dlq")
#create a queue using the dlq as its alternate exchange:
Modified: incubator/qpid/trunk/qpid/python/tests_0-10/broker.py
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/python/tests_0-10/broker.py?rev=568174&r1=568173&r2=568174&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/python/tests_0-10/broker.py (original)
+++ incubator/qpid/trunk/qpid/python/tests_0-10/broker.py Tue Aug 21 08:51:41 2007
@@ -35,7 +35,7 @@
# No ack consumer
ctag = "tag1"
- ch.message_consume(queue = "myqueue", destination = ctag, no_ack = True)
+ ch.message_subscribe(queue = "myqueue", destination = ctag, confirm_mode = 0)
body = "test no-ack"
ch.message_transfer(routing_key = "myqueue", body = body)
msg = self.client.queue(ctag).get(timeout = 5)
@@ -44,7 +44,7 @@
# Acknowledging consumer
self.queue_declare(ch, queue = "otherqueue")
ctag = "tag2"
- ch.message_consume(queue = "otherqueue", destination = ctag, no_ack = False)
+ ch.message_subscribe(queue = "otherqueue", destination = ctag, confirm_mode = 1)
body = "test ack"
ch.message_transfer(routing_key = "otherqueue", body = body)
msg = self.client.queue(ctag).get(timeout = 5)
@@ -60,7 +60,7 @@
self.queue_declare(channel, queue="test-queue")
channel.queue_bind(queue="test-queue", exchange="test-exchange", routing_key="key")
consumer_tag = "tag1"
- channel.message_consume(queue="test-queue", destination=consumer_tag, no_ack=True)
+ channel.message_subscribe(queue="test-queue", destination=consumer_tag, confirm_mode = 0)
queue = self.client.queue(consumer_tag)
body = "Immediate Delivery"
@@ -84,7 +84,7 @@
channel.message_transfer(destination="test-exchange", routing_key="key", body=body)
consumer_tag = "tag1"
- channel.message_consume(queue="test-queue", destination=consumer_tag, no_ack=True)
+ channel.message_subscribe(queue="test-queue", destination=consumer_tag, confirm_mode = 0)
queue = self.client.queue(consumer_tag)
msg = queue.get(timeout=5)
self.assert_(msg.body == body)
@@ -111,7 +111,7 @@
def test_channel_flow(self):
channel = self.channel
channel.queue_declare(queue="flow_test_queue", exclusive=True)
- channel.message_consume(destination="my-tag", queue="flow_test_queue")
+ channel.message_subscribe(destination="my-tag", queue="flow_test_queue")
incoming = self.client.queue("my-tag")
channel.channel_flow(active=False)
Modified: incubator/qpid/trunk/qpid/python/tests_0-10/dtx.py
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/python/tests_0-10/dtx.py?rev=568174&r1=568173&r2=568174&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/python/tests_0-10/dtx.py (original)
+++ incubator/qpid/trunk/qpid/python/tests_0-10/dtx.py Tue Aug 21 08:51:41 2007
@@ -366,7 +366,7 @@
#check the second message is available, but not the first
self.assertMessageCount(1, "tx-queue")
- channel.message_consume(queue="tx-queue", destination="results", no_ack=False)
+ channel.message_subscribe(queue="tx-queue", destination="results", confirm_mode=1)
msg = self.client.queue("results").get(timeout=1)
self.assertEqual("two", msg.message_id)
channel.message_cancel(destination="results")
@@ -597,9 +597,9 @@
channel.message_transfer(routing_key=dest, message_id=msg.message_id, body=msg.body)
def assertMessageCount(self, expected, queue):
- self.assertEqual(expected, self.channel.queue_declare(queue=queue, passive=True).message_count)
+ self.assertEqual(expected, self.channel.queue_query(queue=queue).message_count)
def assertMessageId(self, expected, queue):
- self.channel.message_consume(queue=queue, destination="results", no_ack=True)
+ self.channel.message_subscribe(queue=queue, destination="results")
self.assertEqual(expected, self.client.queue("results").get(timeout=1).message_id)
self.channel.message_cancel(destination="results")
Modified: incubator/qpid/trunk/qpid/python/tests_0-10/example.py
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/python/tests_0-10/example.py?rev=568174&r1=568173&r2=568174&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/python/tests_0-10/example.py (original)
+++ incubator/qpid/trunk/qpid/python/tests_0-10/example.py Tue Aug 21 08:51:41 2007
@@ -68,7 +68,7 @@
# has fields corresponding to the reply method fields, plus a content
# field that is filled if the reply includes content. In this case the
# interesting field is the consumer_tag.
- channel.message_consume(queue="test-queue", destination="consumer_tag")
+ channel.message_subscribe(queue="test-queue", destination="consumer_tag")
# We can use the Client.queue(...) method to access the queue
# corresponding to our consumer_tag.