// Path: src/main/java/net/eusashead/iot/mqtt/MqttMessage.java
// public interface MqttMessage {
// 
//     boolean isRetained();
// 
//     int getQos();
// 
//     byte[] getPayload();
// 
// }
// 
// Path: src/main/java/net/eusashead/iot/mqtt/MqttToken.java
// public interface MqttToken {
// 
//     String getClientId();
// 
//     /**
//      * Returns the topic string for the this token or null if nothing has been
//      * published
//      *
//      * @return {@link String} topics for the token or null
//      */
//     String[] getTopics();
// 
//     /**
//      * Returns the identifier for the message associated with this token
//      *
//      * @return {@link String} message identifier
//      */
//     public int getMessageId();
// 
//     /**
//      * Whether a session is present for this topic
//      *
//      * @return <code>true</code> if session present or <code>false</code>
//      *         otherwise
//      */
//     public boolean getSessionPresent();
// 
// }
// 
// 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/paho/PublishFactory.java
// static final class PublishActionListener
//         extends BaseEmitterMqttActionListener {
// 
//     private final SingleEmitter<? super PublishToken> emitter;
// 
//     public PublishActionListener(
//             final SingleEmitter<? super PublishToken> emitter) {
//         this.emitter = Objects.requireNonNull(emitter);
//     }
// 
//     @Override
//     public OnError getOnError() {
//         return new OnError() {
// 
//             @Override
//             public void onError(final Throwable t) {
//                 PublishActionListener.this.emitter.onError(t);
//             }
//         };
//     }
// 
//     @Override
//     public void onSuccess(final IMqttToken mqttToken) {
// 
//         final PublishToken publishToken = new PublishToken() {
// 
//             @Override
//             public String getClientId() {
//                 return mqttToken.getClient().getClientId();
//             }
// 
//             @Override
//             public String[] getTopics() {
//                 return mqttToken.getTopics();
//             }
// 
//             @Override
//             public int getMessageId() {
//                 return mqttToken.getMessageId();
//             }
// 
//             @Override
//             public boolean getSessionPresent() {
//                 return mqttToken.getSessionPresent();
//             }
// 
//         };
//         this.emitter.onSuccess(publishToken);
//     }
// }


