package com.spinn3r.artemis.activemq;

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.RedeliveryPolicy;
import org.junit.Ignore;
import org.junit.Test;

import javax.jms.*;

import java.util.Date;

import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;

/**
 *
 */
public class TestMessageRedelivery extends TestWithEmbeddedBroker2 {

    private static final int TIMEOUT = 60000;

    @Test
    @Ignore
    public void test1() throws Exception {

        int maximumRedeliveries = 5;

        String queueURL = "foo";

        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory( brokerURL );
        ActiveMQConnection connection = (ActiveMQConnection)connectionFactory.createConnection();

        connection.start();

        Session session = connection.createSession( false, Session.CLIENT_ACKNOWLEDGE );

        Destination dest = session.createQueue( queueURL );

        MessageProducer producer = session.createProducer( dest );

        String text = String.format( "At the tone, the time is %s BEEP.", new Date() );
        Message message = session.createTextMessage(text);

        System.out.printf( "Sending message ...\n" );
        producer.send( message );

        producer.close();
        connection.close();

        //*** now consume these messages.. .and never acknowledge them so they
        // should keep getting resent

        System.out.printf( "Receiving message...\n" );

        for (int i = 0; i < maximumRedeliveries; i++) {

            message = consumeOneMessage( dest, TIMEOUT );

            assertNotNull( message );

            TextMessage textMessage = (TextMessage)message;

            System.out.printf( "Received message %s (%s)\n" , (i+1), textMessage.getText() );

        }

        System.out.printf( "Seeing if we get a final message...\n" );

        message = consumeOneMessage( dest, 5000 );

        assertNull( message );

    }

    private Message consumeOneMessage( Destination dest, int timeout ) throws JMSException {

        ConnectionFactory connectionFactory = new ActiveMQConnectionFactory( brokerURL );

        ActiveMQConnection connection = (ActiveMQConnection)connectionFactory.createConnection();

        connection.start();
        Session session = connection.createSession( false, Session.CLIENT_ACKNOWLEDGE );
        MessageConsumer consumer = session.createConsumer( dest );

        Message message = consumer.receive( timeout );

        // TODO: we have to keep the connection open because activemq is smart enough
        // to realize that the connection closed without an ack.

        //connection.stop();
        //session.close();
        //consumer.close();

        return message;

    }

}