You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by ac...@apache.org on 2008/05/26 20:10:14 UTC
svn commit: r660258 - in /incubator/qpid/trunk/qpid/cpp:
examples/examples/direct/ examples/examples/fanout/
examples/examples/pub-sub/ examples/examples/request-response/
examples/examples/xml-exchange/ rubygen/framing.0-10/ src/ src/qpid/
src/qpid/cl...
Author: aconway
Date: Mon May 26 11:10:05 2008
New Revision: 660258
URL: http://svn.apache.org/viewvc?rev=660258&view=rev
Log:
Changes to Session API:
- Session is synchronous, no futures.
- AsyncSession is async, returns futures.
- Conversion functions sync(s) async(s) return a sync/async view of session s.
- Connection::newSession - takes name, no timeout
- SessionBase::getId - returns SessionId not UUID.
Added:
incubator/qpid/trunk/qpid/cpp/src/qpid/client/AsyncSession.h (with props)
incubator/qpid/trunk/qpid/cpp/src/qpid/client/SessionBase_0_10.cpp
- copied, changed from r660191, incubator/qpid/trunk/qpid/cpp/src/qpid/client/SessionBase.cpp
incubator/qpid/trunk/qpid/cpp/src/qpid/client/SessionBase_0_10.h
- copied, changed from r660191, incubator/qpid/trunk/qpid/cpp/src/qpid/client/SessionBase.h
incubator/qpid/trunk/qpid/cpp/src/qpid/client/SessionBase_0_10Access.h (with props)
Removed:
incubator/qpid/trunk/qpid/cpp/src/qpid/client/SessionBase.cpp
incubator/qpid/trunk/qpid/cpp/src/qpid/client/SessionBase.h
Modified:
incubator/qpid/trunk/qpid/cpp/examples/examples/direct/declare_queues.cpp
incubator/qpid/trunk/qpid/cpp/examples/examples/direct/direct_producer.cpp
incubator/qpid/trunk/qpid/cpp/examples/examples/direct/listener.cpp
incubator/qpid/trunk/qpid/cpp/examples/examples/fanout/fanout_producer.cpp
incubator/qpid/trunk/qpid/cpp/examples/examples/fanout/listener.cpp
incubator/qpid/trunk/qpid/cpp/examples/examples/pub-sub/topic_listener.cpp
incubator/qpid/trunk/qpid/cpp/examples/examples/pub-sub/topic_publisher.cpp
incubator/qpid/trunk/qpid/cpp/examples/examples/request-response/client.cpp
incubator/qpid/trunk/qpid/cpp/examples/examples/request-response/server.cpp
incubator/qpid/trunk/qpid/cpp/examples/examples/xml-exchange/declare_queues.cpp
incubator/qpid/trunk/qpid/cpp/examples/examples/xml-exchange/listener.cpp
incubator/qpid/trunk/qpid/cpp/examples/examples/xml-exchange/xml_producer.cpp
incubator/qpid/trunk/qpid/cpp/rubygen/framing.0-10/Session.rb
incubator/qpid/trunk/qpid/cpp/src/Makefile.am
incubator/qpid/trunk/qpid/cpp/src/qpid/SessionId.h
incubator/qpid/trunk/qpid/cpp/src/qpid/client/AckPolicy.h
incubator/qpid/trunk/qpid/cpp/src/qpid/client/Connection.cpp
incubator/qpid/trunk/qpid/cpp/src/qpid/client/Connection.h
incubator/qpid/trunk/qpid/cpp/src/qpid/client/Dispatcher.cpp
incubator/qpid/trunk/qpid/cpp/src/qpid/client/Dispatcher.h
incubator/qpid/trunk/qpid/cpp/src/qpid/client/Session.h
incubator/qpid/trunk/qpid/cpp/src/qpid/client/SessionImpl.cpp
incubator/qpid/trunk/qpid/cpp/src/qpid/client/SessionImpl.h
incubator/qpid/trunk/qpid/cpp/src/qpid/client/SubscriptionManager.cpp
incubator/qpid/trunk/qpid/cpp/src/qpid/client/SubscriptionManager.h
incubator/qpid/trunk/qpid/cpp/src/tests/BrokerFixture.h
incubator/qpid/trunk/qpid/cpp/src/tests/ClientSessionTest.cpp
incubator/qpid/trunk/qpid/cpp/src/tests/XmlClientSessionTest.cpp
incubator/qpid/trunk/qpid/cpp/src/tests/client_test.cpp
incubator/qpid/trunk/qpid/cpp/src/tests/consume.cpp
incubator/qpid/trunk/qpid/cpp/src/tests/exception_test.cpp
incubator/qpid/trunk/qpid/cpp/src/tests/latencytest.cpp
incubator/qpid/trunk/qpid/cpp/src/tests/perftest.cpp
incubator/qpid/trunk/qpid/cpp/src/tests/publish.cpp
incubator/qpid/trunk/qpid/cpp/src/tests/topic_listener.cpp
incubator/qpid/trunk/qpid/cpp/src/tests/topic_publisher.cpp
incubator/qpid/trunk/qpid/cpp/src/tests/topictest
incubator/qpid/trunk/qpid/cpp/src/tests/txtest.cpp
Modified: incubator/qpid/trunk/qpid/cpp/examples/examples/direct/declare_queues.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/examples/examples/direct/declare_queues.cpp?rev=660258&r1=660257&r2=660258&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/examples/examples/direct/declare_queues.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/examples/examples/direct/declare_queues.cpp Mon May 26 11:10:05 2008
@@ -60,7 +60,7 @@
try {
connection.open(host, port);
- Session session = connection.newSession(ASYNC);
+ Session session = connection.newSession();
//--------- Main body of program --------------------------------------------
Modified: incubator/qpid/trunk/qpid/cpp/examples/examples/direct/direct_producer.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/examples/examples/direct/direct_producer.cpp?rev=660258&r1=660257&r2=660258&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/examples/examples/direct/direct_producer.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/examples/examples/direct/direct_producer.cpp Mon May 26 11:10:05 2008
@@ -67,7 +67,7 @@
Message message;
try {
connection.open(host, port);
- Session session = connection.newSession(ASYNC);
+ Session session = connection.newSession();
//--------- Main body of program --------------------------------------------
@@ -85,7 +85,9 @@
message_data << "Message " << i;
message.setData(message_data.str());
- session.messageTransfer(arg::content=message, arg::destination="amq.direct");
+ // Asynchronous transfer sends messages as quickly as
+ // possible without waiting for confirmation.
+ async(session).messageTransfer(arg::content=message, arg::destination="amq.direct");
}
// And send a final message to indicate termination.
Modified: incubator/qpid/trunk/qpid/cpp/examples/examples/direct/listener.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/examples/examples/direct/listener.cpp?rev=660258&r1=660257&r2=660258&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/examples/examples/direct/listener.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/examples/examples/direct/listener.cpp Mon May 26 11:10:05 2008
@@ -65,7 +65,7 @@
Message msg;
try {
connection.open(host, port);
- Session session = connection.newSession(ASYNC);
+ Session session = connection.newSession();
//--------- Main body of program --------------------------------------------
Modified: incubator/qpid/trunk/qpid/cpp/examples/examples/fanout/fanout_producer.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/examples/examples/fanout/fanout_producer.cpp?rev=660258&r1=660257&r2=660258&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/examples/examples/fanout/fanout_producer.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/examples/examples/fanout/fanout_producer.cpp Mon May 26 11:10:05 2008
@@ -67,7 +67,7 @@
Message message;
try {
connection.open(host, port);
- Session session = connection.newSession(ASYNC);
+ Session session = connection.newSession();
//--------- Main body of program --------------------------------------------
@@ -83,7 +83,9 @@
message_data << "Message " << i;
message.setData(message_data.str());
- session.messageTransfer(arg::content=message, arg::destination="amq.fanout");
+ // Asynchronous transfer sends messages as quickly as
+ // possible without waiting for confirmation.
+ async(session).messageTransfer(arg::content=message, arg::destination="amq.fanout");
}
// And send a final message to indicate termination.
Modified: incubator/qpid/trunk/qpid/cpp/examples/examples/fanout/listener.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/examples/examples/fanout/listener.cpp?rev=660258&r1=660257&r2=660258&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/examples/examples/fanout/listener.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/examples/examples/fanout/listener.cpp Mon May 26 11:10:05 2008
@@ -65,7 +65,7 @@
Message msg;
try {
connection.open(host, port);
- Session session = connection.newSession(ASYNC);
+ Session session = connection.newSession();
//--------- Main body of program --------------------------------------------
@@ -90,9 +90,6 @@
Listener listener(subscriptions);
subscriptions.subscribe(listener, myQueue);
- // Wait for the broker to indicate that our queues have been created.
- session.sync();
-
// Deliver messages until the subscription is cancelled
// by Listener::received()
std::cout << "Listening" << std::endl;
Modified: incubator/qpid/trunk/qpid/cpp/examples/examples/pub-sub/topic_listener.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/examples/examples/pub-sub/topic_listener.cpp?rev=660258&r1=660257&r2=660258&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/examples/examples/pub-sub/topic_listener.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/examples/examples/pub-sub/topic_listener.cpp Mon May 26 11:10:05 2008
@@ -91,7 +91,7 @@
* the queue name parameter with the Session ID.
*/
- queue += session.getId().str();
+ queue += session.getId().getName();
std::cout << "Declaring queue: " << queue << std::endl;
/* Declare an exclusive queue on the broker
@@ -138,7 +138,7 @@
Connection connection;
try {
connection.open(host, port);
- Session session = connection.newSession(ASYNC);
+ Session session = connection.newSession();
//--------- Main body of program --------------------------------------------
@@ -153,9 +153,6 @@
listener.prepareQueue("news", "#.news");
listener.prepareQueue("weather", "#.weather");
- // Wait for the broker to indicate that our queues have been created.
- session.sync();
-
std::cout << "Listening for messages ..." << std::endl;
// Give up control and receive messages
Modified: incubator/qpid/trunk/qpid/cpp/examples/examples/pub-sub/topic_publisher.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/examples/examples/pub-sub/topic_publisher.cpp?rev=660258&r1=660257&r2=660258&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/examples/examples/pub-sub/topic_publisher.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/examples/examples/pub-sub/topic_publisher.cpp Mon May 26 11:10:05 2008
@@ -72,7 +72,9 @@
message_data << "Message " << i;
message.setData(message_data.str());
- session.messageTransfer(arg::content=message, arg::destination="amq.topic");
+ // Asynchronous transfer sends messages as quickly as
+ // possible without waiting for confirmation.
+ async(session).messageTransfer(arg::content=message, arg::destination="amq.topic");
}
}
@@ -101,7 +103,7 @@
Message message;
try {
connection.open(host, port);
- Session session = connection.newSession(ASYNC);
+ Session session = connection.newSession();
//--------- Main body of program --------------------------------------------
Modified: incubator/qpid/trunk/qpid/cpp/examples/examples/request-response/client.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/examples/examples/request-response/client.cpp?rev=660258&r1=660257&r2=660258&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/examples/examples/request-response/client.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/examples/examples/request-response/client.cpp Mon May 26 11:10:05 2008
@@ -114,7 +114,7 @@
Message request;
try {
connection.open(host, port);
- Session session = connection.newSession(ASYNC);
+ Session session = connection.newSession();
//--------- Main body of program --------------------------------------------
@@ -157,7 +157,9 @@
for (int i=0; i<4; i++) {
request.setData(s[i]);
- session.messageTransfer(arg::content=request, arg::destination="amq.direct");
+ // Asynchronous transfer sends messages as quickly as
+ // possible without waiting for confirmation.
+ async(session).messageTransfer(arg::content=request, arg::destination="amq.direct");
std::cout << "Request: " << s[i] << std::endl;
}
Modified: incubator/qpid/trunk/qpid/cpp/examples/examples/request-response/server.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/examples/examples/request-response/server.cpp?rev=660258&r1=660257&r2=660258&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/examples/examples/request-response/server.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/examples/examples/request-response/server.cpp Mon May 26 11:10:05 2008
@@ -120,7 +120,10 @@
// Send it back to the user
response.getDeliveryProperties().setRoutingKey(routingKey);
- session.messageTransfer(arg::content=response, arg::destination="amq.direct");
+
+ // Asynchronous transfer sends messages as quickly as
+ // possible without waiting for confirmation.
+ async(session).messageTransfer(arg::content=response, arg::destination="amq.direct");
}
@@ -131,7 +134,7 @@
Message message;
try {
connection.open(host, port);
- Session session = connection.newSession(ASYNC);
+ Session session = connection.newSession();
//--------- Main body of program --------------------------------------------
Modified: incubator/qpid/trunk/qpid/cpp/examples/examples/xml-exchange/declare_queues.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/examples/examples/xml-exchange/declare_queues.cpp?rev=660258&r1=660257&r2=660258&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/examples/examples/xml-exchange/declare_queues.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/examples/examples/xml-exchange/declare_queues.cpp Mon May 26 11:10:05 2008
@@ -39,7 +39,7 @@
try {
connection.open(host, port);
- Session session = connection.newSession(ASYNC);
+ Session session = connection.newSession();
//--------- Main body of program --------------------------------------------
Modified: incubator/qpid/trunk/qpid/cpp/examples/examples/xml-exchange/listener.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/examples/examples/xml-exchange/listener.cpp?rev=660258&r1=660257&r2=660258&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/examples/examples/xml-exchange/listener.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/examples/examples/xml-exchange/listener.cpp Mon May 26 11:10:05 2008
@@ -65,7 +65,7 @@
Message msg;
try {
connection.open(host, port);
- Session session = connection.newSession(ASYNC);
+ Session session = connection.newSession();
//--------- Main body of program --------------------------------------------
Modified: incubator/qpid/trunk/qpid/cpp/examples/examples/xml-exchange/xml_producer.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/examples/examples/xml-exchange/xml_producer.cpp?rev=660258&r1=660257&r2=660258&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/examples/examples/xml-exchange/xml_producer.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/examples/examples/xml-exchange/xml_producer.cpp Mon May 26 11:10:05 2008
@@ -44,7 +44,7 @@
Message message;
try {
connection.open(host, port);
- Session session = connection.newSession(ASYNC);
+ Session session = connection.newSession();
//--------- Main body of program --------------------------------------------
@@ -66,7 +66,9 @@
std::cout << "Message data: " << message_data.str() << std::endl;
message.setData(message_data.str());
- session.messageTransfer(arg::content=message, arg::destination="xml");
+ // Asynchronous transfer sends messages as quickly as
+ // possible without waiting for confirmation.
+ async(session).messageTransfer(arg::content=message, arg::destination="xml");
}
// And send a final message to indicate termination.
Modified: incubator/qpid/trunk/qpid/cpp/rubygen/framing.0-10/Session.rb
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/rubygen/framing.0-10/Session.rb?rev=660258&r1=660257&r2=660258&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/rubygen/framing.0-10/Session.rb (original)
+++ incubator/qpid/trunk/qpid/cpp/rubygen/framing.0-10/Session.rb Mon May 26 11:10:05 2008
@@ -25,6 +25,33 @@
end
end
+# Sync vs. async APIs
+module SyncAsync
+ def sync_prefix() @async ? "Async" : "" end
+ def sync_adjective() @async ? "asynchronous" : "synchronous" end
+ def sync_convert() @async ? "async" : "sync" end
+
+
+ def decl_ctor_opeq()
+ genl
+ genl "#{@classname}();"
+ genl "#{@classname}(const #{@version_base}& other);"
+ genl "#{@classname}& operator=(const #{@version_base}& other);"
+ end
+
+ def defn_ctor_opeq(inline="")
+ genl
+ genl "#{inline} #{@classname}::#{@classname}() {}"
+ scope("#{inline} #{@classname}::#{@classname}(const #{@version_base}& other) {") {
+ genl "*this = other;"
+ }
+ scope("#{inline} #{@classname}& #{@classname}::operator=(const #{@version_base}& other) {") {
+ genl "impl = static_cast<const #{@classname}&>(other).impl;"
+ genl "return *this;"
+ }
+ end
+end
+
class ContentField # For extra content parameters
def cppname() "content" end
def signature() "const MethodContent& content" end
@@ -49,55 +76,52 @@
fields_c.map { |f| "arg::keyword_tags::"+f.cppname }.join(',') +
">"
end
- def return_type()
- return "TypedResult<qpid::framing::#{result.cpptype.ret_by_val}>" if (result)
- return "Response" if (not responses().empty?)
- return "Completion"
+
+ def return_type(async)
+ if (async)
+ return "TypedResult<qpid::framing::#{result.cpptype.ret_by_val}>" if (result)
+ return "Completion"
+ else
+ return "qpid::framing::#{result.cpptype.ret_by_val}" if (result)
+ return "void"
+ end
end
+
def session_function() "#{parent.name.lcaps}#{name.caps}"; end
end
class SessionNoKeywordGen < CppGen
-
- def initialize(outdir, amqp)
+ include SyncAsync
+
+ def initialize(outdir, amqp, async)
super(outdir, amqp)
+ @async=async
@chassis="server"
@namespace,@classname,@file=
- parse_classname "qpid::client::no_keyword::Session_#{@amqp.version.bars}"
+ parse_classname "qpid::client::no_keyword::#{sync_prefix}Session_#{@amqp.version.bars}"
+ @version_base="SessionBase_#{@amqp.major}_#{@amqp.minor}"
end
def generate()
h_file(@file) {
- include "qpid/client/SessionBase.h"
-
- namespace("qpid::client") {
- genl "using std::string;"
- genl "using framing::Content;"
- genl "using framing::FieldTable;"
- genl "using framing::MethodContent;"
- genl "using framing::SequenceNumber;"
- genl "using framing::SequenceSet;"
- genl "using framing::Uuid;"
- #the following are nasty... would be better to dynamically
- #include such statements based on params required
- genl "using framing::Xid;"
- genl
- namespace("no_keyword") {
- doxygen_comment {
- genl "AMQP #{@amqp.version} session API."
- genl @amqp.class_("session").doc
+ include "qpid/client/#{@version_base}.h"
+ namespace(@namespace) {
+ doxygen_comment {
+ genl "AMQP #{@amqp.version} #{sync_adjective} session API."
+ genl @amqp.class_("session").doc
+ # FIXME aconway 2008-05-23: additional doc on sync/async use.
+ }
+ cpp_class(@classname, "public #{@version_base}") {
+ public
+ decl_ctor_opeq()
+ session_methods.each { |m|
+ genl
+ doxygen(m)
+ args=m.sig_c_default.join(", ")
+ genl "#{m.return_type(@async)} #{m.session_function}(#{args});"
}
- cpp_class(@classname, "public SessionBase") {
- public
- genl "Session_#{@amqp.version.bars}() {}"
- genl "Session_#{@amqp.version.bars}(shared_ptr<SessionImpl> core) : SessionBase(core) {}"
- session_methods.each { |m|
- genl
- doxygen(m)
- args=m.sig_c_default.join(", ")
- genl "#{m.return_type} #{m.session_function}(#{args});"
- }
- }}}}
+ }
+ }}
cpp_file(@file) {
include @classname
@@ -108,32 +132,47 @@
genl
sig=m.signature_c.join(", ")
func="#{@classname}::#{m.session_function}"
- scope("#{m.return_type} #{func}(#{sig}) {") {
- args=(["ProtocolVersion()"]+m.param_names).join(", ")
- body="#{m.body_name}(#{args})"
- sendargs=body
+ scope("#{m.return_type(@async)} #{func}(#{sig}) {") {
+ args=(["ProtocolVersion(#{@amqp.major},#{@amqp.minor})"]+m.param_names).join(", ")
+ genl "#{m.body_name} body(#{args});";
+ genl "body.setSync(#{@async ? 'false':'true'});"
+ sendargs="body"
sendargs << ", content" if m.content
- genl "return #{m.return_type}(impl->send(#{sendargs}), impl);"
- }}}}
+ async_retval="#{m.return_type(true)}(impl->send(#{sendargs}), impl)"
+ if @async then
+ genl "return #{async_retval};"
+ else
+ if m.result
+ genl "return #{async_retval}.get();"
+ else
+ genl "#{async_retval}.wait();"
+ end
+ end
+ }}
+ defn_ctor_opeq()
+ }}
end
end
class SessionGen < CppGen
+ include SyncAsync
- def initialize(outdir, amqp)
+ def initialize(outdir, amqp, async)
super(outdir, amqp)
+ @async=async
@chassis="server"
- session="Session_#{@amqp.version.bars}"
+ session="#{sync_prefix}Session_#{@amqp.version.bars}"
@base="no_keyword::#{session}"
@fqclass=FqClass.new "qpid::client::#{session}"
@classname=@fqclass.name
@fqbase=FqClass.new("qpid::client::#{@base}")
+ @version_base="SessionBase_#{@amqp.major}_#{@amqp.minor}"
end
- def gen_keyword_decl(m, prefix)
+ def gen_keyword_decl(m)
return if m.fields_c.empty? # Inherited function will do.
- scope("BOOST_PARAMETER_MEMFUN(#{m.return_type}, #{m.session_function}, 0, #{m.fields_c.size}, #{m.argpack_name}) {") {
- scope("return #{prefix}#{m.session_function}(",");") {
+ scope("BOOST_PARAMETER_MEMFUN(#{m.return_type(@async)}, #{m.session_function}, 0, #{m.fields_c.size}, #{m.argpack_name}) {") {
+ scope("return #{@base}::#{m.session_function}(",");") {
gen m.fields_c.map { |f| f.unpack() }.join(",\n")
}
}
@@ -143,20 +182,21 @@
def generate()
keyword_methods=session_methods.reject { |m| m.fields_c.empty? }
max_arity = keyword_methods.map{ |m| m.fields_c.size }.max
+
+ h_file("qpid/client/arg.h") {
+ # Generate keyword tag declarations.
+ genl "#define BOOST_PARAMETER_MAX_ARITY #{max_arity}"
+ include "<boost/parameter.hpp>"
+ namespace("qpid::client::arg") {
+ keyword_methods.map{ |m| m.param_names_c }.flatten.uniq.each { |k|
+ genl "BOOST_PARAMETER_KEYWORD(keyword_tags, #{k})"
+ }}
+ }
h_file(@fqclass.file) {
include @fqbase.file
- genl
- genl "#define BOOST_PARAMETER_MAX_ARITY #{max_arity}"
- include "<boost/parameter.hpp>"
- genl
+ include "qpid/client/arg"
namespace("qpid::client") {
- # Generate keyword tag declarations.
- namespace("arg") {
- keyword_methods.map{ |m| m.param_names_c }.flatten.uniq.each { |k|
- genl "BOOST_PARAMETER_KEYWORD(keyword_tags, #{k})"
- }}
- genl
# Doxygen comment.
doxygen_comment {
genl "AMQP #{@amqp.version} session API with keyword arguments."
@@ -183,17 +223,23 @@
}
# Session class.
cpp_class(@classname,"public #{@base}") {
+ public
+ decl_ctor_opeq()
private
- genl "#{@classname}(shared_ptr<SessionImpl> core) : #{ @base}(core) {}"
keyword_methods.each { |m| typedef m.argpack_type, m.argpack_name }
genl "friend class Connection;"
public
- genl "#{@classname}() {}"
- keyword_methods.each { |m| gen_keyword_decl(m,@base+"::") }
- }}}
+ keyword_methods.each { |m| gen_keyword_decl(m) }
+ }
+ genl "/** Conversion to #{@classname} from another session type */"
+ genl "inline #{@classname} #{sync_convert}(const #{@version_base}& other) { return #{@clasname}(other); }"
+ defn_ctor_opeq("inline")
+ }}
end
end
-SessionNoKeywordGen.new(ARGV[0], $amqp).generate()
-SessionGen.new(ARGV[0], $amqp).generate()
+SessionNoKeywordGen.new(ARGV[0], $amqp, true).generate()
+SessionNoKeywordGen.new(ARGV[0], $amqp, false).generate()
+SessionGen.new(ARGV[0], $amqp, true).generate()
+SessionGen.new(ARGV[0], $amqp, false).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=660258&r1=660257&r2=660258&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/Makefile.am (original)
+++ incubator/qpid/trunk/qpid/cpp/src/Makefile.am Mon May 26 11:10:05 2008
@@ -324,7 +324,9 @@
qpid/client/MessageListener.cpp \
qpid/client/Queue.cpp \
qpid/client/Results.cpp \
- qpid/client/SessionBase.cpp \
+ qpid/client/SessionBase_0_10.cpp \
+ qpid/client/SessionBase_0_10.h \
+ qpid/client/SessionBase_0_10Access.h \
qpid/client/SessionImpl.cpp \
qpid/client/StateManager.cpp \
qpid/client/SubscriptionManager.cpp
@@ -446,8 +448,9 @@
qpid/client/MessageQueue.h \
qpid/client/Queue.h \
qpid/client/Results.h \
- qpid/client/SessionBase.h \
+ qpid/client/SessionBase_0_10.h \
qpid/client/Session.h \
+ qpid/client/AsyncSession.h \
qpid/client/SessionImpl.h \
qpid/client/StateManager.h \
qpid/client/SubscriptionManager.h \
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/SessionId.h
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/SessionId.h?rev=660258&r1=660257&r2=660258&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/SessionId.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/SessionId.h Mon May 26 11:10:05 2008
@@ -27,7 +27,17 @@
namespace qpid {
-/** Identifier for a session */
+/** Identifier for a session.
+ * There are two parts to a session identifier:
+ *
+ * getUserId() returns the authentication principal associated with
+ * the session's connection.
+ *
+ * getName() returns the session name.
+ *
+ * The name must be unique among sessions with the same authentication
+ * principal.
+ */
class SessionId : boost::totally_ordered1<SessionId> {
std::string userId;
std::string name;
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/client/AckPolicy.h
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/client/AckPolicy.h?rev=660258&r1=660257&r2=660258&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/client/AckPolicy.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/client/AckPolicy.h Mon May 26 11:10:05 2008
@@ -22,6 +22,7 @@
*/
#include "qpid/framing/SequenceSet.h"
+#include "qpid/client/AsyncSession.h"
namespace qpid {
namespace client {
@@ -44,7 +45,7 @@
*/
AckPolicy(size_t n=1) : interval(n), count(n) {}
- void ack(const Message& msg, Session& session) {
+ void ack(const Message& msg, AsyncSession session) {
accepted.add(msg.getId());
if (!interval) return;
if (--count==0) {
@@ -57,7 +58,7 @@
}
}
- void ackOutstanding(Session& session) {
+ void ackOutstanding(AsyncSession session) {
if (!accepted.empty()) {
session.messageAccept(accepted);
accepted.clear();
Added: incubator/qpid/trunk/qpid/cpp/src/qpid/client/AsyncSession.h
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/client/AsyncSession.h?rev=660258&view=auto
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/client/AsyncSession.h (added)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/client/AsyncSession.h Mon May 26 11:10:05 2008
@@ -0,0 +1,38 @@
+#ifndef QPID_CLIENT_ASYNCSESSION_H
+#define QPID_CLIENT_ASYNCSESSION_H
+
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+#include "qpid/client/AsyncSession_0_10.h"
+
+namespace qpid {
+namespace client {
+
+/**
+ * AsyncSession is an alias for Session_0_10
+ *
+ * \ingroup clientapi
+ */
+typedef AsyncSession_0_10 AsyncSession;
+
+}} // namespace qpid::client
+
+#endif /*!QPID_CLIENT_ASYNCSESSION_H*/
Propchange: incubator/qpid/trunk/qpid/cpp/src/qpid/client/AsyncSession.h
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: incubator/qpid/trunk/qpid/cpp/src/qpid/client/AsyncSession.h
------------------------------------------------------------------------------
svn:keywords = Rev Date
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/client/Connection.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/client/Connection.cpp?rev=660258&r1=660257&r2=660258&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/client/Connection.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/client/Connection.cpp Mon May 26 11:10:05 2008
@@ -22,6 +22,7 @@
#include "ConnectionSettings.h"
#include "Message.h"
#include "SessionImpl.h"
+#include "SessionBase_0_10Access.h"
#include "qpid/log/Logger.h"
#include "qpid/log/Options.h"
#include "qpid/log/Statement.h"
@@ -72,18 +73,16 @@
max_frame_size = impl->getNegotiatedSettings().maxFrameSize;
}
-Session Connection::newSession(SynchronousMode sync,
- uint32_t detachedLifetime)
-{
+Session Connection::newSession(const std::string& name) {
if (!impl)
throw Exception(QPID_MSG("Connection has not yet been opened"));
-
- shared_ptr<SessionImpl> core(
- new SessionImpl(impl, ++channelIdCounter, max_frame_size));
- core->setSync(sync);
- impl->addSession(core);
- core->open(detachedLifetime);
- return Session(core);
+ shared_ptr<SessionImpl> simpl(
+ new SessionImpl(name, impl, ++channelIdCounter, max_frame_size));
+ impl->addSession(simpl);
+ simpl->open(0);
+ Session s;
+ SessionBase_0_10Access(s).set(simpl);
+ return s;
}
void Connection::resume(Session& session) {
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=660258&r1=660257&r2=660258&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/client/Connection.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/client/Connection.h Mon May 26 11:10:05 2008
@@ -26,7 +26,6 @@
#include "ConnectionImpl.h"
#include "qpid/client/Session.h"
#include "qpid/framing/AMQP_HighestVersion.h"
-#include "qpid/framing/Uuid.h"
namespace qpid {
@@ -111,14 +110,11 @@
* multiple streams of work to be multiplexed over the same
* connection.
*
- *@param detachedLifetime: A session may be detached from its
- * channel, either by calling Session::suspend() or because of a
- * network failure. The session state is preserved for
- * detachedLifetime seconds to allow a call to resume(). After
- * that the broker may discard the session state. Default is 0,
- * meaning the session cannot be resumed.
+ *@param name: A name to identify the session. @see qpid::SessionId
+ * If the name is empty (the default) then a unique name will be
+ * chosen using a Universally-unique identifier (UUID) algorithm.
*/
- Session newSession(SynchronousMode sync, uint32_t detachedLifetime=0);
+ Session newSession(const std::string& name=std::string());
/**
* Resume a suspended session. A session may be resumed
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/client/Dispatcher.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/client/Dispatcher.cpp?rev=660258&r1=660257&r2=660258&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/client/Dispatcher.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/client/Dispatcher.cpp Mon May 26 11:10:05 2008
@@ -37,7 +37,8 @@
namespace qpid {
namespace client {
-Subscriber::Subscriber(Session& s, MessageListener* l, AckPolicy a) : session(s), listener(l), autoAck(a) {}
+Subscriber::Subscriber(const Session& s, MessageListener* l, AckPolicy a)
+ : session(s), listener(l), autoAck(a) {}
void Subscriber::received(Message& msg)
{
@@ -47,7 +48,7 @@
}
}
-Dispatcher::Dispatcher(Session& s, const std::string& q)
+Dispatcher::Dispatcher(const Session& s, const std::string& q)
: session(s), running(false), autoStop(true)
{
queue = q.empty() ?
@@ -88,7 +89,7 @@
}
}
}
- session.sync(); // Make sure all our acks are received before returning.
+ sync(session).sync(); // Make sure all our acks are received before returning.
}
catch (const ClosedException&) {} //ignore it and return
catch (const std::exception& e) {
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/client/Dispatcher.h
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/client/Dispatcher.h?rev=660258&r1=660257&r2=660258&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/client/Dispatcher.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/client/Dispatcher.h Mon May 26 11:10:05 2008
@@ -37,13 +37,13 @@
class Subscriber : public MessageListener
{
- Session& session;
+ AsyncSession session;
MessageListener* const listener;
AckPolicy autoAck;
public:
typedef boost::shared_ptr<Subscriber> shared_ptr;
- Subscriber(Session& session, MessageListener* listener, AckPolicy);
+ Subscriber(const Session& session, MessageListener* listener, AckPolicy);
void received(Message& msg);
};
@@ -55,7 +55,7 @@
typedef std::map<std::string, Subscriber::shared_ptr> Listeners;
sys::Mutex lock;
sys::Thread worker;
- Session& session;
+ Session session;
Demux::QueuePtr queue;
bool running;
bool autoStop;
@@ -67,7 +67,7 @@
bool isStopped();
public:
- Dispatcher(Session& session, const std::string& queue = "");
+ Dispatcher(const Session& session, const std::string& queue = "");
void start();
void run();
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/client/Session.h
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/client/Session.h?rev=660258&r1=660257&r2=660258&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/client/Session.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/client/Session.h Mon May 26 11:10:05 2008
@@ -33,6 +33,7 @@
*/
typedef Session_0_10 Session;
+
}} // namespace qpid::client
#endif /*!QPID_CLIENT_SESSION_H*/
Copied: incubator/qpid/trunk/qpid/cpp/src/qpid/client/SessionBase_0_10.cpp (from r660191, incubator/qpid/trunk/qpid/cpp/src/qpid/client/SessionBase.cpp)
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/client/SessionBase_0_10.cpp?p2=incubator/qpid/trunk/qpid/cpp/src/qpid/client/SessionBase_0_10.cpp&p1=incubator/qpid/trunk/qpid/cpp/src/qpid/client/SessionBase.cpp&r1=660191&r2=660258&rev=660258&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/client/SessionBase.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/client/SessionBase_0_10.cpp Mon May 26 11:10:05 2008
@@ -18,67 +18,47 @@
* under the License.
*
*/
-#include "SessionBase.h"
+#include "SessionBase_0_10.h"
#include "qpid/framing/all_method_bodies.h"
namespace qpid {
namespace client {
using namespace framing;
-SessionBase::SessionBase() {}
-SessionBase::~SessionBase() {}
-SessionBase::SessionBase(shared_ptr<SessionImpl> core) : impl(core) {}
-void SessionBase::suspend() { impl->suspend(); }
-void SessionBase::close() { impl->close(); }
-
-void SessionBase::setSynchronous(bool isSync) { impl->setSync(isSync); }
-void SessionBase::setSynchronous(SynchronousMode m) { impl->setSync(m); }
-bool SessionBase::isSynchronous() const { return impl->isSync(); }
-SynchronousMode SessionBase::getSynchronous() const {
- return SynchronousMode(impl->isSync());
-}
+SessionBase_0_10::SessionBase_0_10() {}
+SessionBase_0_10::~SessionBase_0_10() {}
+
+void SessionBase_0_10::close() { impl->close(); }
-Execution& SessionBase::getExecution()
+Execution& SessionBase_0_10::getExecution()
{
return *impl;
}
-void SessionBase::flush()
+void SessionBase_0_10::flush()
{
impl->sendFlush();
}
-// FIXME aconway 2008-04-24: do we need to provide a non-synchronous version
-// of sync() or bool paramter to allow setting a sync point for a later wait?
-void SessionBase::sync()
+void SessionBase_0_10::sync()
{
ExecutionSyncBody b;
b.setSync(true);
impl->send(b).wait(*impl);
}
-void SessionBase::markCompleted(const framing::SequenceNumber& id, bool cumulative, bool notifyPeer)
+void SessionBase_0_10::markCompleted(const framing::SequenceNumber& id, bool cumulative, bool notifyPeer)
{
impl->markCompleted(id, cumulative, notifyPeer);
}
-void SessionBase::sendCompletion()
+void SessionBase_0_10::sendCompletion()
{
impl->sendCompletion();
}
-Uuid SessionBase::getId() const { return impl->getId(); }
-framing::FrameSet::shared_ptr SessionBase::get() { return impl->get(); }
-
-SessionBase::ScopedSync::ScopedSync(SessionBase& s) : session(s), change(!s.isSynchronous())
-{
- if (change) session.setSynchronous(true);
-}
-
-SessionBase::ScopedSync::~ScopedSync()
-{
- if (change) session.setSynchronous(false);
-}
+SessionId SessionBase_0_10::getId() const { return impl->getId(); }
+framing::FrameSet::shared_ptr SessionBase_0_10::get() { return impl->get(); }
}} // namespace qpid::client
Copied: incubator/qpid/trunk/qpid/cpp/src/qpid/client/SessionBase_0_10.h (from r660191, incubator/qpid/trunk/qpid/cpp/src/qpid/client/SessionBase.h)
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/client/SessionBase_0_10.h?p2=incubator/qpid/trunk/qpid/cpp/src/qpid/client/SessionBase_0_10.h&p1=incubator/qpid/trunk/qpid/cpp/src/qpid/client/SessionBase.h&r1=660191&r2=660258&rev=660258&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/client/SessionBase.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/client/SessionBase_0_10.h Mon May 26 11:10:05 2008
@@ -22,7 +22,7 @@
*
*/
-#include "qpid/framing/Uuid.h"
+#include "qpid/SessionId.h"
#include "qpid/framing/amqp_structs.h"
#include "qpid/framing/ProtocolVersion.h"
#include "qpid/framing/MethodContent.h"
@@ -42,105 +42,61 @@
using framing::Content;
using framing::FieldTable;
using framing::MethodContent;
+using framing::SequenceNumber;
+using framing::SequenceSet;
using framing::SequenceNumberSet;
-using framing::Uuid;
+using qpid::SessionId;
+using framing::Xid;
enum CreditUnit { MESSAGE=0, BYTE=1 };
-/** \defgroup clientapi Synchronous mode of a session.
- *
- * SYNC means that Session functions do not return until the remote
- * broker has confirmed that the command was executed.
- *
- * ASYNC means that the client sends commands asynchronously, Session
- * functions return immediately.
- *
- * ASYNC mode gives better performance for high-volume traffic, but
- * requires some additional caution.
- *
- * Session functions return immediately. If the command causes an
- * exception on the broker, the exception will be thrown on a
- * <em>later</em> function call.
- *
- * If you need to notify some extenal agent that some actions have
- * been taken (e.g. binding queues to exchanges), you must call
- * Session::sync() first to ensure that all the commands are complete.
- *
- * You can freely switch between modes by calling Session::setSynchronous().
- *
- * @see Session::sync(), Session::setSynchronous()
- */
-enum SynchronousMode { SYNC=true, ASYNC=false };
-
-
/**
- * Basic session operations that are not derived from AMQP XML methods.
+ * Base class for handles to an AMQP session.
+ *
+ * Subclasses provide the AMQP commands for a given
+ * version of the protocol.
*/
-class SessionBase
-{
+class SessionBase_0_10 {
public:
- /**
- * Instances of this class turn synchronous mode on for the
- * duration of their scope (and revert back to async if required
- * afterwards).
- */
- class ScopedSync
- {
- SessionBase& session;
- const bool change;
- public:
- ScopedSync(SessionBase& s);
- ~ScopedSync();
- };
-
+
+ typedef framing::TransferContent DefaultContent;
- SessionBase();
- ~SessionBase();
+ SessionBase_0_10();
+ ~SessionBase_0_10();
/** Get the next message frame-set from the session. */
framing::FrameSet::shared_ptr get();
/** Get the session ID */
- Uuid getId() const;
+ SessionId getId() const;
- /**
- * In synchronous mode, wait for the broker's response before
- * returning. This gives lower throughput than asynchronous
- * mode.
- *
- * In asynchronous mode commands are sent without waiting
- * for a response (you can use the returned Completion object
- * to wait for completion).
- *
- * @see SynchronousMode
- */
- void setSynchronous(SynchronousMode mode);
- void setSynchronous(bool set);
- bool isSynchronous() const;
- SynchronousMode getSynchronous() const;
-
- /**
- * Suspend the session, which can be resumed on a different connection.
- * @see Connection::resume()
+ /** Close the session.
+ * A session is automatically closed when all handles to it are destroyed.
*/
- void suspend();
-
- /** Close the session */
void close();
- Execution& getExecution();
+ /** Synchronize the session: sync() waits until all commands
+ * issued on this session have been completed. It is equivalent to
+ * calling Session::executionSync()
+ *
+ * Note sync() is always synchronous, even on an AsyncSession object
+ * because that's almost always what you want. You can call
+ * AsyncSession::executionSync() directly in the unusual event
+ * that you want to do an asynchronous sync.
+ */
void sync();
+
+ /** Set the timeout for this session. */
+ uint32_t timeout(uint32_t seconds);
+
+ Execution& getExecution();
void flush();
void markCompleted(const framing::SequenceNumber& id, bool cumulative, bool notifyPeer);
void sendCompletion();
-
- typedef framing::TransferContent DefaultContent;
protected:
- shared_ptr<SessionImpl> impl;
- framing::ProtocolVersion version;
- friend class Connection;
- SessionBase(shared_ptr<SessionImpl>);
+ boost::shared_ptr<SessionImpl> impl;
+ friend class SessionBase_0_10Access;
};
}} // namespace qpid::client
Added: incubator/qpid/trunk/qpid/cpp/src/qpid/client/SessionBase_0_10Access.h
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/client/SessionBase_0_10Access.h?rev=660258&view=auto
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/client/SessionBase_0_10Access.h (added)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/client/SessionBase_0_10Access.h Mon May 26 11:10:05 2008
@@ -0,0 +1,42 @@
+#ifndef QPID_CLIENT_SESSIONBASEACCESS_H
+#define QPID_CLIENT_SESSIONBASEACCESS_H
+
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+#include "qpid/client/SessionBase_0_10.h"
+
+/**@file @internal Internal use only */
+
+namespace qpid {
+namespace client {
+
+class SessionBase_0_10Access {
+ public:
+ SessionBase_0_10Access(SessionBase_0_10& sb_) : sb(sb_) {}
+ void set(const boost::shared_ptr<SessionImpl>& si) { sb.impl = si; }
+ boost::shared_ptr<SessionImpl> get() { return sb.impl; }
+ private:
+ SessionBase_0_10& sb;
+};
+}} // namespace qpid::client
+
+#endif /*!QPID_CLIENT_SESSIONBASEACCESS_H*/
Propchange: incubator/qpid/trunk/qpid/cpp/src/qpid/client/SessionBase_0_10Access.h
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: incubator/qpid/trunk/qpid/cpp/src/qpid/client/SessionBase_0_10Access.h
------------------------------------------------------------------------------
svn:keywords = Rev Date
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/client/SessionImpl.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/client/SessionImpl.cpp?rev=660258&r1=660257&r2=660258&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/client/SessionImpl.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/client/SessionImpl.cpp Mon May 26 11:10:05 2008
@@ -48,18 +48,16 @@
typedef sys::ScopedLock<sys::Semaphore> Acquire;
-SessionImpl::SessionImpl(shared_ptr<ConnectionImpl> conn,
+SessionImpl::SessionImpl(const std::string& name,
+ shared_ptr<ConnectionImpl> conn,
uint16_t ch, uint64_t _maxFrameSize)
: error(OK),
code(NORMAL),
text(EMPTY),
state(INACTIVE),
- syncMode(false),
detachedLifetime(0),
maxFrameSize(_maxFrameSize),
- id(true),//generate unique uuid for each session
- name(id.str()),
- //TODO: may want to allow application defined names instead
+ id(conn->getNegotiatedSettings().username, name.empty() ? Uuid(true).str() : name),
connection(conn),
ioHandler(*this),
channel(ch),
@@ -81,15 +79,6 @@
connection->erase(channel);
}
-void SessionImpl::setSync(bool s) // user thread
-{
- syncMode = s; //syncMode is volatile
-}
-
-bool SessionImpl::isSync() // user thread
-{
- return syncMode; //syncMode is volatile
-}
FrameSet::shared_ptr SessionImpl::get() // user thread
{
@@ -97,7 +86,7 @@
return demux.getDefault()->pop();
}
-const Uuid SessionImpl::getId() const //user thread
+const SessionId SessionImpl::getId() const //user thread
{
return id; //id is immutable
}
@@ -107,7 +96,7 @@
Lock l(state);
if (state == INACTIVE) {
setState(ATTACHING);
- proxy.attach(name, false);
+ proxy.attach(id.getName(), false);
waitFor(ATTACHED);
//TODO: timeout will not be set locally until get response to
//confirm, should we wait for that?
@@ -144,7 +133,7 @@
void SessionImpl::detach() //call with lock held
{
if (state == ATTACHED) {
- proxy.detach(name);
+ proxy.detach(id.getName());
setState(DETACHING);
}
}
@@ -285,15 +274,11 @@
{
Acquire a(sendLock);
SequenceNumber id = nextOut++;
- bool sync;
{
Lock l(state);
checkOpen();
incompleteOut.add(id);
- sync = syncMode;
}
-
- if (sync) command.getMethod()->setSync(true);
Future f(id);
if (command.getMethod()->resultExpected()) {
Lock l(state);
@@ -308,9 +293,6 @@
if (content) {
sendContent(*content);
}
- if (sync) {
- waitForCompletion(id);
- }
return f;
}
void SessionImpl::sendContent(const MethodContent& content)
@@ -441,27 +423,27 @@
void SessionImpl::attached(const std::string& _name)
{
Lock l(state);
- if (name != _name) throw InternalErrorException("Incorrect session name");
+ if (id.getName() != _name) throw InternalErrorException("Incorrect session name");
setState(ATTACHED);
}
void SessionImpl::detach(const std::string& _name)
{
Lock l(state);
- if (name != _name) throw InternalErrorException("Incorrect session name");
+ if (id.getName() != _name) throw InternalErrorException("Incorrect session name");
setState(DETACHED);
- QPID_LOG(info, "Session detached by peer: " << name);
+ QPID_LOG(info, "Session detached by peer: " << id);
}
void SessionImpl::detached(const std::string& _name, uint8_t _code)
{
Lock l(state);
- if (name != _name) throw InternalErrorException("Incorrect session name");
+ if (id.getName() != _name) throw InternalErrorException("Incorrect session name");
setState(DETACHED);
if (_code) {
//TODO: make sure this works with execution.exception - don't
//want to overwrite the code from that
- QPID_LOG(error, "Session detached by peer: " << name << " " << code);
+ QPID_LOG(error, "Session detached by peer: " << id << " " << code);
error = SESSION_DETACH;
code = _code;
text = "Session detached by peer";
@@ -561,8 +543,6 @@
throw NotImplementedException("gap not yet supported");
}
-
-
void SessionImpl::sync() {}
void SessionImpl::result(const framing::SequenceNumber& commandId, const std::string& value)
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/client/SessionImpl.h
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/client/SessionImpl.h?rev=660258&r1=660257&r2=660258&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/client/SessionImpl.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/client/SessionImpl.h Mon May 26 11:10:05 2008
@@ -26,6 +26,7 @@
#include "Execution.h"
#include "Results.h"
+#include "qpid/SessionId.h"
#include "qpid/shared_ptr.h"
#include "qpid/framing/FrameHandler.h"
#include "qpid/framing/ChannelHandler.h"
@@ -59,14 +60,14 @@
private framing::AMQP_ClientOperations::ExecutionHandler
{
public:
- SessionImpl(shared_ptr<ConnectionImpl>, uint16_t channel, uint64_t maxFrameSize);
+ SessionImpl(const std::string& name, shared_ptr<ConnectionImpl>, uint16_t channel, uint64_t maxFrameSize);
~SessionImpl();
//NOTE: Public functions called in user thread.
framing::FrameSet::shared_ptr get();
- const framing::Uuid getId() const;
+ const SessionId getId() const;
uint16_t getChannel() const;
void setChannel(uint16_t channel);
@@ -76,8 +77,6 @@
void resume(shared_ptr<ConnectionImpl>);
void suspend();
- void setSync(bool s);
- bool isSync();
void assertOpen() const;
Future send(const framing::AMQBody& command);
@@ -131,7 +130,8 @@
Future sendCommand(const framing::AMQBody&, const framing::MethodContent* = 0);
void sendContent(const framing::MethodContent&);
void waitForCompletionImpl(const framing::SequenceNumber& id);
-
+ void requestTimeout(uint32_t timeout);
+
void sendCompletionImpl();
// Note: Following methods are called by network thread in
@@ -140,7 +140,6 @@
void attached(const std::string& name);
void detach(const std::string& name);
void detached(const std::string& name, uint8_t detachCode);
- void requestTimeout(uint32_t timeout);
void timeout(uint32_t timeout);
void commandPoint(const framing::SequenceNumber& commandId, uint64_t commandOffset);
void expected(const framing::SequenceSet& commands, const framing::Array& fragments);
@@ -167,11 +166,9 @@
std::string text; // Error text
mutable StateMonitor state;
mutable sys::Semaphore sendLock;
- volatile bool syncMode;
uint32_t detachedLifetime;
const uint64_t maxFrameSize;
- const framing::Uuid id;
- const std::string name;
+ const SessionId id;
shared_ptr<ConnectionImpl> connection;
framing::FrameHandler::MemFunRef<SessionImpl, &SessionImpl::proxyOut> ioHandler;
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/client/SubscriptionManager.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/client/SubscriptionManager.cpp?rev=660258&r1=660257&r2=660258&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/client/SubscriptionManager.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/client/SubscriptionManager.cpp Mon May 26 11:10:05 2008
@@ -32,23 +32,23 @@
namespace qpid {
namespace client {
-SubscriptionManager::SubscriptionManager(Session& s)
+SubscriptionManager::SubscriptionManager(const Session& s)
: dispatcher(s), session(s),
messages(UNLIMITED), bytes(UNLIMITED), window(true),
acceptMode(0), acquireMode(0),
autoStop(true)
{}
-Completion SubscriptionManager::subscribeInternal(
+void SubscriptionManager::subscribeInternal(
const std::string& q, const std::string& dest)
{
- Completion c = session.messageSubscribe(arg::queue=q, arg::destination=dest,
- arg::acceptMode=acceptMode, arg::acquireMode=acquireMode);
- setFlowControl(dest, messages, bytes, window);
- return c;
+ async(session).messageSubscribe( // setFlowControl will sync.
+ arg::queue=q, arg::destination=dest,
+ arg::acceptMode=acceptMode, arg::acquireMode=acquireMode);
+ setFlowControl(dest, messages, bytes, window);
}
-Completion SubscriptionManager::subscribe(
+void SubscriptionManager::subscribe(
MessageListener& listener, const std::string& q, const std::string& d)
{
std::string dest=d.empty() ? q:d;
@@ -56,7 +56,7 @@
return subscribeInternal(q, dest);
}
-Completion SubscriptionManager::subscribe(
+void SubscriptionManager::subscribe(
LocalQueue& lq, const std::string& q, const std::string& d)
{
std::string dest=d.empty() ? q:d;
@@ -68,9 +68,9 @@
void SubscriptionManager::setFlowControl(
const std::string& dest, uint32_t messages, uint32_t bytes, bool window)
{
- session.messageSetFlowMode(dest, window);
- session.messageFlow(dest, 0, messages);
- session.messageFlow(dest, 1, bytes);
+ async(session).messageSetFlowMode(dest, window);
+ async(session).messageFlow(dest, 0, messages);
+ session.messageFlow(dest, 1, bytes); // Only need one sync
}
void SubscriptionManager::setFlowControl(
Modified: incubator/qpid/trunk/qpid/cpp/src/qpid/client/SubscriptionManager.h
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/qpid/client/SubscriptionManager.h?rev=660258&r1=660257&r2=660258&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/qpid/client/SubscriptionManager.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/qpid/client/SubscriptionManager.h Mon May 26 11:10:05 2008
@@ -45,10 +45,10 @@
typedef sys::Mutex::ScopedLock Lock;
typedef sys::Mutex::ScopedUnlock Unlock;
- Completion subscribeInternal(const std::string& q, const std::string& dest);
+ void subscribeInternal(const std::string& q, const std::string& dest);
qpid::client::Dispatcher dispatcher;
- qpid::client::Session& session;
+ qpid::client::Session session;
uint32_t messages;
uint32_t bytes;
bool window;
@@ -58,7 +58,7 @@
bool autoStop;
public:
- SubscriptionManager(Session& session);
+ SubscriptionManager(const Session& session);
/**
* Subscribe a MessagesListener to receive messages from queue.
@@ -68,7 +68,7 @@
*@param tag Unique destination tag for the listener.
* If not specified, the queue name is used.
*/
- Completion subscribe(MessageListener& listener,
+ void subscribe(MessageListener& listener,
const std::string& queue,
const std::string& tag=std::string());
@@ -79,7 +79,7 @@
*@param tag Unique destination tag for the listener.
* If not specified, the queue name is used.
*/
- Completion subscribe(LocalQueue& localQueue,
+ void subscribe(LocalQueue& localQueue,
const std::string& queue,
const std::string& tag=std::string());
Modified: incubator/qpid/trunk/qpid/cpp/src/tests/BrokerFixture.h
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/tests/BrokerFixture.h?rev=660258&r1=660257&r2=660258&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/tests/BrokerFixture.h (original)
+++ incubator/qpid/trunk/qpid/cpp/src/tests/BrokerFixture.h Mon May 26 11:10:05 2008
@@ -92,7 +92,7 @@
qpid::client::LocalQueue lq;
SessionFixtureT() : connection(broker->getPort()),
- session(connection.newSession(qpid::client::ASYNC)),
+ session(connection.newSession("SessionFixture")),
subs(session)
{}
Modified: incubator/qpid/trunk/qpid/cpp/src/tests/ClientSessionTest.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/tests/ClientSessionTest.cpp?rev=660258&r1=660257&r2=660258&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/tests/ClientSessionTest.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/tests/ClientSessionTest.cpp Mon May 26 11:10:05 2008
@@ -106,19 +106,19 @@
QPID_AUTO_TEST_CASE(testQueueQuery) {
ClientSessionFixture fix;
- fix.session = fix.connection.newSession(ASYNC);
+ fix.session = fix.connection.newSession();
fix.session.queueDeclare(queue="my-queue", alternateExchange="amq.fanout", exclusive=true, autoDelete=true);
- TypedResult<QueueQueryResult> result = fix.session.queueQuery(string("my-queue"));
- BOOST_CHECK_EQUAL(false, result.get().getDurable());
- BOOST_CHECK_EQUAL(true, result.get().getExclusive());
+ QueueQueryResult result = fix.session.queueQuery(string("my-queue"));
+ BOOST_CHECK_EQUAL(false, result.getDurable());
+ BOOST_CHECK_EQUAL(true, result.getExclusive());
BOOST_CHECK_EQUAL(string("amq.fanout"),
- result.get().getAlternateExchange());
+ result.getAlternateExchange());
}
QPID_AUTO_TEST_CASE(testTransfer)
{
ClientSessionFixture fix;
- fix.session=fix.connection.newSession(ASYNC);
+ fix.session=fix.connection.newSession();
fix.declareSubscribe();
fix.session.messageTransfer(acceptMode=1, content=TransferContent("my-message", "my-queue"));
//get & test the message:
@@ -133,7 +133,7 @@
QPID_AUTO_TEST_CASE(testDispatcher)
{
ClientSessionFixture fix;
- fix.session =fix.connection.newSession(ASYNC);
+ fix.session =fix.connection.newSession();
fix.declareSubscribe();
size_t count = 100;
for (size_t i = 0; i < count; ++i)
@@ -148,7 +148,7 @@
QPID_AUTO_TEST_CASE(testDispatcherThread)
{
ClientSessionFixture fix;
- fix.session =fix.connection.newSession(ASYNC);
+ fix.session =fix.connection.newSession();
fix.declareSubscribe();
size_t count = 10;
DummyListener listener(fix.session, "my-dest", count);
@@ -162,40 +162,42 @@
BOOST_CHECK_EQUAL(lexical_cast<string>(i), listener.messages[i].getData());
}
-QPID_AUTO_TEST_CASE_EXPECTED_FAILURES(testSuspend0Timeout, 1)
-{
- ClientSessionFixture fix;
- fix.session =fix.connection.newSession(ASYNC, 0);
- fix.session.suspend(); // session has 0 timeout.
- try {
- fix.connection.resume(fix.session);
- BOOST_FAIL("Expected InvalidArgumentException.");
- } catch(const InternalErrorException&) {}
-}
-
-QPID_AUTO_TEST_CASE_EXPECTED_FAILURES(testUseSuspendedError, 1)
-{
- ClientSessionFixture fix;
- fix.session =fix.connection.newSession(ASYNC, 60);
- fix.session.suspend();
- try {
- fix.session.exchangeQuery(name="amq.fanout");
- BOOST_FAIL("Expected session suspended exception");
- } catch(const CommandInvalidException&) {}
-}
+// FIXME aconway 2008-05-26: Re-enable with final resume implementation.
+//
+// QPID_AUTO_TEST_CASE_EXPECTED_FAILURES(testSuspend0Timeout, 1)
+// {
+// ClientSessionFixture fix;
+// fix.session.suspend(); // session has 0 timeout.
+// try {
+// fix.connection.resume(fix.session);
+// BOOST_FAIL("Expected InvalidArgumentException.");
+// } catch(const InternalErrorException&) {}
+// }
+
+// QPID_AUTO_TEST_CASE_EXPECTED_FAILURES(testUseSuspendedError, 1)
+// {
+// ClientSessionFixture fix;
+// fix.session =fix.session.timeout(60);
+// fix.session.suspend();
+// try {
+// fix.session.exchangeQuery(name="amq.fanout");
+// BOOST_FAIL("Expected session suspended exception");
+// } catch(const CommandInvalidException&) {}
+// }
+
+// QPID_AUTO_TEST_CASE_EXPECTED_FAILURES(testSuspendResume, 1)
+// {
+// ClientSessionFixture fix;
+// fix.session.timeout(60);
+// fix.declareSubscribe();
+// fix.session.suspend();
+// // Make sure we are still subscribed after resume.
+// fix.connection.resume(fix.session);
+// fix.session.messageTransfer(content=TransferContent("my-message", "my-queue"));
+// FrameSet::shared_ptr msg = fix.session.get();
+// BOOST_CHECK_EQUAL(string("my-message"), msg->getContent());
+// }
-QPID_AUTO_TEST_CASE_EXPECTED_FAILURES(testSuspendResume, 1)
-{
- ClientSessionFixture fix;
- fix.session =fix.connection.newSession(ASYNC, 60);
- fix.declareSubscribe();
- fix.session.suspend();
- // Make sure we are still subscribed after resume.
- fix.connection.resume(fix.session);
- fix.session.messageTransfer(content=TransferContent("my-message", "my-queue"));
- FrameSet::shared_ptr msg = fix.session.get();
- BOOST_CHECK_EQUAL(string("my-message"), msg->getContent());
-}
QPID_AUTO_TEST_CASE(testSendToSelf) {
ClientSessionFixture fix;
Modified: incubator/qpid/trunk/qpid/cpp/src/tests/XmlClientSessionTest.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/tests/XmlClientSessionTest.cpp?rev=660258&r1=660257&r2=660258&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/tests/XmlClientSessionTest.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/tests/XmlClientSessionTest.cpp Mon May 26 11:10:05 2008
@@ -121,7 +121,7 @@
QPID_AUTO_TEST_CASE(testXmlBinding) {
ClientSessionFixture f;
- Session session = f.connection.newSession(ASYNC);
+ Session session = f.connection.newSession();
SubscriptionManager subscriptions(session);
SubscribedLocalQueue localQueue(subscriptions);
Modified: incubator/qpid/trunk/qpid/cpp/src/tests/client_test.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/tests/client_test.cpp?rev=660258&r1=660257&r2=660258&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/tests/client_test.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/tests/client_test.cpp Mon May 26 11:10:05 2008
@@ -92,7 +92,7 @@
//Create and open a session on the connection through which
//most functionality is exposed:
- Session session = connection.newSession(ASYNC);
+ Session session = connection.newSession();
if (opts.verbose) std::cout << "Opened session." << std::endl;
Modified: incubator/qpid/trunk/qpid/cpp/src/tests/consume.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/tests/consume.cpp?rev=660258&r1=660257&r2=660258&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/tests/consume.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/tests/consume.cpp Mon May 26 11:10:05 2008
@@ -62,7 +62,7 @@
Client()
{
opts.open(connection);
- session = connection.newSession(ASYNC);
+ session = connection.newSession();
}
void consume()
Modified: incubator/qpid/trunk/qpid/cpp/src/tests/exception_test.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/tests/exception_test.cpp?rev=660258&r1=660257&r2=660258&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/tests/exception_test.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/tests/exception_test.cpp Mon May 26 11:10:05 2008
@@ -96,7 +96,6 @@
QPID_AUTO_TEST_CASE(NoSuchQueueTest) {
ProxySessionFixture fix;
- fix.session.setSynchronous(true);
BOOST_CHECK_THROW(fix.subs.subscribe(fix.lq, "no such queue"), NotFoundException);
}
Modified: incubator/qpid/trunk/qpid/cpp/src/tests/latencytest.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/tests/latencytest.cpp?rev=660258&r1=660257&r2=660258&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/tests/latencytest.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/tests/latencytest.cpp Mon May 26 11:10:05 2008
@@ -30,7 +30,7 @@
#include "TestOptions.h"
#include "qpid/client/Connection.h"
#include "qpid/client/Message.h"
-#include "qpid/client/Session.h"
+#include "qpid/client/AsyncSession.h"
#include "qpid/client/SubscriptionManager.h"
using namespace qpid;
@@ -99,7 +99,7 @@
{
protected:
Connection connection;
- Session session;
+ AsyncSession session;
Thread thread;
string queue;
@@ -157,7 +157,7 @@
Client::Client(const string& q) : queue(q)
{
opts.open(connection);
- session = connection.newSession(ASYNC);
+ session = connection.newSession();
}
void Client::start()
@@ -262,7 +262,7 @@
uint64_t sentAt(current_time());
msg.getDeliveryProperties().setTimestamp(sentAt);
//msg.getHeaders().setTimestamp("sent-at", sentAt);//TODO add support for uint64_t to field tables
- session.messageTransfer(arg::content=msg, arg::acceptMode=1);
+ async(session).messageTransfer(arg::content=msg, arg::acceptMode=1);
}
session.sync();
}
@@ -283,7 +283,7 @@
uint64_t sentAt(current_time());
msg.getDeliveryProperties().setTimestamp(sentAt);
//msg.getHeaders().setTimestamp("sent-at", sentAt);//TODO add support for uint64_t to field tables
- session.messageTransfer(arg::content=msg, arg::acceptMode=1);
+ async(session).messageTransfer(arg::content=msg, arg::acceptMode=1);
}
uint64_t timeTaken = (current_time() - start) / TIME_USEC;
if (timeTaken < 1000) {
Modified: incubator/qpid/trunk/qpid/cpp/src/tests/perftest.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/tests/perftest.cpp?rev=660258&r1=660257&r2=660258&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/tests/perftest.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/tests/perftest.cpp Mon May 26 11:10:05 2008
@@ -21,7 +21,7 @@
#include "TestOptions.h"
-#include "qpid/client/Session.h"
+#include "qpid/client/AsyncSession.h"
#include "qpid/client/SubscriptionManager.h"
#include "qpid/client/Connection.h"
#include "qpid/client/Completion.h"
@@ -194,12 +194,12 @@
struct Client : public Runnable {
Connection connection;
- Session session;
+ AsyncSession session;
Thread thread;
Client() {
opts.open(connection);
- session = connection.newSession(ASYNC);
+ session = connection.newSession();
}
~Client() {
@@ -431,7 +431,7 @@
offset = 5;
data += "data:";//marker (requested for latency testing tool scripts)
data += string(sizeof(size_t), 'X');//space for seq no
- data += string(reinterpret_cast<const char*>(session.getId().data()), session.getId().size());
+ data += session.getId().str();
if (opts.size > data.size()) {
data += string(opts.size - data.size(), 'X');
} else if(opts.size < data.size()) {
Modified: incubator/qpid/trunk/qpid/cpp/src/tests/publish.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/tests/publish.cpp?rev=660258&r1=660257&r2=660258&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/tests/publish.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/tests/publish.cpp Mon May 26 11:10:05 2008
@@ -28,7 +28,7 @@
#include "TestOptions.h"
#include "qpid/client/Connection.h"
#include "qpid/client/Message.h"
-#include "qpid/client/Session.h"
+#include "qpid/client/AsyncSession.h"
#include "qpid/client/SubscriptionManager.h"
using namespace qpid;
@@ -61,12 +61,12 @@
struct Client
{
Connection connection;
- Session session;
+ AsyncSession session;
Client()
{
opts.open(connection);
- session = connection.newSession(ASYNC);
+ session = connection.newSession();
}
std::string id(uint i)
Modified: incubator/qpid/trunk/qpid/cpp/src/tests/topic_listener.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/tests/topic_listener.cpp?rev=660258&r1=660257&r2=660258&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/tests/topic_listener.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/tests/topic_listener.cpp Mon May 26 11:10:05 2008
@@ -53,7 +53,7 @@
* defined.
*/
class Listener : public MessageListener{
- Session& session;
+ Session session;
SubscriptionManager& mgr;
const string responseQueue;
const bool transactional;
@@ -64,7 +64,7 @@
void shutdown();
void report();
public:
- Listener(Session& session, SubscriptionManager& mgr, const string& reponseQueue, bool tx);
+ Listener(const Session& session, SubscriptionManager& mgr, const string& reponseQueue, bool tx);
virtual void received(Message& msg);
};
@@ -101,7 +101,7 @@
else {
Connection connection;
args.open(connection);
- Session session = connection.newSession(ASYNC);
+ AsyncSession session = connection.newSession();
if (args.transactional) {
session.txSelect();
}
@@ -127,7 +127,8 @@
mgr.setFlowControl(SubscriptionManager::UNLIMITED, SubscriptionManager::UNLIMITED, false);
}
mgr.subscribe(listener, control);
-
+ session.sync();
+
cout << "topic_listener: listening..." << endl;
mgr.run();
if (args.durable) {
@@ -144,7 +145,7 @@
return 1;
}
-Listener::Listener(Session& s, SubscriptionManager& m, const string& _responseq, bool tx) :
+Listener::Listener(const Session& s, SubscriptionManager& m, const string& _responseq, bool tx) :
session(s), mgr(m), responseQueue(_responseq), transactional(tx), init(false), count(0){}
void Listener::received(Message& message){
Modified: incubator/qpid/trunk/qpid/cpp/src/tests/topic_publisher.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/tests/topic_publisher.cpp?rev=660258&r1=660257&r2=660258&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/tests/topic_publisher.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/tests/topic_publisher.cpp Mon May 26 11:10:05 2008
@@ -37,7 +37,7 @@
#include "TestOptions.h"
#include "qpid/client/Connection.h"
#include "qpid/client/MessageListener.h"
-#include "qpid/client/Session.h"
+#include "qpid/client/AsyncSession.h"
#include "qpid/client/SubscriptionManager.h"
#include "qpid/sys/Monitor.h"
#include <unistd.h>
@@ -56,7 +56,7 @@
* back by the subscribers.
*/
class Publisher {
- Session& session;
+ AsyncSession session;
SubscriptionManager mgr;
LocalQueue queue;
const string controlTopic;
@@ -66,7 +66,7 @@
string generateData(int size);
public:
- Publisher(Session& session, const string& controlTopic, bool tx, bool durable);
+ Publisher(const AsyncSession& session, const string& controlTopic, bool tx, bool durable);
int64_t publish(int msgs, int listeners, int size);
void terminate();
};
@@ -107,7 +107,7 @@
else {
Connection connection;
args.open(connection);
- Session session = connection.newSession(ASYNC);
+ AsyncSession session = connection.newSession();
if (args.transactional) {
session.txSelect();
}
@@ -150,7 +150,7 @@
return 1;
}
-Publisher::Publisher(Session& _session, const string& _controlTopic, bool tx, bool d) :
+Publisher::Publisher(const AsyncSession& _session, const string& _controlTopic, bool tx, bool d) :
session(_session), mgr(session), controlTopic(_controlTopic), transactional(tx), durable(d)
{
mgr.subscribe(queue, "response");
Modified: incubator/qpid/trunk/qpid/cpp/src/tests/topictest
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/tests/topictest?rev=660258&r1=660257&r2=660258&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/tests/topictest (original)
+++ incubator/qpid/trunk/qpid/cpp/src/tests/topictest Mon May 26 11:10:05 2008
@@ -36,5 +36,5 @@
subscribe $i &
done
# FIXME aconway 2007-03-27: Hack around startup race. Fix topic test.
-sleep 1
+sleep 2
publish 2>&1 || exit 1
Modified: incubator/qpid/trunk/qpid/cpp/src/tests/txtest.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/cpp/src/tests/txtest.cpp?rev=660258&r1=660257&r2=660258&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/cpp/src/tests/txtest.cpp (original)
+++ incubator/qpid/trunk/qpid/cpp/src/tests/txtest.cpp Mon May 26 11:10:05 2008
@@ -28,7 +28,7 @@
#include "TestOptions.h"
#include "qpid/client/Connection.h"
#include "qpid/client/Message.h"
-#include "qpid/client/Session.h"
+#include "qpid/client/AsyncSession.h"
#include "qpid/client/SubscriptionManager.h"
using namespace qpid;
@@ -96,12 +96,12 @@
struct Client
{
Connection connection;
- Session session;
+ AsyncSession session;
Client()
{
opts.open(connection);
- session = connection.newSession(ASYNC);
+ session = connection.newSession();
}
~Client()