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

import static org.junit.Assert.assertArrayEquals;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertTrue;

import java.io.IOException;
import java.net.ServerSocket;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.List;

import junit.framework.AssertionFailedError;

import org.apache.qpid.proton.Proton;
import org.apache.qpid.proton.engine.BaseHandler;
import org.apache.qpid.proton.engine.Connection;
import org.apache.qpid.proton.engine.Delivery;
import org.apache.qpid.proton.engine.Event;
import org.apache.qpid.proton.engine.Event.Type;
import org.apache.qpid.proton.engine.Handler;
import org.apache.qpid.proton.engine.HandlerException;
import org.apache.qpid.proton.engine.Sender;
import org.apache.qpid.proton.engine.Session;
import org.apache.qpid.proton.reactor.impl.AcceptorImpl;
import org.apache.qpid.proton.reactor.impl.LeakTestReactor;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
import org.junit.runners.Parameterized.Parameters;

@RunWith(Parameterized.class)
public class ReactorTest {

    public ReactorFactory reactorFactory;
    private Reactor reactor;

    private static interface ReactorFactory {
        Reactor newReactor() throws IOException;
    }

    // Parameterize the tests, and run them once with a reactor obtained by calling
    // 'Proton.reactor()' and once with the LeakTestReactor.
    @Parameters
    public static Collection<ReactorFactory[]> data() throws IOException {
        ReactorFactory classicReactor = new ReactorFactory() {
            @Override public Reactor newReactor() throws IOException {
                return Proton.reactor();
            }
        };
        ReactorFactory newLeakDetection = new ReactorFactory() {
            @Override public Reactor newReactor() throws IOException {
                return new LeakTestReactor();
            }
        };
        return Arrays.asList(new ReactorFactory[][]{{classicReactor}, {newLeakDetection}});
    }

    public ReactorTest(ReactorFactory reactorFactory) {
        this.reactorFactory = reactorFactory;
    }

    @Before
    public void before() throws IOException {
        reactor = reactorFactory.newReactor();
    }

    private void checkForLeaks() {
        if (reactor instanceof LeakTestReactor) {
            ((LeakTestReactor)reactor).assertNoLeaks();
        }
    }

    @After
    public void after() {
        checkForLeaks();
    }

    /**
     * Tests that creating a reactor and running it:
     * <ul>
     *   <li>Doesn't throw any exceptions.</li>
     *   <li>Returns immediately from the run method (as there is no more work to do).</li>
     * </ul>
     * @throws IOException
     */
    @Test
    public void runEmpty() throws IOException {
        assertNotNull(reactor);
        reactor.run();
        reactor.free();
    }

    private static class TestHandler extends BaseHandler {
        private final ArrayList<Type> actual = new ArrayList<Type>();

        @Override
        public void onUnhandled(Event event) {
            assertNotNull(event.getReactor());
            actual.add(event.getType());
        }

        public void assertEvents(Type...expected) {
            assertArrayEquals(expected, actual.toArray());
        }
    }

    /**
     * Tests adding a handler to a reactor and running the reactor.  The
     * expected behaviour is for the reactor to return, and a number of reactor-
     * related events to have been delivered to the handler.
     * @throws IOException
     */
    @Test
    public void handlerRun() throws IOException {
        Handler handler = reactor.getHandler();
        assertNotNull(handler);
        TestHandler testHandler = new TestHandler();
        handler.add(testHandler);
        reactor.run();
        reactor.free();
        testHandler.assertEvents(Type.REACTOR_INIT, Type.SELECTABLE_INIT, Type.SELECTABLE_UPDATED, Type.SELECTABLE_FINAL, Type.REACTOR_FINAL);
    }

    /**
     * Tests basic operation of the Reactor.connection method by creating a
     * connection from a reactor, then running the reactor.  The expected behaviour
     * is for:
     * <ul>
     *   <li>The reactor to end immediately.</li>
     *   <li>The handler associated with the connection receives an init event.</li>
     *   <li>The connection is one of the reactor's children.</li>
     * </ul>
     * @throws IOException
     */
    @Test
    public void connection() throws IOException {
        TestHandler connectionHandler = new TestHandler();
        Connection connection = reactor.connection(connectionHandler);
        assertNotNull(connection);
        assertTrue("connection should be one of the reactor's children", reactor.children().contains(connection));
        reactor.setConnectionHost(connection, "127.0.0.1", 5672);
        assertEquals("connection address configuration failed",
                     reactor.getConnectionAddress(connection), "127.0.0.1:5672");
        TestHandler reactorHandler = new TestHandler();
        reactor.getHandler().add(reactorHandler);
        reactor.run();
        reactor.free();
        reactorHandler.assertEvents(Type.REACTOR_INIT, Type.SELECTABLE_INIT, Type.SELECTABLE_UPDATED, Type.SELECTABLE_FINAL, Type.REACTOR_FINAL);
        connectionHandler.assertEvents(Type.CONNECTION_INIT);
    }

