/*
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.edgent.test.connectors.mqtt;

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.assertTrue;
import static org.junit.Assume.assumeTrue;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collectors;

import org.apache.edgent.connectors.mqtt.MqttConfig;
import org.apache.edgent.connectors.mqtt.MqttStreams;
import org.apache.edgent.function.BiFunction;
import org.apache.edgent.function.Function;
import org.apache.edgent.function.Supplier;
import org.apache.edgent.test.connectors.common.ConnectorTestBase;
import org.apache.edgent.test.connectors.common.TestRepoPath;
import org.apache.edgent.topology.TSink;
import org.apache.edgent.topology.TStream;
import org.apache.edgent.topology.Topology;
import org.apache.edgent.topology.plumbing.PlumbingStreams;
import org.apache.edgent.topology.tester.Condition;
import org.eclipse.paho.client.mqttv3.MqttClientPersistence;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.eclipse.paho.client.mqttv3.persist.MqttDefaultFilePersistence;
import org.junit.Before;
import org.junit.Test;

import com.google.gson.JsonObject;

/**
 * MqttStreams tests
 * <p>
 * The MQTT server is expected to be configured as follows:
 * <ul>
 * <li>serverURL "tcp://localhost:1883"</li>
 * <li>if the server is configured for authentication requiring
 * a username/pw, it is configured for userName="testUserName" and password="testUserNamePw"</li>
 * </ul>
 */
public class MqttStreamsTestManual extends ConnectorTestBase {

    private static final int SEC_TIMEOUT = 15;
    private static final int PUB_DELAY_MSEC = 2*1000;
    private final String BASE_CLIENT_ID = "mqttStreamsTestClientId";
    private static final String uniq = simpleTS();
    private static final String TEST_USERNAME = "testUserName";
    private static final String TEST_PASSWORD = "testUserNamePw";
    protected final Map<String,String> authInfo = new HashMap<>();
    private final String msg1 = "Hello";
    private final String msg2 = "Are you there?";
    
    public String getMsg1() {
        return msg1;
    }

    public String getMsg2() {
        return msg2;
    }

    @Before
    public void setupAuthInfo() {
        authInfo.clear();
        authInfo.put("userID", TEST_USERNAME);
        authInfo.put("password", TEST_PASSWORD);
    }
    
    protected void setSslAuthInfo(String sslAuthMode) {
        String trustStore = getKeystorePath("clientTrustStore.jks");
        authInfo.put("trustStore", trustStore);
        authInfo.put("trustStorePassword", "passw0rd");
        
        if (sslAuthMode.equals("sslClientAuth")) {
            String keyStore = getKeystorePath("clientKeyStore.jks");
            authInfo.put("keyStore", keyStore);
            authInfo.put("keyStorePassword", "passw0rd");
            // authInfo.put("keyPassword", value);
            // authInfo.put("keyCertificateAlias", value);
        }
    }
    
    protected String getKeystorePath(String storeLeaf) {
        return TestRepoPath.getPath("connectors", "mqtt", "src", "test", "keystores", storeLeaf);
    }
    
    private MqttConfig newConfig(String serverURL, String clientId) {
        MqttConfig config = new MqttConfig(serverURL, clientId);
        if (authInfo.get("userID") != null)
            config.setUserName(authInfo.get("userID"));
        if (authInfo.get("password") != null)
            config.setPassword(authInfo.get("password").toCharArray());
        if (authInfo.get("trustStore") != null)
            config.setTrustStore(authInfo.get("trustStore"));
        if (authInfo.get("trustStorePassword") != null)
            config.setTrustStorePassword(authInfo.get("trustStorePassword").toCharArray());
        if (authInfo.get("keyStore") != null)
            config.setKeyStore(authInfo.get("keyStore"));
        if (authInfo.get("keyStorePassword") != null)
            config.setKeyStorePassword(authInfo.get("keyStorePassword").toCharArray());
//        if (authInfo.get("keyPassword") != null)
//            config.setKeyPassword(authInfo.get("keyPassword").toCharArray());
//        if (authInfo.get("keyCertificateAlias") != null)
//            config.setKeyCertificateAlias(authInfo.get("keyCertificateAlias"));
        return config;
    }
       
    private static <T> byte[] serialize(T obj) {
        try {
            ByteArrayOutputStream baos = new ByteArrayOutputStream();
            ObjectOutputStream oos = new ObjectOutputStream(baos);
            oos.writeObject(obj);
            oos.close();
            return baos.toByteArray();
        } catch (IOException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
            return null;
        }
    }
    
    public static <T> T deserialize(byte[] bytes) {
        try {
            ByteArrayInputStream bais = new ByteArrayInputStream(bytes);
            ObjectInputStream ois = new ObjectInputStream(bais);
            @SuppressWarnings("unchecked")
            T obj = (T) ois.readObject();
            ois.close();
            return obj;
        } catch (Exception e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
            return null;
        }
    }
   
    protected String newClientId(String name) {
        String clientId = BASE_CLIENT_ID + "_" + name + "_" + uniq.replaceAll(":", "");
        System.out.println("["+simpleTS()+"] "
                + "Using MQTT clientID " + clientId);
        return clientId;
    }
    
    protected String[] getMqttTopics() {
        String csvTopics = "testTopic1,testTopic2";
        String[] topics = csvTopics.split(",");
        return topics;
    }
    
    protected String getServerURI() {
        return "tcp://localhost:1883";
    }
    
    protected String getSslServerURI() {
        return "ssl://localhost:8883";
    }
    
    protected String getSslClientAuthServerURI() {
        return "ssl://localhost:8884";
    }

