You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by as...@apache.org on 2007/02/06 15:19:51 UTC
svn commit: r504148 - in /incubator/qpid/branches/qpid.0-9: ./
cpp/lib/broker/BrokerAdapter.cpp cpp/tests/MessageHandlerTest.cpp
Author: astitcher
Date: Tue Feb 6 06:19:50 2007
New Revision: 504148
URL: http://svn.apache.org/viewvc?view=rev&rev=504148
Log:
r800@fuschia: andrew | 2007-01-17 17:34:13 +0000
Updated to latest upstream changes
Modified:
incubator/qpid/branches/qpid.0-9/ (props changed)
incubator/qpid/branches/qpid.0-9/cpp/lib/broker/BrokerAdapter.cpp
incubator/qpid/branches/qpid.0-9/cpp/tests/MessageHandlerTest.cpp
Propchange: incubator/qpid/branches/qpid.0-9/
------------------------------------------------------------------------------
--- svk:merge (original)
+++ svk:merge Tue Feb 6 06:19:50 2007
@@ -1 +1 @@
-8427bd24-ae5a-4eba-a324-d2fc9c9c6c77:/local/qpid.0-9.ams:799
+8427bd24-ae5a-4eba-a324-d2fc9c9c6c77:/local/qpid.0-9.ams:800
Modified: incubator/qpid/branches/qpid.0-9/cpp/lib/broker/BrokerAdapter.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/qpid.0-9/cpp/lib/broker/BrokerAdapter.cpp?view=diff&rev=504148&r1=504147&r2=504148
==============================================================================
--- incubator/qpid/branches/qpid.0-9/cpp/lib/broker/BrokerAdapter.cpp (original)
+++ incubator/qpid/branches/qpid.0-9/cpp/lib/broker/BrokerAdapter.cpp Tue Feb 6 06:19:50 2007
@@ -356,5 +356,198 @@
}
+//
+// Message class method handlers
+//
+void
+BrokerAdapter::MessageHandlerImpl::append( u_int16_t /*channel*/,
+ const string& /*reference*/,
+ const string& /*bytes*/ )
+{
+ assert(0); // FIXME astitcher 2007-01-11: 0-9 feature
+}
+
+
+void
+BrokerAdapter::MessageHandlerImpl::cancel( u_int16_t channel,
+ const string& destination )
+{
+ assert(0); // FIXME astitcher 2007-01-11: 0-9 feature
+
+ connection.getChannel(channel).cancel(destination);
+
+ connection.client->getMessageHandler()->ok(channel);
+}
+
+void
+BrokerAdapter::MessageHandlerImpl::checkpoint( u_int16_t /*channel*/,
+ const string& /*reference*/,
+ const string& /*identifier*/ )
+{
+ assert(0); // FIXME astitcher 2007-01-11: 0-9 feature
+}
+
+void
+BrokerAdapter::MessageHandlerImpl::close( u_int16_t /*channel*/,
+ const string& /*reference*/ )
+{
+ assert(0); // FIXME astitcher 2007-01-11: 0-9 feature
+}
+
+void
+BrokerAdapter::MessageHandlerImpl::consume( u_int16_t channelId,
+ u_int16_t /*ticket*/,
+ const string& queueName,
+ const string& destination,
+ bool noLocal,
+ bool noAck,
+ bool exclusive,
+ const qpid::framing::FieldTable& filter )
+{
+ assert(0); // FIXME astitcher 2007-01-11: 0-9 feature
+
+ Queue::shared_ptr queue = connection.getQueue(queueName, channelId);
+ Channel& channel = connection.getChannel(channelId);
+ if(!destination.empty() && channel.exists(destination)){
+ throw ConnectionException(530, "Consumer tags must be unique");
+ }
+
+ try{
+ string newTag = destination;
+ channel.consume(newTag, queue, !noAck, exclusive, noLocal ? &connection : 0, &filter);
+
+ connection.client->getMessageHandler()->ok(channelId);
+
+ //allow messages to be dispatched if required as there is now a consumer:
+ queue->dispatch();
+ }catch(ExclusiveAccessException& e){
+ if(exclusive) throw ChannelException(403, "Exclusive access cannot be granted");
+ else throw ChannelException(403, "Access would violate previously granted exclusivity");
+ }
+}
+
+void
+BrokerAdapter::MessageHandlerImpl::empty( u_int16_t /*channel*/ )
+{
+ assert(0); // FIXME astitcher 2007-01-11: 0-9 feature
+}
+
+void
+BrokerAdapter::MessageHandlerImpl::get( u_int16_t channelId,
+ u_int16_t /*ticket*/,
+ const string& queueName,
+ const string& /*destination*/,
+ bool noAck )
+{
+ assert(0); // FIXME astitcher 2007-01-11: 0-9 feature
+
+ Queue::shared_ptr queue = connection.getQueue(queueName, channelId);
+
+ // FIXME: get is probably Basic specific
+ if(!connection.getChannel(channelId).get(queue, !noAck)){
+
+ connection.client->getMessageHandler()->empty(channelId);
+ }
+
+}
+
+void
+BrokerAdapter::MessageHandlerImpl::offset( u_int16_t /*channel*/,
+ u_int64_t /*value*/ )
+{
+ assert(0); // FIXME astitcher 2007-01-11: 0-9 feature
+}
+
+void
+BrokerAdapter::MessageHandlerImpl::ok( u_int16_t /*channel*/ )
+{
+ assert(0); // FIXME astitcher 2007-01-11: 0-9 feature
+}
+
+void
+BrokerAdapter::MessageHandlerImpl::open( u_int16_t /*channel*/,
+ const string& /*reference*/ )
+{
+ assert(0); // FIXME astitcher 2007-01-11: 0-9 feature
+}
+
+void
+BrokerAdapter::MessageHandlerImpl::qos( u_int16_t channel,
+ u_int32_t prefetchSize,
+ u_int16_t prefetchCount,
+ bool /*global*/ )
+{
+ assert(0); // FIXME astitcher 2007-01-11: 0-9 feature
+
+ //TODO: handle global
+ connection.getChannel(channel).setPrefetchSize(prefetchSize);
+ connection.getChannel(channel).setPrefetchCount(prefetchCount);
+
+ connection.client->getMessageHandler()->ok(channel);
+}
+
+void
+BrokerAdapter::MessageHandlerImpl::recover( u_int16_t channel,
+ bool requeue )
+{
+ assert(0); // FIXME astitcher 2007-01-11: 0-9 feature
+
+ connection.getChannel(channel).recover(requeue);
+
+}
+
+void
+BrokerAdapter::MessageHandlerImpl::reject( u_int16_t /*channel*/,
+ u_int16_t /*code*/,
+ const string& /*text*/ )
+{
+ assert(0); // FIXME astitcher 2007-01-11: 0-9 feature
+}
+
+void
+BrokerAdapter::MessageHandlerImpl::resume( u_int16_t /*channel*/,
+ const string& /*reference*/,
+ const string& /*identifier*/ )
+{
+ assert(0); // FIXME astitcher 2007-01-11: 0-9 feature
+}
+
+void
+BrokerAdapter::MessageHandlerImpl::transfer( u_int16_t channel,
+ u_int16_t /*ticket*/,
+ const string& /*destination*/,
+ bool /*redelivered*/,
+ bool immediate,
+ u_int64_t /*ttl*/,
+ u_int8_t /*priority*/,
+ u_int64_t /*timestamp*/,
+ u_int8_t /*deliveryMode*/,
+ u_int64_t /*expiration*/,
+ const string& exchangeName,
+ const string& routingKey,
+ const string& /*messageId*/,
+ const string& /*correlationId*/,
+ const string& /*replyTo*/,
+ const string& /*contentType*/,
+ const string& /*contentEncoding*/,
+ const string& /*userId*/,
+ const string& /*appId*/,
+ const string& /*transactionId*/,
+ const string& /*securityToken*/,
+ const qpid::framing::FieldTable& /*applicationHeaders*/,
+ qpid::framing::Content /*body*/ )
+{
+ assert(0); // FIXME astitcher 2007-01-11: 0-9 feature
+
+ Exchange::shared_ptr exchange = exchangeName.empty() ?
+ connection.broker.getExchanges().getDefault() : connection.broker.getExchanges().get(exchangeName);
+ if(exchange){
+ Message* msg = new Message(&connection, exchangeName, routingKey, false /*mandatory?*/, immediate);
+ connection.getChannel(channel).handlePublish(msg, exchange);
+ }else{
+ throw ChannelException(404, "Exchange not found '" + exchangeName + "'");
+ }
+}
+
}} // namespace qpid::broker
Modified: incubator/qpid/branches/qpid.0-9/cpp/tests/MessageHandlerTest.cpp
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/qpid.0-9/cpp/tests/MessageHandlerTest.cpp?view=diff&rev=504148&r1=504147&r2=504148
==============================================================================
--- incubator/qpid/branches/qpid.0-9/cpp/tests/MessageHandlerTest.cpp (original)
+++ incubator/qpid/branches/qpid.0-9/cpp/tests/MessageHandlerTest.cpp Tue Feb 6 06:19:50 2007
@@ -23,7 +23,7 @@
#include <amqp_framing.h>
#include <qpid_test_plugin.h>
-#include <SessionHandlerFactoryImpl.h>
+#include <BrokerAdapter.h>
using namespace qpid::framing;
using namespace qpid::broker;