/**
 * 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.activemq.transport.ws;

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

import java.io.IOException;
import java.util.Arrays;
import java.util.Collection;
import java.util.concurrent.TimeUnit;

import org.eclipse.jetty.client.HttpClient;
import org.eclipse.jetty.client.dynamic.HttpClientTransportDynamic;
import org.eclipse.jetty.io.ClientConnector;
import org.eclipse.jetty.util.ssl.SslContextFactory;
import org.eclipse.jetty.websocket.client.ClientUpgradeRequest;
import org.eclipse.jetty.websocket.client.WebSocketClient;
import org.fusesource.hawtbuf.Buffer;
import org.fusesource.hawtbuf.UTF8Buffer;
import org.fusesource.mqtt.client.QoS;
import org.fusesource.mqtt.client.Topic;
import org.fusesource.mqtt.codec.CONNECT;
import org.fusesource.mqtt.codec.PUBACK;
import org.fusesource.mqtt.codec.PUBLISH;
import org.fusesource.mqtt.codec.SUBSCRIBE;
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;

/**
 * This shows that last will and testament messages work with MQTT over WS.
 * This test is modeled after org.apache.activemq.transport.mqtt.MQTTWillTest
 */
@RunWith(Parameterized.class)
public class MQTTWSTransportWillTest extends WSTransportTestSupport {

    protected WebSocketClient wsClient;
    protected MQTTWSConnection wsMQTTConnection1;
    protected MQTTWSConnection wsMQTTConnection2;
    protected ClientUpgradeRequest request;

    private String willTopic = "willTopic";
    private String payload = "last will";
    private boolean closeWithDisconnect;

    //Test both with a proper disconnect and without
    @Parameters(name="closeWithDisconnect={0}")
    public static Collection<Object[]> data() {
        return Arrays.asList(new Object[][] {
                {true},
                {false}
            });
    }

    public MQTTWSTransportWillTest(boolean closeWithDisconnect) {
        this.closeWithDisconnect = closeWithDisconnect;
    }

    @Override
    @Before
    public void setUp() throws Exception {
        //turn off advisory support
        broker = createBroker(true, false);

        SslContextFactory.Client sslContextFactory = new SslContextFactory.Client();
        sslContextFactory.setTrustAll(true);
        ClientConnector clientConnector = new ClientConnector();
        clientConnector.setSslContextFactory(sslContextFactory);

        HttpClient httpClient = new HttpClient(new HttpClientTransportDynamic(clientConnector));
        wsClient = new WebSocketClient(httpClient);
        wsClient.start();

        request = new ClientUpgradeRequest();
        request.setSubProtocols("mqttv3.1");

        wsMQTTConnection1 = new MQTTWSConnection();
        wsMQTTConnection2 = new MQTTWSConnection();

        wsClient.connect(wsMQTTConnection1, wsConnectUri, request);
        if (!wsMQTTConnection1.awaitConnection(30, TimeUnit.SECONDS)) {
            throw new IOException("Could not connect to MQTT WS endpoint");
        }

        wsClient.connect(wsMQTTConnection2, wsConnectUri, request);
        if (!wsMQTTConnection2.awaitConnection(30, TimeUnit.SECONDS)) {
            throw new IOException("Could not connect to MQTT WS endpoint");
        }
    }

    @Override
    @After
    public void tearDown() throws Exception {
        if (wsMQTTConnection1 != null) {
            wsMQTTConnection1.close();
            wsMQTTConnection1 = null;
        }
        if (wsMQTTConnection2 != null) {
            wsMQTTConnection2.close();
            wsMQTTConnection2 = null;
        }
        wsClient.stop();
        wsClient = null;
        super.tearDown();
    }

    @Test(timeout = 60000)
    public void testWill() throws Exception {

        //connect with will retain false
        CONNECT command = getWillConnectCommand(false);

        //connect both connections
        wsMQTTConnection1.connect(command);
        wsMQTTConnection2.connect();

        //Subscribe to topics
        SUBSCRIBE subscribe = new SUBSCRIBE();
        subscribe.topics(new Topic[] {new Topic("#", QoS.EXACTLY_ONCE) });
        wsMQTTConnection2.sendFrame(subscribe.encode());
        wsMQTTConnection2.receive(5, TimeUnit.SECONDS);

        //Test message send/receive
        wsMQTTConnection1.sendFrame(getTestMessage((short) 125).encode());
        assertMessageReceived(wsMQTTConnection2);

        //close the first connection without sending a proper disconnect frame first
        //if closeWithDisconnect is false
        if (closeWithDisconnect) {
            wsMQTTConnection1.disconnect();
        }
        wsMQTTConnection1.close();

        //Make sure LWT message is not received
        if (closeWithDisconnect) {
            assertNull(wsMQTTConnection2.receive(5, TimeUnit.SECONDS));
        //make sure LWT is received
        } else {
            assertWillTopicReceived(wsMQTTConnection2);
        }
    }

