qpid-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From rob...@apache.org
Subject svn commit: r1546399 - in /qpid/jms/trunk/src: main/java/org/apache/qpid/jms/engine/ main/java/org/apache/qpid/jms/impl/ test/java/org/apache/qpid/jms/ test/java/org/apache/qpid/jms/impl/
Date Thu, 28 Nov 2013 17:15:05 GMT
Author: robbie
Date: Thu Nov 28 17:15:04 2013
New Revision: 1546399

URL: http://svn.apache.org/r1546399
Log:
QPIDJMS-9: add Message support for JMSTimestamp operations, with associated handling for producers

Added:
    qpid/jms/trunk/src/test/java/org/apache/qpid/jms/SenderIntegrationTest.java
Modified:
    qpid/jms/trunk/src/main/java/org/apache/qpid/jms/engine/AmqpMessage.java
    qpid/jms/trunk/src/main/java/org/apache/qpid/jms/impl/MessageImpl.java
    qpid/jms/trunk/src/main/java/org/apache/qpid/jms/impl/SenderImpl.java
    qpid/jms/trunk/src/test/java/org/apache/qpid/jms/SessionIntegrationTest.java
    qpid/jms/trunk/src/test/java/org/apache/qpid/jms/impl/MessageImplTest.java
    qpid/jms/trunk/src/test/java/org/apache/qpid/jms/impl/SenderImplTest.java

Modified: qpid/jms/trunk/src/main/java/org/apache/qpid/jms/engine/AmqpMessage.java
URL: http://svn.apache.org/viewvc/qpid/jms/trunk/src/main/java/org/apache/qpid/jms/engine/AmqpMessage.java?rev=1546399&r1=1546398&r2=1546399&view=diff
==============================================================================
--- qpid/jms/trunk/src/main/java/org/apache/qpid/jms/engine/AmqpMessage.java (original)
+++ qpid/jms/trunk/src/main/java/org/apache/qpid/jms/engine/AmqpMessage.java Thu Nov 28 17:15:04
2013
@@ -210,6 +210,16 @@ public abstract class AmqpMessage
         _message.setAddress(to);
     }
 
+    public long getCreationTime()
+    {
+        return _message.getCreationTime();
+    }
+
+    public void setCreationTime(long timeInMillis)
+    {
+        _message.setCreationTime(timeInMillis);
+    }
+
     //===== Application Properties ======
 
     private void createApplicationProperties()

Modified: qpid/jms/trunk/src/main/java/org/apache/qpid/jms/impl/MessageImpl.java
URL: http://svn.apache.org/viewvc/qpid/jms/trunk/src/main/java/org/apache/qpid/jms/impl/MessageImpl.java?rev=1546399&r1=1546398&r2=1546399&view=diff
==============================================================================
--- qpid/jms/trunk/src/main/java/org/apache/qpid/jms/impl/MessageImpl.java (original)
+++ qpid/jms/trunk/src/main/java/org/apache/qpid/jms/impl/MessageImpl.java Thu Nov 28 17:15:04
2013
@@ -132,15 +132,13 @@ public abstract class MessageImpl<T exte
     @Override
     public long getJMSTimestamp() throws JMSException
     {
-        // TODO Auto-generated method stub
-        throw new UnsupportedOperationException("Not Implemented");
+        return _amqpMessage.getCreationTime();
     }
 
     @Override
     public void setJMSTimestamp(long timestamp) throws JMSException
     {
-        // TODO Auto-generated method stub
-        throw new UnsupportedOperationException("Not Implemented");
+        _amqpMessage.setCreationTime(timestamp);
     }
 
     @Override

