// Path: src/main/java/net/eusashead/iot/mqtt/PublishMessage.java
// public interface PublishMessage extends MqttMessage {
// 
//     static PublishMessage create(final byte[] payload, final int qos,
//             final boolean retained) {
//         return new PublishMessageImpl(payload, qos, retained);
//     }
// 
//     class PublishMessageImpl extends AbstractMqttMessage
//             implements PublishMessage {
// 
//         private PublishMessageImpl(final byte[] payload, final int qos,
//                 final boolean retained) {
//             super(payload, qos, retained);
//         }
// 
//         @Override
//         public String toString() {
//             return "PublishMessageImpl [payload=" + Arrays.toString(payload)
//                     + ", qos=" + qos + ", retained=" + retained + "]";
//         }
// 
//     }
// 
// }
// 
// Path: src/main/java/net/eusashead/iot/mqtt/PublishToken.java
// public interface PublishToken extends MqttToken {
// 
// }
// 
// Path: src/main/java/net/eusashead/iot/mqtt/SubscribeMessage.java
// public interface SubscribeMessage extends MqttMessage {
// 
//     int getId();
// 
//     String getTopic();
// 
//     static SubscribeMessage create(final int id, final String topic,
//             final byte[] payload, final int qos, final boolean retained) {
//         return new SubscribeMessageImpl(id, topic, payload, qos, retained);
//     }
// 
//     class SubscribeMessageImpl extends AbstractMqttMessage
//             implements SubscribeMessage {
// 
//         private final int id;
//         private final String topic;
// 
//         private SubscribeMessageImpl(final int id, final String topic,
//                 final byte[] payload, final int qos, final boolean retained) {
//             super(payload, qos, retained);
//             this.id = Objects.requireNonNull(id);
//             this.topic = Objects.requireNonNull(topic);
//         }
// 
//         @Override
//         public int getId() {
//             return this.id;
//         }
// 
//         @Override
//         public String getTopic() {
//             return this.topic;
//         }
// 
//         @Override
//         public int hashCode() {
//             final int prime = 31;
//             int result = super.hashCode();
//             result = prime * result + id;
//             result = prime * result + ((topic == null) ? 0 : topic.hashCode());
//             return result;
//         }
// 
//         @Override
//         public boolean equals(final Object obj) {
//             if (this == obj) {
//                 return true;
//             }
//             if (!super.equals(obj)) {
//                 return false;
//             }
//             if (getClass() != obj.getClass()) {
//                 return false;
//             }
//             final SubscribeMessageImpl other = (SubscribeMessageImpl) obj;
//             if (id != other.id) {
//                 return false;
//             }
//             if (topic == null) {
//                 if (other.topic != null) {
//                     return false;
//                 }
//             } else if (!topic.equals(other.topic)) {
//                 return false;
//             }
//             return true;
//         }
// 
//         @Override
//         public String toString() {
//             return "SubscribeMessageImpl [id=" + id + ", topic=" + topic
//                     + ", payload=" + Arrays.toString(payload) + ", qos=" + qos
//                     + ", retained=" + retained + "]";
//         }
// 
//     }
// 
// }


