/*
 * 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.client;

import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashSet;
import java.util.List;

import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.MessageProducer;
import javax.jms.StreamMessage;

import org.apache.qpid.client.message.AMQPEncodedListMessage;
import org.apache.qpid.client.message.UnprocessedMessage;
import org.apache.qpid.test.utils.QpidTestCase;
import org.apache.qpid.transport.*;
import org.apache.qpid.transport.Connection.SessionFactory;
import org.apache.qpid.transport.Connection.State;
import org.apache.qpid.url.AMQBindingURL;

/**
 * Tests AMQSession_0_10 methods.
 * <p>
 * The main purpose of the tests in this test suite is to check that
 * {@link SessionException} is not thrown from methods of
 * {@link AMQSession_0_10}.
 */
public class AMQSession_0_10Test extends QpidTestCase
{

    public void testExceptionOnCommit()
    {
        AMQSession_0_10 session = createThrowingExceptionAMQSession_0_10();
        try
        {
            session.commit();
            fail("JMSException should be thrown");
        }
        catch (Exception e)
        {
            assertTrue("JMSException is expected", e instanceof JMSException);
            assertEquals("541 error code is expected", "541", ((JMSException) e).getErrorCode());
        }
    }

    public void testExceptionOnCreateMessageProducer()
    {
        AMQSession_0_10 session = createThrowingExceptionAMQSession_0_10();
        try
        {
            session.createMessageProducer(createDestination(), true, true, 1l);
            fail("JMSException should be thrown");
        }
        catch (Exception e)
        {
            assertTrue("JMSException is expected but got:" + e, e instanceof JMSException);
            assertEquals("541 error code is expected", "541", ((JMSException) e).getErrorCode());
        }
    }

    public void testExceptionOnRollback()
    {
        AMQSession_0_10 session = createThrowingExceptionAMQSession_0_10();
        try
        {
            session.rollback();
            fail("JMSException should be thrown");
        }
        catch (Exception e)
        {
            assertTrue("JMSException is expected", e instanceof JMSException);
        }
    }

    public void testExceptionOnRecover()
    {
        AMQSession_0_10 session = createThrowingExceptionAMQSession_0_10(javax.jms.Session.AUTO_ACKNOWLEDGE);
        try
        {
            session.recover();
            fail("JMSException should be thrown");
        }
        catch (Exception e)
        {
            assertTrue("JMSException is expected", e instanceof JMSException);
        }
    }

    public void testExceptionOnCreateBrowser()
    {
        AMQSession_0_10 session = createThrowingExceptionAMQSession_0_10();
        AMQQueue destination = createQueue();
        try
        {
            session.createBrowser(destination);
            fail("JMSException should be thrown");
        }
        catch (Exception e)
        {
            assertTrue("JMSException is expected", e instanceof JMSException);
            assertEquals("541 error code is expected", "541", ((JMSException) e).getErrorCode());
        }
    }

    public void testExceptionOnCreateConsumer()
    {
        AMQSession_0_10 session = createThrowingExceptionAMQSession_0_10();
        AMQAnyDestination destination = createDestination();
        try
        {
            session.createConsumer(destination);
            fail("JMSException should be thrown");
        }
        catch (Exception e)
        {
            assertTrue("JMSException is expected", e instanceof JMSException);
            assertEquals("541 error code is expected", "541", ((JMSException) e).getErrorCode());
        }
    }

    public void testExceptionOnCreateSubscriber()
    {
        AMQSession_0_10 session = createThrowingExceptionAMQSession_0_10();
        AMQAnyDestination destination = createDestination();
        try
        {
            session.createSubscriber(destination);
            fail("JMSException should be thrown");
        }
        catch (Exception e)
        {
            assertTrue("JMSException is expected", e instanceof JMSException);
            assertEquals("541 error code is expected", "541", ((JMSException) e).getErrorCode());
        }
    }

