You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by cc...@apache.org on 2009/07/25 02:44:34 UTC
svn commit: r797698 - in /qpid/trunk/qpid/cpp/src/qpid/replication:
ReplicationExchange.cpp ReplicationExchange.h
Author: cctrieloff
Date: Sat Jul 25 00:44:33 2009
New Revision: 797698
URL: http://svn.apache.org/viewvc?rev=797698&view=rev
Log:
Add missing mgnt counters for replication exchange
Modified:
qpid/trunk/qpid/cpp/src/qpid/replication/ReplicationExchange.cpp
qpid/trunk/qpid/cpp/src/qpid/replication/ReplicationExchange.h
Modified: qpid/trunk/qpid/cpp/src/qpid/replication/ReplicationExchange.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/replication/ReplicationExchange.cpp?rev=797698&r1=797697&r2=797698&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/replication/ReplicationExchange.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/replication/ReplicationExchange.cpp Sat Jul 25 00:44:33 2009
@@ -42,12 +42,18 @@
: Exchange(name, durable, _args, parent, broker), queues(qr), sequence(args.getAsInt64(SEQUENCE_VALUE)), init(false)
{
args.setInt64(SEQUENCE_VALUE, sequence);
+ if (mgmtExchange != 0)
+ mgmtExchange->set_type(typeName);
}
std::string ReplicationExchange::getType() const { return typeName; }
void ReplicationExchange::route(Deliverable& msg, const std::string& /*routingKey*/, const FieldTable* args)
{
+ if (mgmtExchange != 0) {
+ mgmtExchange->inc_msgReceives();
+ mgmtExchange->inc_byteReceives(msg.contentSize());
+ }
if (args) {
int eventType = args->getAsInt(REPLICATION_EVENT_TYPE);
if (eventType) {
@@ -57,7 +63,7 @@
handleEnqueueEvent(args, msg);
return;
case DEQUEUE:
- handleDequeueEvent(args);
+ handleDequeueEvent(args, msg);
return;
default:
throw IllegalArgumentException(QPID_MSG("Illegal value for " << REPLICATION_EVENT_TYPE << ": " << eventType));
@@ -65,6 +71,10 @@
}
} else {
QPID_LOG(warning, "Dropping unexpected message with no headers");
+ if (mgmtExchange != 0) {
+ mgmtExchange->inc_msgDrops();
+ mgmtExchange->inc_byteDrops(msg.contentSize());
+ }
}
}
@@ -79,12 +89,20 @@
headers.erase(REPLICATION_EVENT_TYPE);
msg.deliverTo(queue);
QPID_LOG(debug, "Enqueued replicated message onto " << queueName);
+ if (mgmtExchange != 0) {
+ mgmtExchange->inc_msgRoutes();
+ mgmtExchange->inc_byteRoutes( msg.contentSize());
+ }
} else {
QPID_LOG(error, "Cannot enqueue replicated message. Queue " << queueName << " does not exist");
+ if (mgmtExchange != 0) {
+ mgmtExchange->inc_msgDrops();
+ mgmtExchange->inc_byteDrops(msg.contentSize());
+ }
}
}
-void ReplicationExchange::handleDequeueEvent(const FieldTable* args)
+void ReplicationExchange::handleDequeueEvent(const FieldTable* args, Deliverable& msg)
{
std::string queueName = args->getAsString(REPLICATION_TARGET_QUEUE);
Queue::shared_ptr queue = queues.find(queueName);
@@ -94,11 +112,23 @@
if (queue->acquireMessageAt(position, dequeued)) {
queue->dequeue(0, dequeued);
QPID_LOG(debug, "Processed replicated 'dequeue' event from " << queueName << " at position " << position);
+ if (mgmtExchange != 0) {
+ mgmtExchange->inc_msgRoutes();
+ mgmtExchange->inc_byteRoutes(msg.contentSize());
+ }
} else {
QPID_LOG(warning, "Could not acquire message " << position << " from " << queueName);
+ if (mgmtExchange != 0) {
+ mgmtExchange->inc_msgDrops();
+ mgmtExchange->inc_byteDrops(msg.contentSize());
+ }
}
} else {
QPID_LOG(error, "Cannot process replicated 'dequeue' event. Queue " << queueName << " does not exist");
+ if (mgmtExchange != 0) {
+ mgmtExchange->inc_msgDrops();
+ mgmtExchange->inc_byteDrops(msg.contentSize());
+ }
}
}
Modified: qpid/trunk/qpid/cpp/src/qpid/replication/ReplicationExchange.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/replication/ReplicationExchange.h?rev=797698&r1=797697&r2=797698&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/replication/ReplicationExchange.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/replication/ReplicationExchange.h Sat Jul 25 00:44:33 2009
@@ -59,7 +59,7 @@
bool isDuplicate(const qpid::framing::FieldTable* args);
void handleEnqueueEvent(const qpid::framing::FieldTable* args, qpid::broker::Deliverable& msg);
- void handleDequeueEvent(const qpid::framing::FieldTable* args);
+ void handleDequeueEvent(const qpid::framing::FieldTable* args, qpid::broker::Deliverable& msg);
void encode(framing::Buffer& buffer) const;
};
}} // namespace qpid::replication
---------------------------------------------------------------------
Apache Qpid - AMQP Messaging Implementation
Project: http://qpid.apache.org
Use/Interact: mailto:commits-subscribe@qpid.apache.org