import static org.hamcrest.Matchers.isA;
import org.eclipse.paho.client.mqttv3.IMqttActionListener;
import org.eclipse.paho.client.mqttv3.IMqttAsyncClient;
import org.eclipse.paho.client.mqttv3.IMqttToken;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.junit.Assert;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.ExpectedException;
import org.junit.runner.RunWith;
import org.junit.runners.JUnit4;
import org.mockito.ArgumentCaptor;
import org.mockito.Matchers;
import org.mockito.Mockito;
import io.reactivex.Single;
import io.reactivex.SingleEmitter;
import net.eusashead.iot.mqtt.MqttMessage;
import net.eusashead.iot.mqtt.MqttToken;
import net.eusashead.iot.mqtt.PublishMessage;
import net.eusashead.iot.mqtt.PublishToken;
import net.eusashead.iot.mqtt.paho.PublishFactory.PublishActionListener;
        final ArgumentCaptor<IMqttActionListener> actionListener = ArgumentCaptor
                .forClass(IMqttActionListener.class);

        // When
        final Single<PublishToken> obs = factory.create(topic, msg);

        // Then
        Assert.assertNotNull(obs);
        obs.subscribe();
        Mockito.verify(client).publish(Matchers.same(topic),
                Matchers.same(msg.getPayload()), Matchers.anyInt(),
                Matchers.anyBoolean(), Matchers.any(),
                actionListener.capture());
        Assert.assertTrue(actionListener
                .getValue() instanceof PublishFactory.PublishActionListener);
    }

    @Test
    public void whenCreateIsCalledAndAnErrorOccursThenObserverOnErrorIsCalled()
            throws Throwable {
        this.expectedException.expectCause(isA(MqttException.class));
        final IMqttAsyncClient client = Mockito.mock(IMqttAsyncClient.class);
        Mockito.when(client.publish(Matchers.any(String.class),
                Matchers.any(byte[].class), Matchers.any(int.class),
                Matchers.any(boolean.class), Matchers.isNull(),
                Matchers.any(PublishFactory.PublishActionListener.class)))
                .thenThrow(new MqttException(
                        MqttException.REASON_CODE_CLIENT_CONNECTED));
        final PublishFactory factory = new PublishFactory(client);
        final Single<PublishToken> obs = factory.create("topic1",
// Path: src/main/java/net/eusashead/iot/mqtt/MqttMessage.java
// public interface MqttMessage {
// 
//     boolean isRetained();
// 
//     int getQos();
// 
//     byte[] getPayload();
// 
// }
// 
// Path: src/main/java/net/eusashead/iot/mqtt/MqttToken.java
// public interface MqttToken {
// 
//     String getClientId();
// 
//     /**
//      * Returns the topic string for the this token or null if nothing has been
//      * published
//      *
//      * @return {@link String} topics for the token or null
//      */
//     String[] getTopics();
// 
//     /**
//      * Returns the identifier for the message associated with this token
//      *
//      * @return {@link String} message identifier
//      */
//     public int getMessageId();
// 
//     /**
//      * Whether a session is present for this topic
//      *
//      * @return <code>true</code> if session present or <code>false</code>
//      *         otherwise
//      */
//     public boolean getSessionPresent();
// 
// }
// 
// 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/paho/PublishFactory.java
// static final class PublishActionListener
//         extends BaseEmitterMqttActionListener {
// 
//     private final SingleEmitter<? super PublishToken> emitter;
// 
//     public PublishActionListener(
//             final SingleEmitter<? super PublishToken> emitter) {
//         this.emitter = Objects.requireNonNull(emitter);
//     }
// 
//     @Override
//     public OnError getOnError() {
//         return new OnError() {
// 
//             @Override
//             public void onError(final Throwable t) {
//                 PublishActionListener.this.emitter.onError(t);
//             }
//         };
//     }
// 
//     @Override
//     public void onSuccess(final IMqttToken mqttToken) {
// 
//         final PublishToken publishToken = new PublishToken() {
// 
//             @Override
//             public String getClientId() {
//                 return mqttToken.getClient().getClientId();
//             }
// 
//             @Override
//             public String[] getTopics() {
//                 return mqttToken.getTopics();
//             }
// 
//             @Override
//             public int getMessageId() {
//                 return mqttToken.getMessageId();
//             }
// 
//             @Override
//             public boolean getSessionPresent() {
//                 return mqttToken.getSessionPresent();
//             }
// 
//         };
//         this.emitter.onSuccess(publishToken);
//     }
// }


// Path: src/test/java/net/eusashead/iot/mqtt/paho/PublishFactoryTest.java
import static org.hamcrest.Matchers.isA;
import org.eclipse.paho.client.mqttv3.IMqttActionListener;
import org.eclipse.paho.client.mqttv3.IMqttAsyncClient;
import org.eclipse.paho.client.mqttv3.IMqttToken;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.junit.Assert;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.ExpectedException;
import org.junit.runner.RunWith;
import org.junit.runners.JUnit4;
import org.mockito.ArgumentCaptor;
import org.mockito.Matchers;
import org.mockito.Mockito;
import io.reactivex.Single;
import io.reactivex.SingleEmitter;
import net.eusashead.iot.mqtt.MqttMessage;
import net.eusashead.iot.mqtt.MqttToken;
import net.eusashead.iot.mqtt.PublishMessage;
import net.eusashead.iot.mqtt.PublishToken;
import net.eusashead.iot.mqtt.paho.PublishFactory.PublishActionListener;

        final ArgumentCaptor<IMqttActionListener> actionListener = ArgumentCaptor
                .forClass(IMqttActionListener.class);

        // When
        final Single<PublishToken> obs = factory.create(topic, msg);

        // Then
        Assert.assertNotNull(obs);
        obs.subscribe();
        Mockito.verify(client).publish(Matchers.same(topic),
                Matchers.same(msg.getPayload()), Matchers.anyInt(),
                Matchers.anyBoolean(), Matchers.any(),
                actionListener.capture());
        Assert.assertTrue(actionListener
                .getValue() instanceof PublishFactory.PublishActionListener);
    }

    @Test
    public void whenCreateIsCalledAndAnErrorOccursThenObserverOnErrorIsCalled()
            throws Throwable {
        this.expectedException.expectCause(isA(MqttException.class));
        final IMqttAsyncClient client = Mockito.mock(IMqttAsyncClient.class);
        Mockito.when(client.publish(Matchers.any(String.class),
                Matchers.any(byte[].class), Matchers.any(int.class),
                Matchers.any(boolean.class), Matchers.isNull(),
                Matchers.any(PublishFactory.PublishActionListener.class)))
                .thenThrow(new MqttException(
                        MqttException.REASON_CODE_CLIENT_CONNECTED));
        final PublishFactory factory = new PublishFactory(client);
        final Single<PublishToken> obs = factory.create("topic1",
                Mockito.mock(MqttMessage.class));