/*
 * JBoss, the OpenSource J2EE webOS
 *
 * Distributable under LGPL license.
 * See terms of license at gnu.org.
 */
package org.jboss.test.jbossmq.test;

import java.util.Enumeration;
import java.util.Collections;
import java.util.List;

import javax.jms.DeliveryMode;
import javax.jms.QueueSession;
import javax.jms.Queue;
import javax.jms.QueueSender;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.jms.QueueBrowser;
import javax.jms.QueueReceiver;
import javax.management.ObjectName;
import javax.management.Attribute;

import org.jboss.mq.server.MessageCacheMBean;
import org.jboss.test.JBossTestSetup;

import junit.framework.TestSuite;
import junit.framework.Test;

/**
 * ExpiryDestinationTestCase tests putting messages on a separate queue.
 *
 * @author Elias Ross <genman@noderunner.net>
 * @version $Revision: 1.2.6.3 $
 */
public class ExpiryDestinationTestCase
   extends JBossMQUnitTest
{

   static String DLQ_QUEUE = "queue/DLQ";
   
   public ExpiryDestinationTestCase(String name) throws Exception
   {
      super(name);
      getLog().debug(getName());
   }

   private List list(QueueSession session, Queue queue) throws Exception
   {
      QueueBrowser browser = session.createBrowser( queue );
      Enumeration e = browser.getEnumeration();
      return Collections.list(e);
   }

   private void assertSize(QueueSession session, Queue queue, int size) throws Exception
   {
      List l = list(session, queue);
      assertEquals("Got " + l, size, l.size());
   }

   protected void setUp()
   {
      super.setUp();
      ObjectName dlq = new ObjectName("jboss.mq.destination:service=Queue,name=DLQ");
      getServer().invoke(dlq, "removeAllMessages", null, null);
      ObjectName tq = new ObjectName("jboss.mq.destination:service=Queue,name=testQueue");
      getServer().invoke(tq, "removeAllMessages", null, null);
      getServer().setAttribute(tq, new Attribute("ExpiryDestination", dlq));
      getServer().invoke(tq, "stop", null, null);
      getServer().invoke(tq, "start", null, null);

      connect();
      queueConnection.start();
      drainQueue();
   }

   protected void tearDown()
   {
      super.tearDown();
   }
   
   /**
    * Test that expired messages are moved to a separate queue.
    */
   public void testExpiredMessagesMove() throws Exception
   {
      QueueSession session = queueConnection.createQueueSession(false, Session.AUTO_ACKNOWLEDGE);
      Queue queue = (Queue)context.lookup(TEST_QUEUE);
      QueueSender sender = session.createSender(queue);

      Queue queue2 = (Queue)context.lookup(DLQ_QUEUE);
      QueueBrowser browser = session.createBrowser( queue );

      long now = System.currentTimeMillis();

      TextMessage message = session.createTextMessage();
      message.setStringProperty("foo", "bar");
      String text = "expire on server";
      message.setText(text);
      sender.send(message, DeliveryMode.PERSISTENT, 4, 1);

      Thread.sleep(1000);

      assertSize(session, queue,  0);
      assertSize(session, queue2, 1);

      QueueReceiver receiver = session.createReceiver(queue2);
      message = (TextMessage) receiver.receiveNoWait();
      assertEquals("QUEUE.testQueue", message.getStringProperty("JBOSS_ORIG_DESTINATION"));
      assertTrue(message.getLongProperty("JBOSS_ORIG_EXPIRATION") > now);
      assertEquals(0L, message.getJMSExpiration());
      assertEquals(text, message.getText());
      assertEquals("bar", message.getStringProperty("foo"));
      receiver.close();
      session.close();

      text = "expire on client";
      QueueSession session2 = queueConnection.createQueueSession(false, Session.CLIENT_ACKNOWLEDGE);
      QueueReceiver receiver2 = session2.createReceiver(queue);
      message = session2.createTextMessage();
      message.setText(text);
      sender = session2.createSender(queue);
      sender.send(message, DeliveryMode.PERSISTENT, 4, 500);
      message.setText(text + " 2");
      sender.send(message, DeliveryMode.PERSISTENT, 4, 500);
      queueConnection.stop();
      queueConnection.start();
      message = (TextMessage) receiver2.receive(1000);
      assertEquals(text, message.getText());
      Thread.sleep(1000);
      message = (TextMessage) receiver2.receive(1000);
      assertTrue(message == null);
      receiver2.close();
      session2.recover();
      session2.close();

      QueueSession session3 = queueConnection.createQueueSession(false, Session.CLIENT_ACKNOWLEDGE);
      assertSize(session3, queue, 0);
      assertSize(session3, queue2, 2);

      disconnect();
      getLog().debug("passed");
   }

   public static Test suite() throws Exception
   {
      TestSuite suite = new TestSuite();
      suite.addTest(new JBossTestSetup(new ExpiryDestinationTestCase("testExpiredMessagesMove")));
      return suite;
   }

}