    public void testExceptionOnUnsubscribe()
    {
        AMQSession_0_10 session = createThrowingExceptionAMQSession_0_10();
        try
        {
            session.unsubscribe("whatever");
            fail("JMSExceptiuon should be thrown");
        }
        catch (Exception e)
        {
            assertTrue("JMSException is expected", e instanceof JMSException);
            assertEquals("541 error code is expected", "541", ((JMSException) e).getErrorCode());
        }
    }

    public void testCommit()
    {
        AMQSession_0_10 session = createAMQSession_0_10();
        try
        {
            session.commit();
        }
        catch (Exception e)
        {
            fail("Unexpected exception is caught:" + e.getMessage());
        }
        ProtocolEvent event = findSentProtocolEventOfClass(session, TxCommit.class, false);
        assertNotNull("TxCommit was not sent", event);
    }

    public void testRollback()
    {
        AMQSession_0_10 session = createAMQSession_0_10();
        try
        {
            session.rollback();
        }
        catch (Exception e)
        {
            fail("Unexpected exception is caught:" + e.getMessage());
        }
        ProtocolEvent event = findSentProtocolEventOfClass(session, TxRollback.class, false);
        assertNotNull("TxRollback was not sent", event);
    }

    public void testRecover()
    {
        AMQSession_0_10 session = createAMQSession_0_10(javax.jms.Session.AUTO_ACKNOWLEDGE);
        try
        {
            session.recover();
        }
        catch (Exception e)
        {
            fail("Unexpected exception is caught:" + e.getMessage());
        }
        ProtocolEvent event = findSentProtocolEventOfClass(session, MessageRelease.class, false);
        assertNotNull("MessageRelease was not sent", event);
    }

    public void testCreateProducer()
    {
        AMQSession_0_10 session = createAMQSession_0_10();
        try
        {
            session.createProducer(createQueue());
        }
        catch (Exception e)
        {
            fail("Unexpected exception is caught:" + e.getMessage());
        }
        ProtocolEvent event = findSentProtocolEventOfClass(session, ExchangeDeclare.class, false);
        assertNotNull("ExchangeDeclare was not sent", event);
    }

    public void testCreateConsumer()
    {
        AMQSession_0_10 session = createAMQSession_0_10();
        try
        {
            session.createConsumer(createQueue());
        }
        catch (Exception e)
        {
            fail("Unexpected exception is caught:" + e.getMessage());
        }
        ProtocolEvent event = findSentProtocolEventOfClass(session, MessageSubscribe.class, false);
        assertNotNull("MessageSubscribe was not sent", event);
    }

    public void testSync()
    {
        AMQSession_0_10 session = createAMQSession_0_10();
        try
        {
            session.sync();
        }
        catch (Exception e)
        {
            fail("Unexpected exception is caught:" + e.getMessage());
        }
        ProtocolEvent event = findSentProtocolEventOfClass(session, ExecutionSync.class, false);
        assertNotNull("ExecutionSync was not sent", event);
    }

    public void testSendQueueDelete()
    {
        AMQSession_0_10 session = createAMQSession_0_10();
        try
        {
            session.sendQueueDelete("test");
        }
        catch (Exception e)
        {
            fail("Unexpected exception is caught:" + e.getMessage());
        }
        ProtocolEvent event = findSentProtocolEventOfClass(session, QueueDelete.class, false);
        assertNotNull("QueueDelete event was not sent", event);
        QueueDelete exchangeDelete = (QueueDelete) event;
        assertEquals("test", exchangeDelete.getQueue());
    }

    public void testSendConsume()
    {
        AMQSession_0_10 session = createAMQSession_0_10();
        try
        {
            BasicMessageConsumer_0_10 consumer = session.createMessageConsumer(createDestination(), 1, 1, true, false,
                    null, null, false, true);
            session.sendConsume(consumer, "test", true);
        }
        catch (Exception e)
        {
            fail("Unexpected exception is caught:" + e.getMessage());
        }
        ProtocolEvent event = findSentProtocolEventOfClass(session, MessageSubscribe.class, false);
        assertNotNull("MessageSubscribe event was not sent", event);
    }