    /**
     * Tests operation of the Reactor.acceptor method by creating an acceptor
     * which is immediately closed by the reactor.  The expected behaviour is for:
     * <ul>
     *   <li>The reactor to end immediately (as it has no more work to process).</li>
     *   <li>The handler, associated with the acceptor, to receive no events.</li>
     *   <li>For it's lifetime, the acceptor is one of the reactor's children.</li>
     * </ul>
     * @throws IOException
     */
    @Test
    public void acceptor() throws IOException {
        final Acceptor acceptor = reactor.acceptor("127.0.0.1", 0);
        assertNotNull(acceptor);
        assertTrue("acceptor should be one of the reactor's children", reactor.children().contains(acceptor));
        TestHandler acceptorHandler = new TestHandler();
        BaseHandler.setHandler(acceptor, acceptorHandler);
        reactor.getHandler().add(new BaseHandler() {
            @Override
            public void onReactorInit(Event event) {
                acceptor.close();
            }
        });
        reactor.run();
        reactor.free();
        acceptorHandler.assertEvents();
        assertFalse("acceptor should have been removed from the reactor's children", reactor.children().contains(acceptor));
    }

    private static class ServerHandler extends TestHandler {
        private Acceptor acceptor;
        public void setAcceptor(Acceptor acceptor) {
            this.acceptor = acceptor;
        }
        @Override
        public void onConnectionRemoteOpen(Event event) {
            super.onConnectionRemoteOpen(event);
            event.getConnection().open();
        }
        @Override
        public void onConnectionRemoteClose(Event event) {
            super.onConnectionRemoteClose(event);
            acceptor.close();
            event.getConnection().close();
            event.getConnection().free();
        }
    }

    /**
     * Tests end to end behaviour of the reactor by creating an acceptor and then
     * a connection (which connects to the port the acceptor is listening on).
     * As soon as the connection is established, both the acceptor and connection
     * are closed.  The events generated by the acceptor and the connection are
     * compared to a set of expected events.
     * @throws IOException
     */
    @Test
    public void connect() throws IOException {
        ServerHandler sh = new ServerHandler();
        Acceptor acceptor = reactor.acceptor("127.0.0.1",  0, sh);
        final int listeningPort = ((AcceptorImpl)acceptor).getPortNumber();
        sh.setAcceptor(acceptor);

        class ClientHandler extends TestHandler {
            @Override
            public void onConnectionInit(Event event) {
                super.onConnectionInit(event);
                event.getReactor().setConnectionHost(event.getConnection(),
                                                     "127.0.0.1",
                                                     listeningPort);
                event.getConnection().open();
            }
            @Override
            public void onConnectionRemoteOpen(Event event) {
                super.onConnectionRemoteOpen(event);
                event.getConnection().close();
            }
            @Override
            public void onConnectionRemoteClose(Event event) {
                super.onConnectionRemoteClose(event);
                event.getConnection().free();
            }
        }
        ClientHandler ch = new ClientHandler();
        Connection connection = reactor.connection(ch);

        assertTrue("acceptor should be one of the reactor's children", reactor.children().contains(acceptor));
        assertTrue("connection should be one of the reactor's children", reactor.children().contains(connection));

        reactor.run();
        reactor.free();

        assertFalse("acceptor should have been removed from the reactor's children", reactor.children().contains(acceptor));
        assertFalse("connection should have been removed from the reactor's children", reactor.children().contains(connection));
        sh.assertEvents(Type.CONNECTION_INIT, Type.CONNECTION_BOUND,
                // XXX: proton-c generates a PN_TRANSPORT event here
                Type.CONNECTION_REMOTE_OPEN, Type.CONNECTION_LOCAL_OPEN,
                Type.TRANSPORT, Type.CONNECTION_REMOTE_CLOSE,
                Type.TRANSPORT_TAIL_CLOSED, Type.CONNECTION_LOCAL_CLOSE,
                Type.TRANSPORT, Type.TRANSPORT_HEAD_CLOSED,
                Type.TRANSPORT_CLOSED, Type.CONNECTION_UNBOUND,
                Type.CONNECTION_FINAL);

        ch.assertEvents(Type.CONNECTION_INIT, Type.CONNECTION_LOCAL_OPEN,
                Type.CONNECTION_BOUND,
                // XXX: proton-c generates two PN_TRANSPORT events here
                Type.CONNECTION_REMOTE_OPEN, Type.CONNECTION_LOCAL_CLOSE,
                Type.TRANSPORT, Type.TRANSPORT_HEAD_CLOSED,
                Type.CONNECTION_REMOTE_CLOSE, Type.TRANSPORT_TAIL_CLOSED,
                Type.TRANSPORT_CLOSED, Type.CONNECTION_UNBOUND,
                Type.CONNECTION_FINAL);

    }