    @Test
    public void testStringPublish() throws Exception {
        Topology top = newTopology("testStringPublish");
        MsgGenerator mgen = new MsgGenerator(top.getName());
        int qos = 0;
        boolean retain = false;
        String clientId = newClientId(top.getName());
        String topic = getMqttTopics()[0];
        List<String> msgs = createMsgs(mgen, topic, getMsg1(), getMsg2());
        
        TStream<String> s = PlumbingStreams.blockingOneShotDelay(
                top.collection(msgs), PUB_DELAY_MSEC, TimeUnit.MILLISECONDS);
        
        // Test publish(TStream<String>, topic, qos)
        // Test TStream<String> subscribe(topic, qos)
        
        MqttConfig config = newConfig(getServerURI(), clientId);
        MqttStreams mqtt = new MqttStreams(top, () -> config);

        TSink<String> sink = mqtt.publish(s, topic, qos, retain);
        TStream<String> rcvd = mqtt.subscribe(topic, qos);

        completeAndValidate(clientId, top, rcvd, mgen, SEC_TIMEOUT, msgs.toArray(new String[0]));
        
        assertNotNull(sink);
    }
    
    @Test
    public void testAutoClientId() throws Exception {
        Topology top = newTopology("testAutoClientId");
        MsgGenerator mgen = new MsgGenerator(top.getName());
        int qos = 0;
        boolean retain = false;
        String topic = getMqttTopics()[0];
        List<String> msgs = createMsgs(mgen, topic, getMsg1(), getMsg2());
        
        TStream<String> s = PlumbingStreams.blockingOneShotDelay(
                top.collection(msgs), PUB_DELAY_MSEC, TimeUnit.MILLISECONDS);
        
        // Test with auto-generated clientId
        MqttConfig config = newConfig(getServerURI(), null/*clientId*/);
        MqttStreams mqtt = new MqttStreams(top, () -> config);

        mqtt.publish(s, topic, qos, retain);
        TStream<String> rcvd = mqtt.subscribe(topic, qos);

        completeAndValidate("some-auto-clientId", top, rcvd, mgen, SEC_TIMEOUT, msgs.toArray(new String[0]));
    }
    
    @Test
    public void testQoS1() throws Exception {
        Topology top = newTopology("testQoS1");
        MsgGenerator mgen = new MsgGenerator(top.getName());
        int qos = 1;
        boolean retain = false;
        String clientId = newClientId(top.getName());
        String topic = getMqttTopics()[0];
        List<String> msgs = createMsgs(mgen, topic, getMsg1(), getMsg2());
        
        TStream<String> s = PlumbingStreams.blockingOneShotDelay(
                top.collection(msgs), PUB_DELAY_MSEC, TimeUnit.MILLISECONDS);
        
        // TODO something to verify that we actually provide
        // the QoS semantics.
        // Test publish(TStream<String>, topic, qos)
        // Test TStream<String> subscribe(topic, qos)
        
        MqttConfig config = newConfig(getServerURI(), clientId);
        MqttStreams mqtt = new MqttStreams(top, () -> config);

        mqtt.publish(s, topic, qos, retain);
        TStream<String> rcvd = mqtt.subscribe(topic, qos);

        completeAndValidate(clientId, top, rcvd, mgen, SEC_TIMEOUT, msgs.toArray(new String[0]));
    }

    @Test
    public void testQoS2() throws Exception {
        Topology top = newTopology("testQoS2");
        MsgGenerator mgen = new MsgGenerator(top.getName());
        int qos = 2;
        boolean retain = false;
        String clientId = newClientId(top.getName());
        String topic = getMqttTopics()[0];
        List<String> msgs = createMsgs(mgen, topic, getMsg1(), getMsg2());
        
        TStream<String> s = PlumbingStreams.blockingOneShotDelay(
                top.collection(msgs), PUB_DELAY_MSEC, TimeUnit.MILLISECONDS);
        
        // TODO something to verify that we actually provide
        // the QoS semantics.
        // Also improve code coverage with persistence override
        // Test publish(TStream<String>, topic, qos)
        // Test TStream<String> subscribe(topic, qos)
        
        MqttConfig config = newConfig(getServerURI(), clientId);
        config.setPersistence(new MemoryPersistence());
        MqttStreams mqtt = new MqttStreams(top, () -> config);

        mqtt.publish(s, topic, qos, retain);
        TStream<String> rcvd = mqtt.subscribe(topic, qos);

        completeAndValidate(clientId, top, rcvd, mgen, SEC_TIMEOUT, msgs.toArray(new String[0]));
    }
    
    @Test
    public void testGenericPublish() throws Exception {
        Topology top = newTopology("testGenericPublish");
        MsgGenerator mgen = new MsgGenerator(top.getName());
        int qos = 0;
        boolean retain = false;
        String clientId = newClientId(top.getName());
        // avoid mucking up other test subscribers expecting UTF-8 string payloads
        // and use a different topic
        String topic = getMqttTopics()[0] + "-Generic";
        List<String> msgs = createMsgs(mgen, topic, getMsg1(), getMsg2());
        List<String> expMsgsAsStr = 
                msgs
                .stream()
                .map(t -> (new Msg(t, topic)).toString())
                .collect(Collectors.toList());

        TStream<Msg> s = PlumbingStreams.blockingOneShotDelay(
                top.collection(msgs), PUB_DELAY_MSEC, TimeUnit.MILLISECONDS)
                .map(t -> new Msg(t, topic));
        
        // Test publish(TStream<Msg>, topicFn, payloadFn, qosFn)
        // Test subscribe(topic, qos, message2Tuple)
        
        MqttConfig config = newConfig(getServerURI(), clientId);
        MqttStreams mqtt = new MqttStreams(top, () -> config);

        TSink<Msg> sink = mqtt.publish(s, 
                                t -> t.getTopic(),
                                t -> serialize(t),
                                t -> qos,
                                t -> retain);
        TStream<Msg> rcvd = mqtt.subscribe(topic, qos,
                                (topicArg, payload) -> deserialize(payload));

        TStream<String> rcvdAsStr = rcvd.map(t -> t.toString());
        completeAndValidate(clientId, top, rcvdAsStr, mgen, SEC_TIMEOUT, expMsgsAsStr.toArray(new String[0]));
        
        assertNotNull(sink);
    }
    