Modified: qpid/jms/trunk/src/main/java/org/apache/qpid/jms/impl/SenderImpl.java
URL: http://svn.apache.org/viewvc/qpid/jms/trunk/src/main/java/org/apache/qpid/jms/impl/SenderImpl.java?rev=1546399&r1=1546398&r2=1546399&view=diff
==============================================================================
--- qpid/jms/trunk/src/main/java/org/apache/qpid/jms/impl/SenderImpl.java (original)
+++ qpid/jms/trunk/src/main/java/org/apache/qpid/jms/impl/SenderImpl.java Thu Nov 28 17:15:04
2013
@@ -46,6 +46,11 @@ public class SenderImpl extends LinkImpl
         getConnectionImpl().lock();
         try
         {
+            long timestamp = System.currentTimeMillis();
+
+            //set the timestamp
+            message.setJMSTimestamp(timestamp);
+
             //set the Destination
             message.setJMSDestination(_destination);
 

Added: qpid/jms/trunk/src/test/java/org/apache/qpid/jms/SenderIntegrationTest.java
URL: http://svn.apache.org/viewvc/qpid/jms/trunk/src/test/java/org/apache/qpid/jms/SenderIntegrationTest.java?rev=1546399&view=auto
==============================================================================
--- qpid/jms/trunk/src/test/java/org/apache/qpid/jms/SenderIntegrationTest.java (added)
+++ qpid/jms/trunk/src/test/java/org/apache/qpid/jms/SenderIntegrationTest.java Thu Nov 28
17:15:04 2013
@@ -0,0 +1,102 @@
+/*
+ * 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.
+ */
+package org.apache.qpid.jms;
+
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.greaterThanOrEqualTo;
+
+import java.util.Calendar;
+import java.util.Date;
+
+import javax.jms.Connection;
+import javax.jms.Message;
+import javax.jms.MessageProducer;
+import javax.jms.Queue;
+import javax.jms.Session;
+
+import org.apache.qpid.jms.test.testpeer.TestAmqpPeer;
+import org.apache.qpid.jms.test.testpeer.matchers.sections.MessageHeaderSectionMatcher;
+import org.apache.qpid.jms.test.testpeer.matchers.sections.MessagePropertiesSectionMatcher;
+import org.apache.qpid.jms.test.testpeer.matchers.sections.TransferPayloadCompositeMatcher;
+import org.apache.qpid.jms.test.testpeer.matchers.types.EncodedAmqpValueMatcher;
+import org.junit.Test;
+
+public class SenderIntegrationTest extends QpidJmsTestCase
+{
+    private final IntegrationTestFixture _testFixture = new IntegrationTestFixture();
+
+    @Test
+    public void testSendingMessageSetsJMSDestination() throws Exception
+    {
+        try(TestAmqpPeer testPeer = new TestAmqpPeer(IntegrationTestFixture.PORT);)
+        {
+            Connection connection = _testFixture.establishConnecton(testPeer);
+            testPeer.expectBegin();
+            testPeer.expectSenderAttach();
+
+            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+            String queueName = "myQueue";
+            Queue queue = session.createQueue(queueName);
+            MessageProducer producer = session.createProducer(queue);
+
+            String text = "myMessage";
+            MessageHeaderSectionMatcher headersMatcher = new MessageHeaderSectionMatcher(true).withDurable(equalTo(true));
+            MessagePropertiesSectionMatcher propsMatcher = new MessagePropertiesSectionMatcher(true).withTo(equalTo(queueName));
+            TransferPayloadCompositeMatcher messageMatcher = new TransferPayloadCompositeMatcher();
+            messageMatcher.setHeadersMatcher(headersMatcher);
+            messageMatcher.setPropertiesMatcher(propsMatcher);
+            messageMatcher.setMessageContentMatcher(new EncodedAmqpValueMatcher(text));
+            testPeer.expectTransfer(messageMatcher);
+
+            Message message = session.createTextMessage(text);
+
+            producer.send(message);
+        }
+    }
+
+    @Test
+    public void testSendingMessageSetsJMSTimestamp() throws Exception
+    {
+        try(TestAmqpPeer testPeer = new TestAmqpPeer(IntegrationTestFixture.PORT);)
+        {
+            Connection connection = _testFixture.establishConnecton(testPeer);
+            testPeer.expectBegin();
+            testPeer.expectSenderAttach();
+
+            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+            String queueName = "myQueue";
+            Queue queue = session.createQueue(queueName);
+            MessageProducer producer = session.createProducer(queue);
+
+            Date currentTime = Calendar.getInstance().getTime();
+            String text = "myMessage";
+            MessageHeaderSectionMatcher headersMatcher = new MessageHeaderSectionMatcher(true).withDurable(equalTo(true));
+            MessagePropertiesSectionMatcher propsMatcher = new MessagePropertiesSectionMatcher(true).withCreationTime(greaterThanOrEqualTo(currentTime));
+            TransferPayloadCompositeMatcher messageMatcher = new TransferPayloadCompositeMatcher();
+            messageMatcher.setHeadersMatcher(headersMatcher);
+            messageMatcher.setPropertiesMatcher(propsMatcher);
+            messageMatcher.setMessageContentMatcher(new EncodedAmqpValueMatcher(text));
+            testPeer.expectTransfer(messageMatcher);
+
+            Message message = session.createTextMessage(text);
+
+            producer.send(message);
+        }
+    }
+}

Modified: qpid/jms/trunk/src/test/java/org/apache/qpid/jms/SessionIntegrationTest.java
URL: http://svn.apache.org/viewvc/qpid/jms/trunk/src/test/java/org/apache/qpid/jms/SessionIntegrationTest.java?rev=1546399&r1=1546398&r2=1546399&view=diff
==============================================================================
--- qpid/jms/trunk/src/test/java/org/apache/qpid/jms/SessionIntegrationTest.java (original)
+++ qpid/jms/trunk/src/test/java/org/apache/qpid/jms/SessionIntegrationTest.java Thu Nov 28
17:15:04 2013
@@ -85,7 +85,7 @@ public class SessionIntegrationTest exte
 
             String text = "myMessage";
             MessageHeaderSectionMatcher headersMatcher = new MessageHeaderSectionMatcher(true).withDurable(equalTo(true));
-            MessagePropertiesSectionMatcher propsMatcher = new MessagePropertiesSectionMatcher(true).withTo(equalTo(queueName));
+            MessagePropertiesSectionMatcher propsMatcher = new MessagePropertiesSectionMatcher(true);
             TransferPayloadCompositeMatcher messageMatcher = new TransferPayloadCompositeMatcher();
             messageMatcher.setHeadersMatcher(headersMatcher);
             messageMatcher.setPropertiesMatcher(propsMatcher);
@@ -143,7 +143,7 @@ public class SessionIntegrationTest exte
             MessageProducer producer = session.createProducer(queue);
 
             MessageHeaderSectionMatcher headersMatcher = new MessageHeaderSectionMatcher(true).withDurable(equalTo(true));
-            MessagePropertiesSectionMatcher propsMatcher = new MessagePropertiesSectionMatcher(true).withTo(equalTo(queueName));
+            MessagePropertiesSectionMatcher propsMatcher = new MessagePropertiesSectionMatcher(true);
             TransferPayloadCompositeMatcher messageMatcher = new TransferPayloadCompositeMatcher();
             messageMatcher.setHeadersMatcher(headersMatcher);
             messageMatcher.setPropertiesMatcher(propsMatcher);

Modified: qpid/jms/trunk/src/test/java/org/apache/qpid/jms/impl/MessageImplTest.java
URL: http://svn.apache.org/viewvc/qpid/jms/trunk/src/test/java/org/apache/qpid/jms/impl/MessageImplTest.java?rev=1546399&r1=1546398&r2=1546399&view=diff
==============================================================================
--- qpid/jms/trunk/src/test/java/org/apache/qpid/jms/impl/MessageImplTest.java (original)
+++ qpid/jms/trunk/src/test/java/org/apache/qpid/jms/impl/MessageImplTest.java Thu Nov 28
17:15:04 2013
@@ -480,6 +480,44 @@ public class MessageImplTest extends Qpi
         assertEquals(newQueueExpected, _testMessage.getJMSDestination());
     }
 