    public void testCreateMessageProducer()
    {
        AMQSession_0_10 session = createAMQSession_0_10();
        try
        {
            session.createMessageProducer(createDestination(), true, true, 1l);
        }
        catch (Exception e)
        {
            fail("Unexpected exception is caught:" + e.getMessage());
        }
        ProtocolEvent event = findSentProtocolEventOfClass(session, ExchangeDeclare.class, false);
        assertNotNull("ExchangeDeclare event was not sent", event);
    }

    public void testSendExchangeDelete()
    {
        AMQSession_0_10 session = createAMQSession_0_10();
        try
        {
            session.sendExchangeDelete("test", true);
        }
        catch (Exception e)
        {
            fail("Unexpected exception is caught:" + e.getMessage());
        }
        ProtocolEvent event = findSentProtocolEventOfClass(session, ExchangeDelete.class, false);
        assertNotNull("ExchangeDelete event was not sent", event);
        ExchangeDelete exchangeDelete = (ExchangeDelete) event;
        assertEquals("test", exchangeDelete.getExchange());
    }

    public void testExceptionOnMessageConsumerReceive()
    {
        AMQSession_0_10 session = createThrowingExceptionAMQSession_0_10();
        try
        {
            BasicMessageConsumer_0_10 consumer = session.createMessageConsumer(createDestination(), 1, 1, true, false,
                    null, null, false, true);
            session.start();
            consumer.receive(1);
            fail("JMSException should be thrown");
        }
        catch (Exception e)
        {
            assertTrue("JMSException is expected", e instanceof JMSException);
            assertEquals("541 error code is expected", "541", ((JMSException) e).getErrorCode());
        }
    }

    public void testMessageConsumerReceive()
    {
        AMQSession_0_10 session = createAMQSession_0_10();
        try
        {
            BasicMessageConsumer_0_10 consumer = session.createMessageConsumer(createDestination(), 1, 1, true, false,
                    null, null, false, true);
            session.start();
            consumer.receive(1);
        }
        catch (Exception e)
        {
            fail("Unexpected exception is caught:" + e.getMessage());
        }
        ProtocolEvent event = findSentProtocolEventOfClass(session, MessageFlow.class, false);
        assertNotNull("MessageFlow event was not sent", event);
    }

    public void testExceptionOnMessageConsumerReceiveNoWait()
    {
        AMQSession_0_10 session = createThrowingExceptionAMQSession_0_10();
        try
        {
            BasicMessageConsumer_0_10 consumer = session.createMessageConsumer(createDestination(), 1, 1, true, false,
                    null, null, false, true);
            session.start();
            consumer.receiveNoWait();
            fail("JMSException should be thrown");
        }
        catch (Exception e)
        {
            assertTrue("JMSException is expected", e instanceof JMSException);
            assertEquals("541 error code is expected", "541", ((JMSException) e).getErrorCode());
        }
    }

    public void testMessageConsumerClose()
    {
        AMQSession_0_10 session = createAMQSession_0_10();
        try
        {
            BasicMessageConsumer_0_10 consumer = session.createMessageConsumer(createDestination(), 1, 1, true, false,
                    null, null, false, true);
            consumer.setConsumerTag("");
            consumer.close();
        }
        catch (Exception e)
        {
            fail("Unexpected exception is caught:" + e.getMessage());
        }
        ProtocolEvent event = findSentProtocolEventOfClass(session, MessageCancel.class, false);
        assertNotNull("MessageCancel event was not sent", event);
    }

    public void testExceptionOnMessageConsumerClose()
    {
        AMQSession_0_10 session = createThrowingExceptionAMQSession_0_10();
        try
        {
            BasicMessageConsumer_0_10 consumer = session.createMessageConsumer(createDestination(), 1, 1, true, false,
                    null, null, false, true);
            consumer.close();
            fail("JMSException should be thrown");
        }
        catch (Exception e)
        {
            assertTrue("JMSException is expected", e instanceof JMSException);
            assertEquals("541 error code is expected", "541", ((JMSException) e).getErrorCode());
        }
    }