    private String checkVhost(String vhost) throws IOException {

        class ServerVhostHandler extends ServerHandler {
            public String peerVhost;

            @Override
            public void onConnectionRemoteOpen(Event event) {
                super.onConnectionRemoteOpen(event);
                peerVhost = event.getConnection().getRemoteHostname();
            }
        }

        class ClientVhostHandler extends TestHandler {
            private int port;
            private String vhost;

            ClientVhostHandler(String vhost, int port) {
                this.port = port;
                this.vhost = vhost;
            }

            @Override
            public void onConnectionInit(Event event) {
                super.onConnectionInit(event);
                event.getReactor().setConnectionHost(event.getConnection(),
                                                     "127.0.0.1", port);
                if (vhost != null) {
                    event.getConnection().setHostname(vhost);
                }
                event.getConnection().open();
            }
            @Override
            public void onConnectionRemoteOpen(Event event) {
                super.onConnectionRemoteOpen(event);
                event.getConnection().close();
            }
            @Override
            public void onConnectionRemoteClose(Event event) {
                super.onConnectionRemoteClose(event);
                event.getConnection().free();
            }
        }
        ServerVhostHandler sh = new ServerVhostHandler();
        Acceptor acceptor = reactor.acceptor("127.0.0.1",  0, sh);
        final int listeningPort = ((AcceptorImpl)acceptor).getPortNumber();
        sh.setAcceptor(acceptor);

        ClientVhostHandler ch = new ClientVhostHandler(vhost, listeningPort);
        Connection connection = reactor.connection(ch);

        reactor.run();
        reactor.free();
        checkForLeaks();

        return sh.peerVhost;
    }

    /**
     * Tests the virtual host default configuration - should be set to host
     * used for the connection.
     * @throws IOException
     **/
    @Test
    public void checkVhostDefault() throws IOException {
        String vhost = checkVhost(null);
        assertEquals("The default virtual host is not correct",
                     "127.0.0.1", vhost);
    }

    /**
     * Tests the virtual host override - should be set to connection's
     * hostname.
     * @throws IOException
     **/
    @Test
    public void checkVhostOverride() throws IOException {
        String vhost = checkVhost("my.vhost");
        assertEquals("The virtual host is not correct",
                     "my.vhost", vhost);
    }

    /**
     * Tests eliminating the virtual host configuration - expects no vhost for
     * the connection.
     * @throws IOException
     **/
    @Test
    public void checkNoVhost() throws IOException {
        String vhost = checkVhost("");
        assertEquals("The virtual host is present",
                     null, vhost);
    }


    private static class SinkHandler extends BaseHandler {
        protected int received = 0;

        @Override
        public void onDelivery(Event event) {
            Delivery dlv = event.getDelivery();
            if (!dlv.isPartial()) {
                dlv.settle();
                ++received;
            }
        }
    }

    private static class SourceHandler extends BaseHandler {
        private int remaining;

        protected SourceHandler(int count) {
            remaining = count;
        }

        @Override
        public void onConnectionInit(Event event) {
            Connection conn = event.getConnection();
            Session ssn = conn.session();
            Sender snd = ssn.sender("sender");
            conn.open();
            ssn.open();
            snd.open();
        }

        @Override
        public void onLinkFlow(Event event) {
            Sender link = (Sender)event.getLink();
            while (link.getCredit() > 0 && remaining > 0) {
                Delivery dlv = link.delivery(new byte[0]);
                assertNotNull(dlv);
                dlv.settle();
                link.advance();
                --remaining;
            }

            if (remaining == 0) {
                event.getConnection().close();
            }
        }

        @Override
        public void onConnectionRemoteClose(Event event) {
            event.getConnection().free();
        }
    }