import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import org.eclipse.paho.client.mqttv3.IMqttAsyncClient;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttAsyncClient;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.JUnit4;
import io.reactivex.Completable;
import io.reactivex.Single;
import net.eusashead.iot.mqtt.PublishMessage;
import net.eusashead.iot.mqtt.PublishToken;
import net.eusashead.iot.mqtt.SubscribeMessage;
    @Test
    public void itCanSubscribe() throws Throwable {

        AsyncPahoUtils.connect(this.asyncClient);

        final CountDownLatch latch = new CountDownLatch(2);
        final AtomicReference<IMqttDeliveryToken> token = new AtomicReference<>();
        final AtomicReference<SubscribeMessage> result = new AtomicReference<>();

        // Callback to monitor delivery completion
        this.asyncClient.setCallback(new MqttCallback() {

            @Override
            public void messageArrived(final String topic,
                    final MqttMessage m)
                    throws Exception {
            }

            @Override
            public void deliveryComplete(final IMqttDeliveryToken t) {
                token.set(t);
                latch.countDown();
            }

            @Override
            public void connectionLost(final Throwable cause) {
            }
        });

        // Subscribe
// Path: src/main/java/net/eusashead/iot/mqtt/PublishMessage.java
// public interface PublishMessage extends MqttMessage {
// 
//     static PublishMessage create(final byte[] payload, final int qos,
//             final boolean retained) {
//         return new PublishMessageImpl(payload, qos, retained);
//     }
// 
//     class PublishMessageImpl extends AbstractMqttMessage
//             implements PublishMessage {
// 
//         private PublishMessageImpl(final byte[] payload, final int qos,
//                 final boolean retained) {
//             super(payload, qos, retained);
//         }
// 
//         @Override
//         public String toString() {
//             return "PublishMessageImpl [payload=" + Arrays.toString(payload)
//                     + ", qos=" + qos + ", retained=" + retained + "]";
//         }
// 
//     }
// 
// }
// 
// Path: src/main/java/net/eusashead/iot/mqtt/PublishToken.java
// public interface PublishToken extends MqttToken {
// 
// }
// 
// Path: src/main/java/net/eusashead/iot/mqtt/SubscribeMessage.java
// public interface SubscribeMessage extends MqttMessage {
// 
//     int getId();
// 
//     String getTopic();
// 
//     static SubscribeMessage create(final int id, final String topic,
//             final byte[] payload, final int qos, final boolean retained) {
//         return new SubscribeMessageImpl(id, topic, payload, qos, retained);
//     }
// 
//     class SubscribeMessageImpl extends AbstractMqttMessage
//             implements SubscribeMessage {
// 
//         private final int id;
//         private final String topic;
// 
//         private SubscribeMessageImpl(final int id, final String topic,
//                 final byte[] payload, final int qos, final boolean retained) {
//             super(payload, qos, retained);
//             this.id = Objects.requireNonNull(id);
//             this.topic = Objects.requireNonNull(topic);
//         }
// 
//         @Override
//         public int getId() {
//             return this.id;
//         }
// 
//         @Override
//         public String getTopic() {
//             return this.topic;
//         }
// 
//         @Override
//         public int hashCode() {
//             final int prime = 31;
//             int result = super.hashCode();
//             result = prime * result + id;
//             result = prime * result + ((topic == null) ? 0 : topic.hashCode());
//             return result;
//         }
// 
//         @Override
//         public boolean equals(final Object obj) {
//             if (this == obj) {
//                 return true;
//             }
//             if (!super.equals(obj)) {
//                 return false;
//             }
//             if (getClass() != obj.getClass()) {
//                 return false;
//             }
//             final SubscribeMessageImpl other = (SubscribeMessageImpl) obj;
//             if (id != other.id) {
//                 return false;
//             }
//             if (topic == null) {
//                 if (other.topic != null) {
//                     return false;
//                 }
//             } else if (!topic.equals(other.topic)) {
//                 return false;
//             }
//             return true;
//         }
// 
//         @Override
//         public String toString() {
//             return "SubscribeMessageImpl [id=" + id + ", topic=" + topic
//                     + ", payload=" + Arrays.toString(payload) + ", qos=" + qos
//                     + ", retained=" + retained + "]";
//         }
// 
//     }
// 
// }


// Path: src/test/java/net/eusashead/iot/mqtt/paho/PahoObservableMqttClientITCase.java
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import org.eclipse.paho.client.mqttv3.IMqttAsyncClient;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttAsyncClient;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.JUnit4;
import io.reactivex.Completable;
import io.reactivex.Single;
import net.eusashead.iot.mqtt.PublishMessage;
import net.eusashead.iot.mqtt.PublishToken;
import net.eusashead.iot.mqtt.SubscribeMessage;

    @Test
    public void itCanSubscribe() throws Throwable {

        AsyncPahoUtils.connect(this.asyncClient);

        final CountDownLatch latch = new CountDownLatch(2);
        final AtomicReference<IMqttDeliveryToken> token = new AtomicReference<>();
        final AtomicReference<SubscribeMessage> result = new AtomicReference<>();

        // Callback to monitor delivery completion
        this.asyncClient.setCallback(new MqttCallback() {

            @Override
            public void messageArrived(final String topic,
                    final MqttMessage m)
                    throws Exception {
            }

            @Override
            public void deliveryComplete(final IMqttDeliveryToken t) {
                token.set(t);
                latch.countDown();
            }

            @Override
            public void connectionLost(final Throwable cause) {
            }
        });

        // Subscribe
        final PublishMessage expected = PublishMessage