    public void testMessageProducerSend()
    {
        AMQSession_0_10 session = createAMQSession_0_10();
        try
        {
            MessageProducer producer = session.createProducer(createQueue());
            producer.send(session.createTextMessage("Test"));
            session.commit();
        }
        catch (Exception e)
        {
            fail("Unexpected exception is caught:" + e.getMessage());
        }
        ProtocolEvent event = findSentProtocolEventOfClass(session, MessageTransfer.class, false);
        assertNotNull("MessageTransfer event was not sent", event);
        event = findSentProtocolEventOfClass(session, ExchangeDeclare.class, false);
        assertNotNull("ExchangeDeclare event was not sent", event);
    }

    public void testCreateStreamMessage() throws Exception
    {
        AMQSession_0_10 session = createAMQSession_0_10();
        StreamMessage m = session.createStreamMessage();
        assertTrue("Legacy Stream message encoding should be the default" + m.getClass(),!(m instanceof AMQPEncodedListMessage));
    }

    public void testGetQueueDepthWithSync()
    {
        // slow down a flush thread
        setTestSystemProperty("qpid.session.max_ack_delay", "10000");
        AMQSession_0_10 session =  createAMQSession_0_10(false, javax.jms.Session.DUPS_OK_ACKNOWLEDGE);
        try
        {
            session.acknowledgeMessage(-1, false);
            session.getQueueDepth(createDestination(), true);
        }
        catch (Exception e)
        {
            fail("Unexpected exception is caught:" + e.getMessage());
        }
        ProtocolEvent command = findSentProtocolEventOfClass(session, MessageAccept.class, false);
        assertNotNull("MessageAccept command was not sent", command);
        command = findSentProtocolEventOfClass(session, ExecutionSync.class, false);
        assertNotNull("ExecutionSync command was not sent", command);
        command = findSentProtocolEventOfClass(session, QueueQuery.class, false);
        assertNotNull("QueueQuery command was not sent", command);
    }

    public void testResubscribe() throws Exception
    {
        AMQSession_0_10 session = createAMQSession_0_10(AMQSession_0_10.AUTO_ACKNOWLEDGE);

        AMQQueue queue1 = new AMQQueue(new AMQBindingURL("direct://amq.direct//test1?routingkey='test1'&durable='true'"));
        session.createProducer(queue1);
        BasicMessageConsumer_0_10 consumer1 = (BasicMessageConsumer_0_10)session.createConsumer(queue1);

        AMQQueue queue2 = new AMQQueue(new AMQBindingURL("direct://amq.direct//test2?routingkey='test2'"));
        session.createProducer(queue2);
        BasicMessageConsumer_0_10 consumer2 = (BasicMessageConsumer_0_10)session.createConsumer(queue2);

        UnprocessedMessage[] messages = new UnprocessedMessage[4];
        for (int i =0; i< messages.length;i++ )
        {
            String consumerTag = i % 2 == 0 ? consumer1.getConsumerTag() : consumer2.getConsumerTag();
            int deliveryTag = i + 1;
            messages[i]= createMockMessage(deliveryTag, consumerTag);
            session.messageReceived(messages[i]);
            if (deliveryTag % 2 == 0)
            {
                session.addUnacknowledgedMessage(deliveryTag);
            }
        }

        assertEquals("Unexpected highest delivery tag", 4, session.getHighestDeliveryTag().get());
        assertFalse("Unexpected unacknowledged message tags", session.getUnacknowledgedMessageTags().isEmpty());
        assertEquals("Unexpected consumers", new HashSet<>(Arrays.asList(consumer1, consumer2)), new HashSet<>(session.getConsumers()));

        // verify test messages were not dispatched
        for (UnprocessedMessage message: messages )
        {
            verify(message, never()).dispatch(session);
        }

        session.resubscribe();

        assertEquals("Unexpected highest delivery tag", -1, session.getHighestDeliveryTag().get());
        assertTrue("Unexpected unacknowledged message tags", session.getUnacknowledgedMessageTags().isEmpty());
        assertTrue("Unexpected pre-fetched message tags", session.getPrefetchedMessageTags().isEmpty());
        assertEquals("Unexpected consumers", new HashSet<>(Arrays.asList(consumer1, consumer2)), new HashSet<>(session.getConsumers()));
    }