+    // ====== JMSTimestamp =======
+
+    @Test
+    public void testGetJMSTimestampOnNewMessage() throws Exception
+    {
+        assertEquals("expected JMSTimestamp value not present", 0, _testMessage.getJMSTimestamp());
+    }
+
+    @Test
+    public void testSetGetJMSTimestampOnNewMessage() throws Exception
+    {
+        long timestamp = System.currentTimeMillis();
+
+        _testMessage.setJMSTimestamp(timestamp);
+        assertEquals("expected JMSTimestamp value not present", timestamp, _testMessage.getJMSTimestamp());
+    }
+
+    @Test
+    public void testSetJMSTimestampOnNewMessage() throws Exception
+    {
+        assertEquals(0, _testAmqpMessage.getCreationTime());
+
+        long timestamp = System.currentTimeMillis();
+        _testMessage.setJMSTimestamp(timestamp);
+
+        assertEquals(timestamp, _testAmqpMessage.getCreationTime());
+    }
+
+    @Test
+    public void testGetJMSTimestampOnRecievedMessageWithCreationTime() throws Exception
+    {
+        long timestamp = System.currentTimeMillis();
+        _testAmqpMessage.setCreationTime(timestamp);
+        _testMessage = new TestMessageImpl(_testAmqpMessage, _mockSessionImpl, _mockConnectionImpl);
+
+        assertEquals("expected JMSTimestamp value not present", timestamp, _testMessage.getJMSTimestamp());
+    }
+
     // ====== utility methods =======
 
     private void assertGetPropertyThrowsMessageFormatException(TestMessageImpl testMessage,

Modified: qpid/jms/trunk/src/test/java/org/apache/qpid/jms/impl/SenderImplTest.java
URL: http://svn.apache.org/viewvc/qpid/jms/trunk/src/test/java/org/apache/qpid/jms/impl/SenderImplTest.java?rev=1546399&r1=1546398&r2=1546399&view=diff
==============================================================================
--- qpid/jms/trunk/src/test/java/org/apache/qpid/jms/impl/SenderImplTest.java (original)
+++ qpid/jms/trunk/src/test/java/org/apache/qpid/jms/impl/SenderImplTest.java Thu Nov 28 17:15:04
2013
@@ -101,4 +101,27 @@ public class SenderImplTest extends Qpid
 
         assertSame(_mockQueue, testMessage.getJMSDestination());
     }