    @Test
    public void testMultiConnector() throws Exception {
        Topology top = newTopology("testMultiConnector");
        MsgGenerator mgen = new MsgGenerator(top.getName());
        int qos = 0;
        boolean retain = false;
        String pubClientId = newClientId(top.getName())+"_pub";
        String subClientId = newClientId(top.getName())+"_sub";
        String topic = getMqttTopics()[0];
        List<String> msgs = createMsgs(mgen, topic, getMsg1(), getMsg2());
        
        TStream<String> s = PlumbingStreams.blockingOneShotDelay(
                top.collection(msgs), PUB_DELAY_MSEC, TimeUnit.MILLISECONDS);
        
        // Test separate connectors for pub and sub
        
        MqttConfig config = newConfig(getServerURI(), pubClientId);
        MqttStreams mqttPub = new MqttStreams(top, () -> config);
        mqttPub.publish(s, topic, qos, retain);
        
        MqttConfig configSub = newConfig(getServerURI(), subClientId);
        MqttStreams mqttSub = new MqttStreams(top, () -> configSub);
        TStream<String> rcvd = mqttSub.subscribe(topic, qos);

        completeAndValidate(pubClientId, top, rcvd, mgen, SEC_TIMEOUT, msgs.toArray(new String[0]));
    }
    
    @Test
    public void testMultiTopicPublish() throws Exception {
        Topology top = newTopology("testMultiTopicPublish");
        MsgGenerator mgen = new MsgGenerator(top.getName());
        int qos = 0;
        boolean retain = false;
        String clientId = newClientId(top.getName());
        String topic1 = getMqttTopics()[0];
        String topic2 = getMqttTopics()[1];
        List<String> msgs1 = createMsgs(mgen, topic1, getMsg1(), getMsg2());
        List<String> msgs2 = createMsgs(mgen, topic2, getMsg1(), getMsg2());
        // create an interleaved list
        List<Msg> msgs = new ArrayList<>();
        for (int i = 0; i < msgs1.size(); i++) {
            msgs.add(new Msg(msgs1.get(i), topic1));
            msgs.add(new Msg(msgs2.get(i), topic2));
        }

        TStream<Msg> s = PlumbingStreams.blockingOneShotDelay(
                top.collection(msgs), PUB_DELAY_MSEC, TimeUnit.MILLISECONDS);
        
        // Test multi-topic publish(TStream<Msg>, topicFn, payloadFn, qosFn)
        //
        // Work around current limitation that we don't support 
        // multiple subscribe() on a single connection (the last subscribe()
        // receives all msgs and the other subscribes() get none).
        
        MqttConfig config = newConfig(getServerURI(), clientId);
        MqttStreams mqtt = new MqttStreams(top, () -> config);
        MqttConfig config2 = newConfig(getServerURI(), clientId+"_sub2");
        MqttStreams mqttSub2 = new MqttStreams(top, () -> config2);

        mqtt.publish(s, 
                t -> t.getTopic(),
                t -> t.getMessage().getBytes(StandardCharsets.UTF_8),
                t -> qos,
                t -> retain);
        TStream<String> rcvd1 = mqtt.subscribe(topic1, qos);
        TStream<String> rcvd2 = mqttSub2.subscribe(topic2, qos);
        
        // validation
        
        rcvd1 = rcvd1.filter(tuple -> tuple.matches(mgen.pattern()));
        rcvd1.sink(t -> System.out.println("rcvd1: "+t));
        rcvd2 = rcvd2.filter(tuple -> tuple.matches(mgen.pattern()));
        rcvd2.sink(t -> System.out.println("rcvd2: "+t));
        
        Condition<Long> tc1 = top.getTester().tupleCount(rcvd1, msgs1.size());
        Condition<Long> tc2 = top.getTester().tupleCount(rcvd2, msgs2.size());
        
        List<Condition<Long>> conditions = new ArrayList<>();
        conditions.add(tc1);
        conditions.add(tc2);
        Condition<?> tc = top.getTester().and(tc1, tc2);

        Condition<List<String>> contents1 = top.getTester().streamContents(rcvd1, msgs1.toArray(new String[0]));
        Condition<List<String>> contents2 = top.getTester().streamContents(rcvd2, msgs2.toArray(new String[0]));

        complete(top, tc, SEC_TIMEOUT, TimeUnit.SECONDS);

        assertTrue(clientId + " contents1:" + contents1.getResult(), contents1.valid());
        assertTrue(clientId + " contents2:" + contents2.getResult(), contents2.valid());
        assertTrue("valid:" + tc, tc.valid());
    }
    
    @Test
    public void testMultiTopicSubscribe() throws Exception {
        Topology top = newTopology("testMultiTopicSubscribe");
        MsgGenerator mgen = new MsgGenerator(top.getName());
        int qos = 0;
        boolean retain = false;
        String clientId = newClientId(top.getName());
        String topic1 = getMqttTopics()[0] + "/1";
        String topic2 = getMqttTopics()[0] + "/2";
        String topics = getMqttTopics()[0] + "/+";
        List<String> msgs1 = createMsgs(mgen, topic1, getMsg1(), getMsg2());
        List<String> msgs2 = createMsgs(mgen, topic2, getMsg1(), getMsg2());
        List<String> msgs = new ArrayList<>();
        msgs.addAll(msgs1);
        msgs.addAll(msgs2);

        TStream<String> s1 = PlumbingStreams.blockingOneShotDelay(
                top.collection(msgs1), PUB_DELAY_MSEC, TimeUnit.MILLISECONDS);
        TStream<String> s2 = PlumbingStreams.blockingOneShotDelay(
                top.collection(msgs2), PUB_DELAY_MSEC, TimeUnit.MILLISECONDS);
        
        // Test multi-topic subscribe
        
        MqttConfig config = newConfig(getServerURI(), clientId);
        MqttStreams mqtt = new MqttStreams(top, () -> config);

        mqtt.publish(s1, topic1, qos, retain);
        mqtt.publish(s2, topic2, qos, retain);
        TStream<String> rcvd = mqtt.subscribe(topics, qos);
        
        completeAndValidate(false/*ordered*/, clientId, top, rcvd, mgen, SEC_TIMEOUT, msgs.toArray(new String[0]));
    }

    @Test(expected = IllegalStateException.class)
    public void testMultiSubscribeNeg() throws Exception {
        Topology top = newTopology("testMultiSubscribeNeg");
        int qos = 0;
        String clientId = newClientId(top.getName());
        String topic = getMqttTopics()[0];

        // Verify the current behavior of at-most-one subscribe()
        // for a MqttStreams instance
        
        MqttStreams mqtt = new MqttStreams(top, getServerURI(), clientId);

        mqtt.subscribe(topic, qos);
        mqtt.subscribe(topic, qos); // should throw
    }
    