    public void testFailoverPrep() throws Exception
    {
        AMQSession_0_10 session = createAMQSession_0_10(AMQSession_0_10.AUTO_ACKNOWLEDGE);

        UnprocessedMessage[] messages = new UnprocessedMessage[4];
        for (int i =0; i< messages.length;i++ )
        {
            String consumerTag = String.valueOf(i % 2);
            int deliveryTag = i + 1;
            messages[i]= createMockMessage(deliveryTag, consumerTag);
            session.messageReceived(messages[i]);
            if (deliveryTag % 2 == 0)
            {
                session.addUnacknowledgedMessage(deliveryTag);
            }
        }

        // verify test messages were not dispatched
        for (UnprocessedMessage message: messages )
        {
            verify(message, never()).dispatch(session);
        }

        session.failoverPrep();

        // verify dispatcher queue is drained
        for (UnprocessedMessage message: messages )
        {
            verify(message).dispatch(session);
        }
    }

    private UnprocessedMessage createMockMessage(long deliveryTag, String consumerTag)
    {
        UnprocessedMessage message = mock(UnprocessedMessage.class);
        when(message.getConsumerTag()).thenReturn(consumerTag);
        when(message.getDeliveryTag()).thenReturn(deliveryTag);
        return message;
    }


    private AMQAnyDestination createDestination()
    {
        AMQAnyDestination destination = null;
        try
        {
            destination = new AMQAnyDestination("amq.direct", "direct",
                    "test", false, true, "test", true, null);
        }
        catch (Exception e)
        {
            fail("Failued to create destination:" + e.getMessage());
        }
        return destination;
    }

    private AMQQueue createQueue()
    {
        AMQQueue destination = null;
        try
        {
            destination = new AMQQueue("amq.direct", "test", "test");
        }
        catch (Exception e)
        {
            fail("Failed to create destination:" + e.getMessage());
        }
        return destination;
    }

    private AMQSession_0_10 createThrowingExceptionAMQSession_0_10()
    {
        return createAMQSession_0_10(true, javax.jms.Session.SESSION_TRANSACTED);
    }

    private AMQSession_0_10 createThrowingExceptionAMQSession_0_10(int akcnowledgeMode)
    {
        return createAMQSession_0_10(true, akcnowledgeMode);
    }

    private ProtocolEvent findSentProtocolEventOfClass(AMQSession_0_10 session, Class<? extends ProtocolEvent> class1,
            boolean isLast)
    {
        ProtocolEvent found = null;
        List<ProtocolEvent> events = ((MockSession) session.getQpidSession()).getSender().getSendEvents();
        assertNotNull("Events list should not be null", events);
        assertFalse("Events list should not be empty", events.isEmpty());
        if (isLast)
        {
            ProtocolEvent event = events.get(events.size() - 1);
            if (event.getClass().isAssignableFrom(class1))
            {
                found = event;
            }
        }
        else
        {
            for (ProtocolEvent protocolEvent : events)
            {
                if (protocolEvent.getClass().isAssignableFrom(class1))
                {
                    found = protocolEvent;
                    break;
                }
            }

        }
        return found;
    }

    private AMQSession_0_10 createAMQSession_0_10()
    {
        return createAMQSession_0_10(false, javax.jms.Session.SESSION_TRANSACTED);
    }

    private AMQSession_0_10 createAMQSession_0_10(int acknowledgeMode)
    {
        return createAMQSession_0_10(false, acknowledgeMode);
    }