    private void transfer(int count, int window) throws IOException {
        reactor = reactorFactory.newReactor();
        ServerHandler sh = new ServerHandler();
        Acceptor acceptor = reactor.acceptor("127.0.0.1", 0, sh);
        sh.setAcceptor(acceptor);
        sh.add(new Handshaker());
        // XXX: a window of 1 doesn't work unless the flowcontroller is
        // added after the thing that settles the delivery
        sh.add(new FlowController(window));
        SinkHandler snk = new SinkHandler();
        sh.add(snk);

        SourceHandler src = new SourceHandler(count);
        reactor.connectionToHost("127.0.0.1", ((AcceptorImpl)acceptor).getPortNumber(),
                                 src);
        reactor.run();
        reactor.free();
        assertEquals("Did not receive the expected number of messages", count, snk.received);
        checkForLeaks();
    }

    @Test
    public void transfer_0to64_2() throws IOException {
        for (int i = 0; i < 64; ++i) {
            transfer(i, 2);
        }
    }

    @Test
    public void transfer_1024_64() throws IOException {
        transfer(1024, 64);
    }

    @Test
    public void transfer_4096_1024() throws IOException {
        transfer(4*1024, 1024);
    }

    @Test
    public void schedule() throws IOException {
        TestHandler reactorHandler = new TestHandler();
        reactor.getHandler().add(reactorHandler);
        TestHandler taskHandler = new TestHandler();
        reactor.schedule(0, taskHandler);
        reactor.run();
        reactor.free();
        reactorHandler.assertEvents(Type.REACTOR_INIT, Type.SELECTABLE_INIT, Type.SELECTABLE_UPDATED, Type.REACTOR_QUIESCED, Type.SELECTABLE_UPDATED,
                Type.SELECTABLE_FINAL, Type.REACTOR_FINAL);
        taskHandler.assertEvents(Type.TIMER_TASK);
    }

    @Test
    public void scheduleWithEqualDeadline() throws IOException {
        final int count = 10;
        final ArrayList<Integer> taskRunOrder = new ArrayList<Integer>();

        class TaskHandler extends BaseHandler {
            private int _counter;

            private TaskHandler(int counter) {
                _counter = counter;
            }

            @Override
            public void onTimerTask(Event event) {
                taskRunOrder.add(_counter);
            }
        }

        final List<Integer> expectedOrder = new ArrayList<>();
        for(int i = 0 ; i < count; i++) {
            reactor.schedule(0, new TaskHandler(i));
            expectedOrder.add(i);
        }

        reactor.run();
        reactor.free();

        assertEquals(expectedOrder, taskRunOrder);
    }

    private class BarfException extends RuntimeException {
        private static final long serialVersionUID = -5891140258375562884L;
    }

    private class BarfOnSomethingHandler extends BaseHandler {
        protected final BarfException exception;

        protected BarfOnSomethingHandler(BarfException exception) {
            this.exception = exception;
        }
    }

    private class BarfOnReactorInit extends BarfOnSomethingHandler {

        protected BarfOnReactorInit(BarfException exception) {
            super(exception);
        }

        @Override
        public void onReactorInit(Event e) {
            throw exception;
        }
    }

    private class BarfOnReactorFinal extends BarfOnSomethingHandler {

        protected BarfOnReactorFinal(BarfException exception) {
            super(exception);
        }

        @Override
        public void onReactorFinal(Event event) {
            throw exception;
        }
    }

    private class BarfOnConnectionInit extends BarfOnSomethingHandler {

        protected BarfOnConnectionInit(BarfException exception) {
            super(exception);
        }

        @Override
        public void onConnectionInit(Event e) {
            throw exception;
        }
    }

    private class BarfOnSessionInit extends BarfOnSomethingHandler {

        protected BarfOnSessionInit(BarfException exception) {
            super(exception);
        }

        @Override
        public void onSessionInit(Event e) {
            throw exception;
        }
    }

    private class BarfOnLinkInit extends BarfOnSomethingHandler {

        protected BarfOnLinkInit(BarfException exception) {
            super(exception);
        }

        @Override
        public void onLinkInit(Event e) {
            throw exception;
        }
    }

    private class BarfOnTask extends BarfOnSomethingHandler {

        protected BarfOnTask(BarfException exception) {
            super(exception);
        }

        @Override
        public void onTimerTask(Event e) {
            throw exception;
        }
    }

    private void assertReactorRunBarfsOnHandler(Reactor reactor, BarfException expectedException, Handler expectedHandler) {
        try {
            reactor.run();
            throw new AssertionFailedError("Reactor.run() should have thrown an exception");
        } catch(HandlerException handlerException) {
            assertSame("Linked exception does not match expected exception", expectedException, handlerException.getCause());
            assertSame("Handler in exception does not match expected handler", expectedHandler, handlerException.getHandler());
        }
    }