    @Test
    public void testConnectFail() throws Exception {
        Topology top = newTopology("testConnectFail");
        MsgGenerator mgen = new MsgGenerator(top.getName());
        int qos = 0;
        boolean retain = false;
        String clientId = newClientId(top.getName());
        String topic = getMqttTopics()[0];
        List<String> msgs = createMsgs(mgen, topic, getMsg1(), getMsg2());
        
        TStream<String> s = PlumbingStreams.blockingOneShotDelay(
                top.collection(msgs), PUB_DELAY_MSEC, TimeUnit.MILLISECONDS);
        
        // Code coverage test: induce connection failure
        //
        // At this point the only thing we can check is an expected
        // result of 0 msgs received.
        
        MqttConfig config = newConfig("tcp://localhost:31999", clientId);
        MqttStreams mqtt = new MqttStreams(top, () -> config);

        mqtt.publish(s, topic, qos, retain);
        TStream<String> rcvd = mqtt.subscribe(topic, qos);

        completeAndValidate(clientId, top, rcvd, mgen, SEC_TIMEOUT, new String[0]);
    }
    
    private String retainTestSetup(boolean isRetained, MsgGenerator mgen) throws Exception {

        // publish a msg to [not] retain.
        
        Topology top = newTopology("retainTestSetup"+isRetained);
        int qos = 0;
        String clientId = newClientId(top.getName());
        String topic = getMqttTopics()[0];
        String retainedMsg = mgen.create(topic, isRetained ? "RETAIN-THIS" : "DO-NOT-RETAIN-THIS");

        MqttConfig config = newConfig(getServerURI(), clientId+"-setup");
        MqttStreams mqtt = new MqttStreams(top, () -> config);

        TStream<String> s = PlumbingStreams.blockingOneShotDelay(
                top.strings(retainedMsg), PUB_DELAY_MSEC, TimeUnit.MILLISECONDS);
        mqtt.publish(s, topic, qos, isRetained);
        TStream<String> rcvd = mqtt.subscribe(topic, qos);

        completeAndValidate(clientId, top, rcvd, mgen, SEC_TIMEOUT, retainedMsg);
        
        // N.B. the topology should be shutdown at this point but it isn't
        // and this topology will also receive and print the tuples
        // generated by the main portion of the test.
        // issue#46
        
        return retainedMsg;
    }

    @Test
    public void testRetainedFalse() throws Exception {
        Topology top = newTopology("testRetainedFalse");
        MsgGenerator mgen = new MsgGenerator(top.getName());
        int qos = 0;
        boolean isRetained = false;
        String clientId = newClientId(top.getName());
        String topic = getMqttTopics()[0];
        
        // Test that "retain" control works
        
        // publish the msg to [not] retain. use the same mgen but a different clientId
        String retainedMsg = retainTestSetup(isRetained, mgen);
        
        System.out.println("=============== setup complete");
        
        // verify the next connect/subscribe [doesn't] sees the retain and then new msgs
        List<String> msgs = createMsgs(mgen, topic, getMsg1(), getMsg2());
        List<String> expMsgsAsStr = 
                msgs
                .stream()
                .map(t -> (new Msg(t, topic)).toString())
                .collect(Collectors.toList());
        if (isRetained)
            expMsgsAsStr.add(0, (new Msg(retainedMsg, topic)).toString());

        TStream<String> s = PlumbingStreams.blockingOneShotDelay(
                top.collection(msgs), PUB_DELAY_MSEC, TimeUnit.MILLISECONDS);
        
        MqttConfig config = newConfig(getServerURI(), clientId);
        MqttStreams mqtt = new MqttStreams(top, () -> config);

        mqtt.publish(s, topic, qos, false/*retain*/); 
        TStream<Msg> rcvd = mqtt.subscribe(topic, qos,
                                (topicArg, payload) -> {
                                    String m = new String(payload, StandardCharsets.UTF_8);
                                    return new Msg(m, topicArg);
                                });

        TStream<String> rcvdAsStr = rcvd.map(t -> t.toString());
        completeAndValidate(clientId, top, rcvdAsStr, mgen, SEC_TIMEOUT, expMsgsAsStr.toArray(new String[0]));
    }

    @Test
    public void testRetainedTrue() throws Exception {
        Topology top = newTopology("testRetainedTrue");
        MsgGenerator mgen = new MsgGenerator(top.getName());
        int qos = 0;
        boolean isRetained = true;
        String clientId = newClientId(top.getName());
        String topic = getMqttTopics()[0];
        // Test that "retain" control works
        
        // publish the msg to [not] retain. use the same mgen but a different clientId
        String retainedMsg = retainTestSetup(isRetained, mgen);
        
        System.out.println("=============== setup complete");
        
        // verify the next connect/subscribe [doesn't] sees the retain and then new msgs
        List<String> msgs = createMsgs(mgen, topic, getMsg1(), getMsg2());
        List<String> expMsgsAsStr = 
                msgs
                .stream()
                .map(t -> (new Msg(t, topic)).toString())
                .collect(Collectors.toList());
        if (isRetained)
            expMsgsAsStr.add(0, (new Msg(retainedMsg, topic)).toString());

        TStream<String> s = PlumbingStreams.blockingOneShotDelay(
                top.collection(msgs), PUB_DELAY_MSEC, TimeUnit.MILLISECONDS);
        
        MqttConfig config = newConfig(getServerURI(), clientId);
        MqttStreams mqtt = new MqttStreams(top, () -> config);

        mqtt.publish(s, topic, qos, false/*retain*/); 
        TStream<Msg> rcvd = mqtt.subscribe(topic, qos,
                                (topicArg, payload) -> {
                                    String m = new String(payload, StandardCharsets.UTF_8);
                                    return new Msg(m, topicArg);
                                });

        TStream<String> rcvdAsStr = rcvd.map(t -> t.toString());
        completeAndValidate(clientId, top, rcvdAsStr, mgen, SEC_TIMEOUT, expMsgsAsStr.toArray(new String[0]));
    }
    
