You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@activemq.apache.org by ta...@apache.org on 2013/04/16 23:24:05 UTC
svn commit: r1468614 - in
/activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src:
main/csharp/Util/ConnectionAudit.cs test/csharp/Util/ConnectionAuditTest.cs
Author: tabish
Date: Tue Apr 16 21:24:05 2013
New Revision: 1468614
URL: http://svn.apache.org/r1468614
Log:
Add class ConnectionAudit and a unit test for it
Added:
activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/Util/ConnectionAudit.cs (with props)
activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/test/csharp/Util/ConnectionAuditTest.cs (with props)
Added: activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/Util/ConnectionAudit.cs
URL: http://svn.apache.org/viewvc/activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/Util/ConnectionAudit.cs?rev=1468614&view=auto
==============================================================================
--- activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/Util/ConnectionAudit.cs (added)
+++ activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/Util/ConnectionAudit.cs Tue Apr 16 21:24:05 2013
@@ -0,0 +1,139 @@
+/*
+ * 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.
+ */
+using System;
+using System.Collections.Generic;
+using Apache.NMS.ActiveMQ.Commands;
+
+namespace Apache.NMS.ActiveMQ.Util
+{
+ public class ConnectionAudit
+ {
+ private readonly object mutex = new object();
+
+ private readonly Dictionary<ActiveMQDestination, ActiveMQMessageAudit> destinations =
+ new Dictionary<ActiveMQDestination, ActiveMQMessageAudit>();
+ private readonly Dictionary<IDispatcher, ActiveMQMessageAudit> dispatchers =
+ new Dictionary<IDispatcher, ActiveMQMessageAudit>();
+
+ private bool checkForDuplicates = true;
+ public bool CheckForDuplicates
+ {
+ get { return this.checkForDuplicates; }
+ set { this.checkForDuplicates = value; }
+ }
+
+ private int auditDepth = ActiveMQMessageAudit.DEFAULT_WINDOW_SIZE;
+ public int AuditDepth
+ {
+ get { return this.auditDepth; }
+ set { this.auditDepth = value; }
+ }
+
+ private int auditMaximumProducerNumber = ActiveMQMessageAudit.MAXIMUM_PRODUCER_COUNT;
+ public int AuditMaximumProducerNumber
+ {
+ get { return this.auditMaximumProducerNumber; }
+ set { this.auditMaximumProducerNumber = value; }
+ }
+
+ public ConnectionAudit()
+ {
+ }
+
+ public ConnectionAudit(int auditDepth, int auditMaximumProducerNumber)
+ {
+ this.auditDepth = auditDepth;
+ this.auditMaximumProducerNumber = auditMaximumProducerNumber;
+ }
+
+ public void RemoveDispatcher(IDispatcher dispatcher)
+ {
+ lock(this.mutex)
+ {
+ dispatchers.Remove(dispatcher);
+ }
+ }
+
+ public bool IsDuplicate(IDispatcher dispatcher, Message message)
+ {
+ bool result = false;
+
+ lock(this.mutex)
+ {
+ if (checkForDuplicates && message != null)
+ {
+ ActiveMQDestination destination = message.Destination;
+ if (destination != null)
+ {
+ if (destination.IsQueue)
+ {
+ ActiveMQMessageAudit audit = null;
+ if (!destinations.TryGetValue(destination, out audit))
+ {
+ audit = new ActiveMQMessageAudit(auditDepth, auditMaximumProducerNumber);
+ destinations.Add(destination, audit);
+ }
+ result = audit.IsDuplicate(message.MessageId);
+ }
+ else
+ {
+ ActiveMQMessageAudit audit = null;
+ if (!dispatchers.TryGetValue(dispatcher, out audit))
+ {
+ audit = new ActiveMQMessageAudit(auditDepth, auditMaximumProducerNumber);
+ dispatchers.Add(dispatcher, audit);
+ }
+ result = audit.IsDuplicate(message.MessageId);
+ }
+ }
+ }
+ }
+ return result;
+ }
+
+ public void RollbackDuplicate(IDispatcher dispatcher, Message message)
+ {
+ lock(this.mutex)
+ {
+ if (checkForDuplicates && message != null)
+ {
+ ActiveMQDestination destination = message.Destination;
+ if (destination != null)
+ {
+ if (destination.IsQueue)
+ {
+ ActiveMQMessageAudit audit = null;
+ if (destinations.TryGetValue(destination, out audit))
+ {
+ audit.Rollback(message.MessageId);
+ }
+ }
+ else
+ {
+ ActiveMQMessageAudit audit = null;
+ if (dispatchers.TryGetValue(dispatcher, out audit))
+ {
+ audit.Rollback(message.MessageId);
+ }
+ }
+ }
+ }
+ }
+ }
+ }
+}
+
Propchange: activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/Util/ConnectionAudit.cs
------------------------------------------------------------------------------
svn:eol-style = native
Added: activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/test/csharp/Util/ConnectionAuditTest.cs
URL: http://svn.apache.org/viewvc/activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/test/csharp/Util/ConnectionAuditTest.cs?rev=1468614&view=auto
==============================================================================
--- activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/test/csharp/Util/ConnectionAuditTest.cs (added)
+++ activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/test/csharp/Util/ConnectionAuditTest.cs Tue Apr 16 21:24:05 2013
@@ -0,0 +1,133 @@
+/*
+ * 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.
+ */
+
+using System;
+using System.Collections.Generic;
+using Apache.NMS;
+using Apache.NMS.ActiveMQ;
+using Apache.NMS.ActiveMQ.Commands;
+using Apache.NMS.ActiveMQ.Util;
+using NUnit.Framework;
+
+namespace Apache.NMS.ActiveMQ.Test
+{
+ [TestFixture]
+ public class ConnectionAuditTest
+ {
+ internal class MyDispatcher : IDispatcher
+ {
+ public void Dispatch(MessageDispatch messageDispatch)
+ {
+ }
+ }
+
+ [Test]
+ public void TestConstructor1()
+ {
+ ConnectionAudit audit = new ConnectionAudit();
+ Assert.IsTrue(audit.CheckForDuplicates);
+ Assert.IsTrue(audit.AuditDepth == ActiveMQMessageAudit.DEFAULT_WINDOW_SIZE);
+ Assert.IsTrue(audit.AuditMaximumProducerNumber == ActiveMQMessageAudit.MAXIMUM_PRODUCER_COUNT);
+ }
+
+ [Test]
+ public void TestConstructor2()
+ {
+ ConnectionAudit audit = new ConnectionAudit(100, 200);
+ Assert.IsTrue(audit.CheckForDuplicates);
+ Assert.IsTrue(audit.AuditDepth == 100);
+ Assert.IsTrue(audit.AuditMaximumProducerNumber == 200);
+ }
+
+ [Test]
+ public void testIsDuplicate()
+ {
+ int count = 10000;
+ ConnectionAudit audit = new ConnectionAudit();
+ List<MessageId> list = new List<MessageId>();
+ MyDispatcher dispatcher = new MyDispatcher();
+
+ ProducerId pid = new ProducerId();
+ pid.ConnectionId = "test";
+ pid.SessionId = 0;
+ pid.Value = 1;
+
+ ActiveMQDestination destination = new ActiveMQQueue("TEST.QUEUE");
+ Message message = new Message();
+ message.Destination = destination;
+
+ for (int i = 0; i < count; i++)
+ {
+ MessageId id = new MessageId();
+ id.ProducerId = pid;
+ id.ProducerSequenceId = i;
+ list.Add(id);
+
+ message.MessageId = id;
+ Assert.IsFalse(audit.IsDuplicate(dispatcher, message));
+ }
+
+ int index = list.Count -1 -audit.AuditDepth;
+ for (; index < list.Count; index++)
+ {
+ MessageId id = list[index];
+ message.MessageId = id;
+ Assert.IsTrue(audit.IsDuplicate(dispatcher, message), "duplicate msg:" + id);
+ }
+ }
+
+ [Test]
+ public void testRollbackDuplicate()
+ {
+ int count = 10000;
+ ConnectionAudit audit = new ConnectionAudit();
+ List<MessageId> list = new List<MessageId>();
+ MyDispatcher dispatcher = new MyDispatcher();
+
+ ProducerId pid = new ProducerId();
+ pid.ConnectionId = "test";
+ pid.SessionId = 0;
+ pid.Value = 1;
+
+ ActiveMQDestination destination = new ActiveMQQueue("TEST.QUEUE");
+ Message message = new Message();
+ message.Destination = destination;
+
+ for (int i = 0; i < count; i++)
+ {
+ MessageId id = new MessageId();
+ id.ProducerId = pid;
+ id.ProducerSequenceId = i;
+ list.Add(id);
+
+ message.MessageId = id;
+ Assert.IsFalse(audit.IsDuplicate(dispatcher, message));
+ }
+
+ int index = list.Count -1 -audit.AuditDepth;
+ for (; index < list.Count; index++)
+ {
+ MessageId id = list[index];
+ message.MessageId = id;
+ Assert.IsTrue(audit.IsDuplicate(dispatcher, message), "duplicate msg:" + id);
+ audit.RollbackDuplicate(dispatcher, message);
+ Assert.IsFalse(audit.IsDuplicate(dispatcher, message), "error: duplicate msg:" + id);
+ }
+ }
+ }
+}
+
Propchange: activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/test/csharp/Util/ConnectionAuditTest.cs
------------------------------------------------------------------------------
svn:eol-style = native