    @Test
    public void barfInReactorFinal() throws IOException {
        BarfException expectedBarf = new BarfException();
        Handler expectedHandler = new BarfOnReactorFinal(expectedBarf);
        reactor.getGlobalHandler().add(expectedHandler);
        assertReactorRunBarfsOnHandler(reactor, expectedBarf, expectedHandler);
        reactor.free();
    }

    @Test
    public void barfOnGlobalSet() throws IOException {
        BarfException expectedBarf = new BarfException();
        Handler expectedHandler = new BarfOnReactorInit(expectedBarf);
        reactor.setGlobalHandler(expectedHandler);
        assertReactorRunBarfsOnHandler(reactor, expectedBarf, expectedHandler);
        reactor.free();
    }

    @Test
    public void barfOnGlobalAdd() throws IOException {
        BarfException expectedBarf = new BarfException();
        Handler expectedHandler = new BarfOnReactorInit(expectedBarf);
        reactor.getGlobalHandler().add(expectedHandler);
        assertReactorRunBarfsOnHandler(reactor, expectedBarf, expectedHandler);
        reactor.free();
    }

    @Test
    public void barfOnReactorSet() throws IOException {
        BarfException expectedBarf = new BarfException();
        Handler expectedHandler = new BarfOnReactorInit(expectedBarf);
        reactor.setHandler(expectedHandler);
        assertReactorRunBarfsOnHandler(reactor, expectedBarf, expectedHandler);
        reactor.free();
    }

    @Test
    public void barfOnReactorAdd() throws IOException {
        BarfException expectedBarf = new BarfException();
        Handler expectedHandler = new BarfOnReactorInit(expectedBarf);
        reactor.getHandler().add(expectedHandler);
        assertReactorRunBarfsOnHandler(reactor, expectedBarf, expectedHandler);
        reactor.free();
    }

    @Test
    public void barfOnConnection() throws IOException {
        BarfException expectedBarf = new BarfException();
        Handler expectedHandler = new BarfOnConnectionInit(expectedBarf);
        reactor.connection(expectedHandler);
        assertReactorRunBarfsOnHandler(reactor, expectedBarf, expectedHandler);
        reactor.free();
    }

    @Test
    public void barfOnSession() throws IOException {
        BarfException expectedBarf = new BarfException();
        Handler expectedHandler = new BarfOnSessionInit(expectedBarf);
        reactor.connection(expectedHandler).session();
        assertReactorRunBarfsOnHandler(reactor, expectedBarf, expectedHandler);
        reactor.free();
    }

    @Test
    public void barfOnLink() throws IOException {
        BarfException expectedBarf = new BarfException();
        Handler expectedHandler = new BarfOnLinkInit(expectedBarf);
        reactor.connection(expectedHandler).session().sender("barf");
        assertReactorRunBarfsOnHandler(reactor, expectedBarf, expectedHandler);
        reactor.free();
    }

    @Test
    public void barfOnSchedule() throws IOException {
        BarfException expectedBarf = new BarfException();
        Handler expectedHandler = new BarfOnTask(expectedBarf);
        reactor.schedule(0, expectedHandler);
        assertReactorRunBarfsOnHandler(reactor, expectedBarf, expectedHandler);
        reactor.free();
    }

    @Test
    public void connectionRefused() throws IOException {
        final ServerSocket serverSocket = new ServerSocket(0, 0);

        class ConnectionHandler extends TestHandler {
            @Override
            public void onConnectionInit(Event event) {
                super.onConnectionInit(event);
                Connection connection = event.getConnection();
                connection.open();
                try {
                    serverSocket.close();
                } catch(IOException e) {
                    AssertionFailedError afe = new AssertionFailedError();
                    afe.initCause(e);
                    throw afe;
                }
            }
        }
        TestHandler connectionHandler = new ConnectionHandler();
        reactor.connectionToHost("127.0.0.1", serverSocket.getLocalPort(), connectionHandler);
        reactor.run();
        reactor.free();
        serverSocket.close();
        connectionHandler.assertEvents(Type.CONNECTION_INIT, Type.CONNECTION_LOCAL_OPEN, Type.CONNECTION_BOUND, Type.TRANSPORT_ERROR, Type.TRANSPORT_TAIL_CLOSED,
                Type.TRANSPORT_HEAD_CLOSED, Type.TRANSPORT_CLOSED, Type.CONNECTION_UNBOUND, Type.TRANSPORT);
    }
}