    @Test
    public void testConfig() throws Exception {
        Topology top = newTopology("testConfig");
        String clientId = newClientId(top.getName());
        
        // Test MqttConfig ctors/setters/getters

        MqttConfig config = new MqttConfig();
        assertEquals(null, config.getClientId());
        assertArrayEquals(null, config.getServerURLs());

        config = new MqttConfig(getServerURI(), clientId);
        
        {
            assertEquals(-1, config.getActionTimeToWaitMillis());
            config.setActionTimeToWaitMillis(31);
            assertEquals(31, config.getActionTimeToWaitMillis());
        }
        {
            assertEquals(clientId, config.getClientId());
            config.setClientId("xyzzyClient");
            assertEquals("xyzzyClient", config.getClientId());
        }
        {
            assertTrue(config.isCleanSession());
            config.setCleanSession(true);
            assertTrue(config.isCleanSession());
            config.setCleanSession(false);
            assertFalse(config.isCleanSession());
        }
        {
            assertEquals(30, config.getConnectionTimeout());
            config.setConnectionTimeout(11);
            assertEquals(11, config.getConnectionTimeout());
        }
        {
            assertEquals(60, config.getKeepAliveInterval());
            config.setKeepAliveInterval(12);
            assertEquals(12, config.getKeepAliveInterval());
        }
        {
            assertEquals(null, config.getKeyStore());
            config.setKeyStore("some/path");
            assertEquals("some/path", config.getKeyStore());
        }
        {
            assertArrayEquals(null, config.getKeyStorePassword());
            config.setKeyStorePassword("xyzzy".toCharArray());
            assertArrayEquals("xyzzy".toCharArray(), config.getKeyStorePassword());
        }
        {
            assertEquals(0, config.getIdleTimeout());
            config.setIdleTimeout(13);
            assertEquals(13, config.getIdleTimeout());
            config.setIdleTimeout(-1);
            assertEquals(0, config.getIdleTimeout());
        }
        {
            assertEquals(60, config.getSubscriberIdleReconnectInterval());
            config.setSubscriberIdleReconnectInterval(14);
            assertEquals(14, config.getSubscriberIdleReconnectInterval());
            config.setSubscriberIdleReconnectInterval(-1);
            assertEquals(0, config.getSubscriberIdleReconnectInterval());
        }
        {
            assertArrayEquals(null, config.getPassword());
            config.setPassword("xyzzy".toCharArray());
            assertArrayEquals("xyzzy".toCharArray(), config.getPassword());
        }
        {
            assertEquals(null, config.getPersistence());
            MqttClientPersistence p = new MqttDefaultFilePersistence();
            config.setPersistence(p);
            assertEquals(p, config.getPersistence());
            p = new MemoryPersistence();
            config.setPersistence(p);
            assertEquals(p, config.getPersistence());
        }
        {
            assertEquals(null, config.getTrustStore());
            config.setTrustStore("some/path");
            assertEquals("some/path", config.getTrustStore());
        }
        {
            assertArrayEquals(null, config.getTrustStorePassword());
            config.setTrustStorePassword("xyzzy".toCharArray());
            assertArrayEquals("xyzzy".toCharArray(), config.getTrustStorePassword());
        }
        {
            assertArrayEquals(new String[]{getServerURI()}, config.getServerURLs());
            String[] urls = new String[]{"tcp://localhost:30000","tcp://localhost:30001"};
            config.setServerURLs(urls);
            assertArrayEquals(urls, config.getServerURLs());
        }
        {
            assertEquals(null, config.getUserName());
            config.setUserName("joe");
            assertEquals("joe", config.getUserName());
        }
        {
            assertEquals(null, config.getWillDestination());
            assertArrayEquals(null, config.getWillPayload());
            assertEquals(0, config.getWillQOS());
            assertFalse(config.getWillRetained());
            
            config.setWill("willTopic", "I-AM-DEAD".getBytes(), 1, true);
            assertEquals("willTopic", config.getWillDestination());
            assertArrayEquals("I-AM-DEAD".getBytes(), config.getWillPayload());
            assertEquals(1, config.getWillQOS());
            assertTrue(config.getWillRetained());
        }
    }
    
    private static class PropertyTester {
        private final Properties props;
        private List<Checker> checkers = new ArrayList<>();

        private static class Checker {
            String name;
            String value;
            Supplier<String> getterFn;
            public Checker(String name, String value, Supplier<String> getterFn) {
                this.name = name;
                this.value = value;
                this.getterFn = getterFn;
            }
        }
        
        public PropertyTester(Properties props) {
            this.props = props;
        }
        
        public void add(String name, String value, Supplier<String> getterFn) {
            props.setProperty(name, value);
            checkers.add(new Checker(name, value, getterFn));
        }
        
        public void checkAll() throws Exception {
            for (Checker checker : checkers) {
                String val = checker.getterFn.get();
                //System.out.println("checking name="+checker.name+" exp="+checker.value+" act="+val);
                assertEquals(checker.name, checker.value, val);
            }
        }
    }
    
