// Path: lib/src/main/java/com/github/niqdev/openwebnet/message/OpenMessage.java
// public interface OpenMessage {
// 
//     String ACK = "*#*1##";
//     String NACK = "*#*0##";
//     String FRAME_START = "*";
//     String FRAME_END = "##";
// 
//     /**
//      * Raw message value.
//      *
//      * @return value
//      */
//     String getValue();
// 
// }
// 
// Path: lib/src/main/java/com/github/niqdev/openwebnet/message/ResponseOpenMessage.java
// public class ResponseOpenMessage implements OpenMessage {
// 
//     private final String value;
// 
//     public ResponseOpenMessage(String value) {
//         this.value = value;
//     }
// 
//     @Override
//     public String getValue() {
//         return value;
//     }
// 
//     @Override
//     public String toString() {
//         return String.format("{response=%s}", value);
//     }
// }
// 
// Path: lib/src/main/java/com/github/niqdev/openwebnet/OpenSession.java
// public static OpenSession newSession(OpenMessage request) {
//     return new OpenSession(request);
// }
// 
// Path: lib/src/main/java/com/github/niqdev/openwebnet/OpenWebNet.java
// enum Channel {
// 
//     COMMAND("*99*0##"),
//     EVENT("*99*1##");
// 
//     private final String value;
// 
//     Channel(String value) {
//         this.value = value;
//     }
// 
//     public String value() {
//         return value;
//     }
// 
// }
// 
// Path: lib/src/main/java/com/github/niqdev/openwebnet/OpenWebNet.java
// public interface OpenGateway {
// 
//     /**
//      * Default gateway port is 20000.
//      */
//     int DEFAULT_PORT = 20000;
// 
//     /**
//      * Returns the gateway ip address or domain.
//      *
//      * @return host
//      */
//     String getHost();
// 
//     /**
//      * Returns the gateway port.
//      *
//      * @return port
//      */
//     int getPort();
// 
//     /**
//      * Returns the gateway password.
//      *
//      * @return password
//      */
//     String getPassword();
// }


import com.github.niqdev.openwebnet.message.OpenMessage;
import com.github.niqdev.openwebnet.message.ResponseOpenMessage;
import com.google.common.base.Splitter;
import com.google.common.collect.FluentIterable;
import com.google.common.collect.ImmutableList;
import rx.Observable;
import rx.Statement;
import rx.functions.Func1;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;
import static com.github.niqdev.openwebnet.OpenSession.newSession;
import static com.github.niqdev.openwebnet.OpenWebNet.Channel;
import static com.github.niqdev.openwebnet.OpenWebNet.OpenGateway;
import static com.github.niqdev.openwebnet.message.OpenMessage.ACK;
import static com.github.niqdev.openwebnet.message.OpenMessage.FRAME_END;
        return Observable.defer(() -> {
            try {
                OpenContext context = OpenContext.setup(gateway);
                context.connect();
                log("connected!");
                return Observable.just(context);
            } catch (IOException e) {
                return Observable.error(e);
            }
        })
        .doAfterTerminate(() -> {
            // TODO how retrieve context to handle unsubscribe/close socket?
            //context.disconnect();
        })
        .timeout(5, TimeUnit.SECONDS)
        .doOnError(throwable -> {
            System.out.println("ERROR connect: " + throwable);
        });
    }

    static Func1<OpenContext, Observable<OpenContext>> doHandshake(Channel channel) {
        return context -> Observable.just(context)
            .flatMap(read())
            .flatMap(expectedAck(context))
            .flatMap(write(channel.value()))
            .flatMap(read())
            .flatMap(verifyCredential(context))
            .flatMap(expectedAck(context));
    }

// Path: lib/src/main/java/com/github/niqdev/openwebnet/message/OpenMessage.java
// public interface OpenMessage {
// 
//     String ACK = "*#*1##";
//     String NACK = "*#*0##";
//     String FRAME_START = "*";
//     String FRAME_END = "##";
// 
//     /**
//      * Raw message value.
//      *
//      * @return value
//      */
//     String getValue();
// 
// }
// 
// Path: lib/src/main/java/com/github/niqdev/openwebnet/message/ResponseOpenMessage.java
// public class ResponseOpenMessage implements OpenMessage {
// 
//     private final String value;
// 
//     public ResponseOpenMessage(String value) {
//         this.value = value;
//     }
// 
//     @Override
//     public String getValue() {
//         return value;
//     }
// 
//     @Override
//     public String toString() {
//         return String.format("{response=%s}", value);
//     }
// }
// 
// Path: lib/src/main/java/com/github/niqdev/openwebnet/OpenSession.java
// public static OpenSession newSession(OpenMessage request) {
//     return new OpenSession(request);
// }
// 
// Path: lib/src/main/java/com/github/niqdev/openwebnet/OpenWebNet.java
// enum Channel {
// 
//     COMMAND("*99*0##"),
//     EVENT("*99*1##");
// 
//     private final String value;
// 
//     Channel(String value) {
//         this.value = value;
//     }
// 
//     public String value() {
//         return value;
//     }
// 
// }
// 
// Path: lib/src/main/java/com/github/niqdev/openwebnet/OpenWebNet.java
// public interface OpenGateway {
// 
//     /**
//      * Default gateway port is 20000.
//      */
//     int DEFAULT_PORT = 20000;
// 
//     /**
//      * Returns the gateway ip address or domain.
//      *
//      * @return host
//      */
//     String getHost();
// 
//     /**
//      * Returns the gateway port.
//      *
//      * @return port
//      */
//     int getPort();
// 
//     /**
//      * Returns the gateway password.
//      *
//      * @return password
//      */
//     String getPassword();
// }


// Path: lib/src/main/java/com/github/niqdev/openwebnet/OpenWebNetObservable.java
import com.github.niqdev.openwebnet.message.OpenMessage;
import com.github.niqdev.openwebnet.message.ResponseOpenMessage;
import com.google.common.base.Splitter;
import com.google.common.collect.FluentIterable;
import com.google.common.collect.ImmutableList;
import rx.Observable;
import rx.Statement;
import rx.functions.Func1;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;
import static com.github.niqdev.openwebnet.OpenSession.newSession;
import static com.github.niqdev.openwebnet.OpenWebNet.Channel;
import static com.github.niqdev.openwebnet.OpenWebNet.OpenGateway;
import static com.github.niqdev.openwebnet.message.OpenMessage.ACK;
import static com.github.niqdev.openwebnet.message.OpenMessage.FRAME_END;

        return Observable.defer(() -> {
            try {
                OpenContext context = OpenContext.setup(gateway);
                context.connect();
                log("connected!");
                return Observable.just(context);
            } catch (IOException e) {
                return Observable.error(e);
            }
        })
        .doAfterTerminate(() -> {
            // TODO how retrieve context to handle unsubscribe/close socket?
            //context.disconnect();
        })
        .timeout(5, TimeUnit.SECONDS)
        .doOnError(throwable -> {
            System.out.println("ERROR connect: " + throwable);
        });
    }

    static Func1<OpenContext, Observable<OpenContext>> doHandshake(Channel channel) {
        return context -> Observable.just(context)
            .flatMap(read())
            .flatMap(expectedAck(context))
            .flatMap(write(channel.value()))
            .flatMap(read())
            .flatMap(verifyCredential(context))
            .flatMap(expectedAck(context));
    }

    static Func1<OpenContext, Observable<List<OpenSession>>> doRequests(List<OpenMessage> requests) {