    private AMQSession_0_10 createAMQSession_0_10(boolean throwException, int acknowledgeMode)
    {
        AMQConnection amqConnection = null;
        try
        {
            amqConnection = new MockAMQConnection(
                    "amqp://guest:guest@client/test?brokerlist='tcp://localhost:1'&maxprefetch='0'");
        }
        catch (Exception e)
        {
            fail("Failure to create a mock connection:" + e.getMessage());
        }
        boolean isTransacted = acknowledgeMode == javax.jms.Session.SESSION_TRANSACTED ? true : false;
        AMQSession_0_10 session = new AMQSession_0_10(createConnection(throwException), amqConnection, 1, isTransacted, acknowledgeMode,
                 10, 10, "test");
        return session;
    }

    private Connection createConnection(final boolean throwException)
    {
        MockTransportConnection connection = new MockTransportConnection();
        connection.setState(State.OPEN);
        connection.setSender(new MockSender());
        connection.setSessionFactory(new SessionFactory()
        {

            public Session newSession(Connection conn, Binary name, long expiry, boolean isNoReplay)
            {
                return new MockSession(conn, new SessionDelegate(), name, expiry, throwException);
            }
        });
        return connection;
    }

    private final class MockMessageListener implements MessageListener
    {
        public void onMessage(Message arg0)
        {
        }
    }

    class MockSession extends Session
    {
        private final boolean _throwException;
        private final Connection _connection;
        private final SessionDelegate _delegate;

        protected MockSession(Connection connection, SessionDelegate delegate, Binary name, long expiry,
                boolean throwException)
        {
            super(connection, delegate, name, expiry);
            _throwException = throwException;
            setState(State.OPEN);
            _connection = connection;
            _delegate = delegate;
        }

        public void invoke(Method m, Runnable postIdSettingAction)
        {
            if (_throwException)
            {
                if (m instanceof SessionAttach || m instanceof SessionRequestTimeout || m instanceof TxSelect)
                {
                    // do not throw exception for SessionAttach,
                    // SessionRequestTimeout and TxSelect
                    // session needs to be instantiated
                    return;
                }
                ExecutionException e = new ExecutionException();
                e.setErrorCode(ExecutionErrorCode.INTERNAL_ERROR);
                throw new SessionException(e);
            }
            else
            {
                super.invoke(m, postIdSettingAction);
                if (m instanceof SessionDetach)
                {
                    setState(State.CLOSED);
                }
            }
        }

        public void sync()
        {
            // to avoid recursive calls
            setAutoSync(false);
            // simply send sync command
            super.executionSync(Option.SYNC);
        }

        protected <T> Future<T> invoke(Method m, Class<T> klass)
        {
            int commandId = getCommandsOut();
            Future<T> future = super.invoke(m, klass);
            ExecutionResult result = new ExecutionResult();
            result.setCommandId(commandId);
            if (m instanceof ExchangeBound)
            {
                ExchangeBoundResult struc = new ExchangeBoundResult();
                result.setValue(struc);
            }
            else if (m instanceof ExchangeQuery)
            {
                ExchangeQueryResult struc = new ExchangeQueryResult();
                result.setValue(struc);
            }
            else if (m instanceof QueueQuery)
            {
                QueueQueryResult struc = new QueueQueryResult();
                result.setValue(struc);
            }
            _delegate.executionResult(this, result);
            return future;
        }

        public MockSender getSender()
        {
            return (MockSender) _connection.getSender();
        }
    }

    class MockTransportConnection extends Connection
    {
        public void setState(State state)
        {
            super.setState(state);
        }
    }

    class MockSender implements ProtocolEventSender
    {
        private List<ProtocolEvent> _sendEvents = new ArrayList<ProtocolEvent>();

        private void setIdleTimeout(int i)
        {
        }

        public void send(ProtocolEvent msg)
        {
            _sendEvents.add(msg);
        }

        public void flush()
        {
        }

        public void close()
        {
        }

        public List<ProtocolEvent> getSendEvents()
        {
            return _sendEvents;
        }

    }

}