    @Test
    public void testConfigFromProperties() throws Exception {
        
        Properties props = new Properties();
        AtomicReference<MqttConfig> configRef = new AtomicReference<>();
        
        PropertyTester propTester = new PropertyTester(props);
        
        propTester.add("mqtt.actionTimeToWaitMillis", "10", 
                () -> ((Long)configRef.get().getActionTimeToWaitMillis()).toString());
        propTester.add("mqtt.cleanSession", "false",
                () -> ((Boolean)configRef.get().isCleanSession()).toString());
        propTester.add("mqtt.clientId", "xyzzy-clientId",
                () -> configRef.get().getClientId());
        propTester.add("mqtt.connectionTimeoutSec", "11", 
                () -> ((Integer)configRef.get().getConnectionTimeout()).toString());
        propTester.add("mqtt.idleTimeoutSec", "12", 
                () -> ((Integer)configRef.get().getIdleTimeout()).toString());
        propTester.add("mqtt.keepAliveSec", "13", 
                () -> ((Integer)configRef.get().getKeepAliveInterval()).toString());
        propTester.add("mqtt.keyStore", "some/path/keystore",
                () -> configRef.get().getKeyStore());
        propTester.add("mqtt.keyStorePassword", "some-keystore-password",
                () -> new String(configRef.get().getKeyStorePassword()));
//        propTester.add("mqtt.keyPassword", "some-key-password",
//                () -> new String(configRef.get().getKeyPassword()));
//        propTester.add("mqtt.keyCertificateAlias", "someKeyCertificateAlias",
//                () -> configRef.get().getKeyCertificateAlias());
        propTester.add("mqtt.password", "some-password",
                () -> new String(configRef.get().getPassword()));
//        propTester.add("mqtt.persistence", "some.persistence.classname",
//                () -> configRef.get().getPersistence());
        propTester.add("mqtt.serverURLs", "tcp://somehost:1234,ssl://somehost:5678",
                () -> String.join(",", configRef.get().getServerURLs()));
        propTester.add("mqtt.subscriberIdleReconnectIntervalSec", "14", 
                () -> ((Integer)configRef.get().getSubscriberIdleReconnectInterval()).toString());
        propTester.add("mqtt.trustStore", "some/path/truststore",
                () -> configRef.get().getTrustStore());
        propTester.add("mqtt.trustStorePassword", "some-truststore-password",
                () -> new String(configRef.get().getTrustStorePassword()));
        propTester.add("mqtt.userName", "xyzzy-username",
                () -> configRef.get().getUserName());
        
        JsonObject will = new JsonObject();
        will.addProperty("topic", "someWillTopic");
        will.addProperty("payload", "someWillPayload");
        will.addProperty("qos", 1);
        will.addProperty("retained", true);
        propTester.add("mqtt.will", will.toString(),
                () -> {
                        JsonObject actWill = new JsonObject();
                        actWill.addProperty("topic", configRef.get().getWillDestination());
                        actWill.addProperty("payload", new String(configRef.get().getWillPayload()));
                        actWill.addProperty("qos", configRef.get().getWillQOS());
                        actWill.addProperty("retained", configRef.get().getWillRetained());
                        return actWill.toString();
                      });

        configRef.set(MqttConfig.fromProperties(props));
        propTester.checkAll();
    }
    
    @Test
    public void testMultipleServerURL() throws Exception {
        Topology top = newTopology("testMultipleServerURL");
        MsgGenerator mgen = new MsgGenerator(top.getName());
        int qos = 0;
        boolean retain = false;
        String clientId = newClientId(top.getName());
        String topic = getMqttTopics()[0];
        List<String> msgs = createMsgs(mgen, topic, getMsg1(), getMsg2());
        
        TStream<String> s = PlumbingStreams.blockingOneShotDelay(
                top.collection(msgs), PUB_DELAY_MSEC, TimeUnit.MILLISECONDS);
        
        // Test server URL selection - supply a bogus 1st URL.
        
        String[] serverURLs = new String[] {"tcp://localhost:31999", getServerURI()};
        MqttConfig config = newConfig(serverURLs[0], clientId);
        config.setServerURLs(serverURLs);
        MqttStreams mqtt = new MqttStreams(top, () -> config);

        mqtt.publish(s, topic, qos, retain);
        TStream<String> rcvd = mqtt.subscribe(topic, qos);

        completeAndValidate(clientId, top, rcvd, mgen, SEC_TIMEOUT, msgs.toArray(new String[0]));
    }
    
    @Test
    public void testActionTime() throws Exception {
        Topology top = newTopology("testActionTime");
        MsgGenerator mgen = new MsgGenerator(top.getName());
        int qos = 0;
        boolean retain = false;
        String clientId = newClientId(top.getName());
        String topic = getMqttTopics()[0];
        List<String> msgs = createMsgs(mgen, topic, getMsg1(), getMsg2());
        
        TStream<String> s = PlumbingStreams.blockingOneShotDelay(
                top.collection(msgs), PUB_DELAY_MSEC, TimeUnit.MILLISECONDS);
        
        // Test publish(TStream<String>, topic, qos)
        // Test TStream<String> subscribe(topic, qos)
        
        MqttConfig config = newConfig(getServerURI(), clientId);
        config.setActionTimeToWaitMillis(3*1000);
        MqttStreams mqtt = new MqttStreams(top, () -> config);

        mqtt.publish(s, topic, qos, retain);
        TStream<String> rcvd = mqtt.subscribe(topic, qos);

        completeAndValidate(clientId, top, rcvd, mgen, SEC_TIMEOUT, msgs.toArray(new String[0]));
    }
    
    @Test
    public void testIdleSubscribe() throws Exception {
        Topology top = newTopology("testIdleSubscribe");
        MsgGenerator mgen = new MsgGenerator(top.getName());
        int qos = 0;
        boolean retain = false;
        String pubClientId = newClientId(top.getName()+"_pub");
        String subClientId = newClientId(top.getName()+"_sub");
        String topic = getMqttTopics()[0];
        List<String> msgs = createMsgs(mgen, topic, getMsg1(), getMsg2());

        // Exercise idle timeouts.  We won't have any direct
        // evidence that an idle disconnect/reconnect happen
        // but the code coverage will verify the paths were
        // exercised
        
        // Hmm... this test may be prone to intermittent failure:
        // if the subscriber isn't connected when the tuples
        // are published they won't be received.  May have to
        // change to qos=1 and cleanSession=false.
        
        // delay enough to let idle disconnect/reconnect trigger
        TStream<String> s = PlumbingStreams.blockingOneShotDelay(
                top.collection(msgs), 7, TimeUnit.SECONDS);
        
        MqttConfig pubConfig = newConfig(getServerURI(), pubClientId);
        MqttStreams pubMqtt = new MqttStreams(top, () -> pubConfig);

        MqttConfig subConfig = newConfig(getServerURI(), subClientId);
        subConfig.setIdleTimeout(1);
        subConfig.setSubscriberIdleReconnectInterval(1);
        MqttStreams subMqtt = new MqttStreams(top, () -> subConfig);

        pubMqtt.publish(s, topic, qos, retain);

        TStream<String> rcvd = subMqtt.subscribe(topic, qos);

        completeAndValidate(subClientId, top, rcvd, mgen, SEC_TIMEOUT, msgs.toArray(new String[0]));
    }
    