    @Test(timeout = 60 * 1000)
    public void testRetainWillMessage() throws Exception {

        //create connection with will retain true
        CONNECT command = getWillConnectCommand(true);

        wsMQTTConnection1.connect(command);
        wsMQTTConnection2.connect();

        //set to at most once to test will retain
        SUBSCRIBE subscribe = new SUBSCRIBE();
        subscribe.topics(new Topic[] {new Topic("#", QoS.AT_MOST_ONCE) });
        wsMQTTConnection2.sendFrame(subscribe.encode());
        wsMQTTConnection2.receive(5, TimeUnit.SECONDS);

        //Test message send/receive
        PUBLISH pub = getTestMessage((short) 127);
        wsMQTTConnection1.sendFrame(pub.encode());
        assertMessageReceived(wsMQTTConnection2);
        PUBACK ack = new PUBACK();
        ack.messageId(pub.messageId());
        wsMQTTConnection2.sendFrame(ack.encode());

        //Properly close connection 2 and improperly close connection 1 for LWT test
        wsMQTTConnection2.disconnect();
        wsMQTTConnection2.close();
        Thread.sleep(1000);
        //close the first connection without sending a proper disconnect frame first
        //if closeWithoutDisconnect is false
        if (closeWithDisconnect) {
            wsMQTTConnection1.disconnect();
        }
        wsMQTTConnection1.close();
        Thread.sleep(1000);

        //Do the reconnect of the websocket after close
        wsMQTTConnection2 = new MQTTWSConnection();
        wsClient.connect(wsMQTTConnection2, wsConnectUri, request);
        if (!wsMQTTConnection2.awaitConnection(30, TimeUnit.SECONDS)) {
            throw new IOException("Could not connect to MQTT WS endpoint");
        }


        //Make sure the will message is received on reconnect
        wsMQTTConnection2.connect();
        wsMQTTConnection2.sendFrame(subscribe.encode());
        wsMQTTConnection2.receive(5, TimeUnit.SECONDS);

        //Make sure LWT message not received
        if (closeWithDisconnect) {
            assertNull(wsMQTTConnection2.receive(5, TimeUnit.SECONDS));
        //make sure LWT is received
        } else {
            assertWillTopicReceived(wsMQTTConnection2);
        }
    }

    private PUBLISH getTestMessage(short id) {
        PUBLISH publish = new PUBLISH();
        publish.dup(false);
        publish.messageId(id);
        publish.qos(QoS.AT_LEAST_ONCE);
        publish.payload(new Buffer("hello world".getBytes()));
        publish.topicName(new UTF8Buffer("test"));

        return publish;
    }

    private CONNECT getWillConnectCommand(boolean willRetain) {
        CONNECT command = new CONNECT();
        command.clientId(new UTF8Buffer("clientId"));
        command.cleanSession(false);
        command.version(3);
        command.keepAlive((short) 0);
        command.willMessage(new UTF8Buffer(payload));
        command.willQos(QoS.AT_LEAST_ONCE);
        command.willTopic(new UTF8Buffer(willTopic));
        command.willRetain(willRetain);

        return command;
    }

    private void assertMessageReceived(MQTTWSConnection wsMQTTConnection2) throws Exception {
        PUBLISH msg = new PUBLISH();
        msg.decode(wsMQTTConnection2.receive(5, TimeUnit.SECONDS));
        assertNotNull(msg);
        assertEquals("hello world", msg.payload().ascii().toString());
        assertEquals("test", msg.topicName().toString());
    }

    private void assertWillTopicReceived(MQTTWSConnection wsMQTTConnection2) throws Exception {
        PUBLISH willMsg = new PUBLISH();
        willMsg.decode(wsMQTTConnection2.receive(5, TimeUnit.SECONDS));
        assertNotNull(willMsg);
        assertEquals(payload, willMsg.payload().ascii().toString());
        assertEquals(willTopic,  willMsg.topicName().toString());
    }

}