+
+    @Test
+    public void testSenderSetsJMSTimestampOnMessage() throws Exception
+    {
+        //Create mock sent message token, ensure that it is immediately marked as Accepted
+        AmqpSentMessageToken _mockToken = Mockito.mock(AmqpSentMessageToken.class);
+        Mockito.when(_mockToken.getRemoteDeliveryState()).thenReturn(Accepted.getInstance());
+        Mockito.when(_mockAmqpSender.sendMessage(Mockito.any(AmqpMessage.class))).thenReturn(_mockToken);
+        ImmediateWaitUntil.mockWaitUntil(_mockConnection);
+
+        SenderImpl senderImpl = new SenderImpl(_mockSession, _mockConnection, _mockAmqpSender,
_mockQueue);
+
+        TestAmqpMessage testAmqpMessage = new TestAmqpMessage();
+        TestMessageImpl testMessage = new TestMessageImpl(testAmqpMessage, null, null);
+
+        assertEquals(0, testMessage.getJMSTimestamp());
+        long timestamp = System.currentTimeMillis();
+
+        senderImpl.send(testMessage);
+
+        //verify the timestamp was set, allowing for a 3second delta
+        assertEquals(timestamp, testMessage.getJMSTimestamp(), 3000);
+    }
 }



---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org


Mime
View raw message