    @Test
    public void testIdlePublish() throws Exception {
        Topology top = newTopology("testIdlePublish");
        MsgGenerator mgen = new MsgGenerator(top.getName());
        int qos = 0;
        boolean retain = false;
        String pubClientId = newClientId(top.getName()+"_pub");
        String subClientId = newClientId(top.getName()+"_sub");
        String topic = getMqttTopics()[0];
        List<String> msgs = createMsgs(mgen, topic, getMsg1(), getMsg2());

        // Exercise idle timeouts.  We won't have any direct
        // evidence that an idle disconnect/reconnect happen
        // but the code coverage will verify the paths were
        // exercised
        
        TStream<String> s = PlumbingStreams.blockingOneShotDelay(
                top.collection(msgs), PUB_DELAY_MSEC, TimeUnit.MILLISECONDS);
        s = PlumbingStreams.blockingThrottle(
                top.collection(msgs), 4, TimeUnit.SECONDS);

        MqttConfig pubConfig = newConfig(getServerURI(), pubClientId);
        pubConfig.setIdleTimeout(2);
        MqttStreams pubMqtt = new MqttStreams(top, () -> pubConfig);

        MqttConfig subConfig = newConfig(getServerURI(), subClientId);
        MqttStreams subMqtt = new MqttStreams(top, () -> subConfig);

        pubMqtt.publish(s, topic, qos, retain);

        TStream<String> rcvd = subMqtt.subscribe(topic, qos);

        completeAndValidate(subClientId, top, rcvd, mgen, SEC_TIMEOUT, msgs.toArray(new String[0]));
    }
    
    @Test
    public void testConnectRetryPub() throws Exception {
        Topology top = newTopology("testConnectRetryPub");
        MsgGenerator mgen = new MsgGenerator(top.getName());
        int qos = 0;
        boolean retain = false;
        String pubClientId = newClientId(top.getName()+"_pub");
        String subClientId = newClientId(top.getName()+"_sub");
        String topic = getMqttTopics()[0];
        List<String> msgs = createMsgs(mgen, topic, getMsg1(), getMsg2());

        // Exercise connection retry by first specifying
        // a bogus server url then a good one.
        
        // Note qos==0 so its critical for the test that the subscriber
        // is connected before the pub connects and sends.
        
        TStream<String> s = PlumbingStreams.blockingOneShotDelay(
                top.collection(msgs), PUB_DELAY_MSEC, TimeUnit.MILLISECONDS);
        
        final String goodServerURI = getServerURI();
        MqttConfig pubConfig = newConfig("tcp://localhost:31999", pubClientId);
        MqttStreams pubMqtt = new MqttStreams(top, 
                new Supplier<MqttConfig>() {
                    private static final long serialVersionUID = 1L;
                    private int getCnt;

                    @Override
                    public MqttConfig get() {
                        ++getCnt;
                        System.err.println("**** getCnt:"+getCnt);
                        // delay enough to induce multiple retry w/o exceeding test harness timeout
                        if (getCnt == 4)
                            pubConfig.setServerURLs(new String[]{ goodServerURI });
                        return pubConfig;
                    }
                });

        MqttConfig subConfig = newConfig(getServerURI(), subClientId);
        MqttStreams subMqtt = new MqttStreams(top, () -> subConfig);

        pubMqtt.publish(s, topic, qos, retain);

        TStream<String> rcvd = subMqtt.subscribe(topic, qos);

        // add extra TMO delay. getting intermittent TMO failures
        completeAndValidate(subClientId, top, rcvd, mgen, SEC_TIMEOUT + 5, msgs.toArray(new String[0]));
    }
    
    @Test
    public void testConnectRetrySub() throws Exception {
        // Timing variances on shared machines can cause this test to fail
        assumeTrue(!Boolean.getBoolean("edgent.build.ci"));

        Topology top = newTopology("testConnectRetrySub");
        MsgGenerator mgen = new MsgGenerator(top.getName());
        int qos = 0;
        boolean retain = false;
        String pubClientId = newClientId(top.getName()+"_pub");
        String subClientId = newClientId(top.getName()+"_sub");
        String topic = getMqttTopics()[0];
        List<String> msgs = createMsgs(mgen, topic, getMsg1(), getMsg2());

        // Exercise connection retry by first specifying
        // a bogus server url then a good one.

        // Note qos==0 so its critical for the test that the subscriber
        // is connected before the pub connects and sends.
        // Even if we pub @ qos>0, since the broker doesn't know of
        // the sub's new clientId, the msgs will get tossed.
        // (and if it did know of the clientId we'd also have to
        // subConfig.setCleanSession(false) to avoid the msgs getting
        // tossed upon connect).
        //
        // The 15s pub delay below *should* cover things.  We've seen
        // intermittent 11s sub connect delays.  Still, would be
        // better to have a more robust scheme/interlock here.
        
        TStream<String> s = PlumbingStreams.blockingOneShotDelay(
                top.collection(msgs), 15, TimeUnit.SECONDS);
        
        final String goodServerURI = getServerURI();

        MqttConfig pubConfig = newConfig(getServerURI(), pubClientId);
        MqttStreams pubMqtt = new MqttStreams(top, () -> pubConfig);
        
        MqttConfig subConfig = newConfig("tcp://localhost:31999", subClientId);
        MqttStreams subMqtt = new MqttStreams(top, 
                new Supplier<MqttConfig>() {
                    private static final long serialVersionUID = 1L;
                    private int getCnt;

                    @Override
                    public MqttConfig get() {
                        ++getCnt;
                        System.err.println("**** getCnt:"+getCnt);                            
                        // delay enough to induce retry w/o exceeding test harness timeout
                        // or missing published tuples
                       if (getCnt == 4)
                            subConfig.setServerURLs(new String[]{ goodServerURI });
                        return subConfig;
                    }
                });

        pubMqtt.publish(s, topic, qos, retain);

        TStream<String> rcvd = subMqtt.subscribe(topic, qos);

        // add extra TMO delay. getting intermittent TMO failures
        completeAndValidate(subClientId, top, rcvd, mgen, SEC_TIMEOUT + 5, msgs.toArray(new String[0]));
    }
    
    @Test
    public void testSubscribeFnThrow() throws Exception {
        Topology top = newTopology("testSubscribeFnThrow");
        MsgGenerator mgen = new MsgGenerator(top.getName());
        int qos = 0;
        boolean retain = false;
        String clientId = newClientId(top.getName());
        String topic = getMqttTopics()[0];
        List<String> msgs = createMsgs(mgen, topic, getMsg1(), getMsg2());
        List<String> expectMsgs = new ArrayList<>(msgs);
        expectMsgs.remove(0); // should only lose the 1st tuple
        
        TStream<String> s = PlumbingStreams.blockingOneShotDelay(
                top.collection(msgs), PUB_DELAY_MSEC, TimeUnit.MILLISECONDS);
        
        // improve code coverage by having the subscribe fn throw
        
        MqttConfig config = newConfig(getServerURI(), clientId);
        MqttStreams mqtt = new MqttStreams(top, () -> config);

        mqtt.publish(s, topic, qos, retain);
        TStream<String> rcvd = mqtt.subscribe(topic, qos,
                new BiFunction<String, byte[], String>() {
                    private static final long serialVersionUID = 1L;
                    int tupCnt;

                    @Override
                    public String apply(String t, byte[] payload) {
                        String msg = new String(payload, StandardCharsets.UTF_8);
                        // caution "retained" msgs from prior tests
                        if (msg.matches(mgen.pattern())) {
                            ++tupCnt;
                            if (tupCnt == 1)
                                throw new RuntimeException("TEST message2tuple thrown exception");
                        }
                        return msg;
                    }
                });

        completeAndValidate(clientId, top, rcvd, mgen, SEC_TIMEOUT, expectMsgs.toArray(new String[0]));
    }
    
    @Test
    public void testPublishFnThrow() throws Exception {
        Topology top = newTopology("testPublishFnThrow");
        MsgGenerator mgen = new MsgGenerator(top.getName());
        int qos = 0;
        boolean retain = false;
        String clientId = newClientId(top.getName());
        String topic = getMqttTopics()[0];
        List<String> msgs = createMsgs(mgen, topic, getMsg1(), getMsg2());
        List<String> expectMsgs = new ArrayList<>(msgs);
        expectMsgs.remove(0); // should only lose the 1st tuple
        
        TStream<String> s = PlumbingStreams.blockingOneShotDelay(
                top.collection(msgs), PUB_DELAY_MSEC, TimeUnit.MILLISECONDS);

        // improve code coverage by having a publish tuple fn throw
        
        MqttConfig config = newConfig(getServerURI(), clientId);
        MqttStreams mqtt = new MqttStreams(top, () -> config);

        mqtt.publish(s, 
                tuple -> topic,
                tuple -> tuple.getBytes(StandardCharsets.UTF_8),
                tuple -> qos,
                new Function<String,Boolean>() {
                    private static final long serialVersionUID = 1L;
                    int tupCnt;

                    @Override
                    public Boolean apply(String tuple) {
                        tupCnt++;
                        if (tupCnt == 1)
                            throw new RuntimeException("TEST RETAIN-FN THROWN");
                        return retain;
                    }
                });
        TStream<String> rcvd = mqtt.subscribe(topic, qos);

        completeAndValidate(clientId, top, rcvd, mgen, SEC_TIMEOUT, expectMsgs.toArray(new String[0]));
    }
    
    /* 
     * See mqtt/src/test/keystores/README for info about SSL/TLS and mosquitto
     */
    
    @Test
    public void testSsl() throws Exception {
        Topology top = newTopology("testSsl");
        MsgGenerator mgen = new MsgGenerator(top.getName());
        int qos = 0;
        boolean retain = false;
        String clientId = newClientId(top.getName());
        String topic = getMqttTopics()[0];
        List<String> msgs = createMsgs(mgen, topic, getMsg1(), getMsg2());
        
        TStream<String> s = PlumbingStreams.blockingOneShotDelay(
                top.collection(msgs), PUB_DELAY_MSEC, TimeUnit.MILLISECONDS);
        
        // Test publish(TStream<String>, topic, qos)
        // Test TStream<String> subscribe(topic, qos)

        // System.setProperty("javax.net.debug", "ssl"); // or "all"; "help" for full list
        
        setSslAuthInfo("ssl");
        MqttConfig config = newConfig(getSslServerURI(), clientId);
        MqttStreams mqtt = new MqttStreams(top, () -> config);

        TSink<String> sink = mqtt.publish(s, topic, qos, retain);
        TStream<String> rcvd = mqtt.subscribe(topic, qos);

        completeAndValidate(clientId, top, rcvd, mgen, SEC_TIMEOUT, msgs.toArray(new String[0]));
        
        assertNotNull(sink);
    }
    
    @Test
    public void testSslClientAuth() throws Exception {
        Topology top = newTopology("testSslClientAuth");
        MsgGenerator mgen = new MsgGenerator(top.getName());
        int qos = 0;
        boolean retain = false;
        String clientId = newClientId(top.getName());
        String topic = getMqttTopics()[0];
        List<String> msgs = createMsgs(mgen, topic, getMsg1(), getMsg2());
        
        TStream<String> s = PlumbingStreams.blockingOneShotDelay(
                top.collection(msgs), PUB_DELAY_MSEC, TimeUnit.MILLISECONDS);
        
        // Test publish(TStream<String>, topic, qos)
        // Test TStream<String> subscribe(topic, qos)
        
        // System.setProperty("javax.net.debug", "ssl"); // or "all"; "help" for full list
        
        setSslAuthInfo("sslClientAuth");
        MqttConfig config = newConfig(getSslClientAuthServerURI(), clientId);
        MqttStreams mqtt = new MqttStreams(top, () -> config);

        TSink<String> sink = mqtt.publish(s, topic, qos, retain);
        TStream<String> rcvd = mqtt.subscribe(topic, qos);

        completeAndValidate(clientId, top, rcvd, mgen, SEC_TIMEOUT, msgs.toArray(new String[0]));
        
        assertNotNull(sink);
    }

}