// Path: cocaine-client/src/main/java/cocaine/netty/ServiceMessageHandler.java
// public class ServiceMessageHandler extends ChannelInboundHandlerAdapter {
// 
//     private static final Logger logger = Logger.getLogger(ServiceMessageHandler.class);
// 
//     private final String service;
//     private final Sessions sessions;
// 
//     public ServiceMessageHandler(String service, Sessions sessions) {
//         this.service = service;
//         this.sessions = sessions;
//     }
// 
//     @Override
//     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
//         logger.info("Handling message: " + msg);
// 
//         Message message = (Message) msg;
//         long session = message.getSession();
// 
//         switch (message.getType()) {
//             case CHUNK: {
//                 ChunkMessage chunk = (ChunkMessage) msg;
//                 sessions.onChunk(session, chunk.getData());
//                 break;
//             }
//             case CHOKE: {
//                 sessions.onCompleted(session);
//                 break;
//             }
//             case ERROR: {
//                 ErrorMessage error = (ErrorMessage) msg;
//                 sessions.onError(session, new ServiceErrorException(service, error.getMessage(), error.getCode()));
//                 break;
//             }
//             default: {
//                 sessions.onError(session, new UnexpectedServiceMessageException(service, message));
//                 break;
//             }
//         }
//     }
// 
//     @Override
//     public void channelInactive(ChannelHandlerContext ctx) throws Exception {
//         sessions.onCompleted();
//     }
// }


import java.io.IOException;
import java.net.SocketAddress;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import cocaine.netty.ServiceMessageHandler;
import com.google.common.base.Joiner;
import com.google.common.base.Supplier;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import org.apache.log4j.Logger;
import org.msgpack.MessagePackable;
import org.msgpack.packer.Packer;
import org.msgpack.unpacker.Unpacker;
import rx.Observable;
        return new Service(name, api, bootstrap, endpoint);
    }

    public Observable<byte[]> invoke(String method, Object... args) {
        return invoke(method, Arrays.asList(args));
    }

    public Observable<byte[]> invoke(String method, List<Object> args) {
        logger.debug("Invoking " + method + "(" + Joiner.on(", ").join(args) + ") asynchronously");

        Sessions.Session session = sessions.create();
        int requestedMethod = api.getMethod(method);
        channel.write(new InvocationRequest(requestedMethod, session.getId(), args));

        return session.getInput();
    }

    @Override
    public void close() throws Exception {
        if (closed.compareAndSet(false, true)) {
            channel.close();
        }
    }

    @Override
    public String toString() {
        return name + "/" + channel.remoteAddress();
    }

    private void connect(final Bootstrap bootstrap, final Supplier<SocketAddress> endpoint,
// Path: cocaine-client/src/main/java/cocaine/netty/ServiceMessageHandler.java
// public class ServiceMessageHandler extends ChannelInboundHandlerAdapter {
// 
//     private static final Logger logger = Logger.getLogger(ServiceMessageHandler.class);
// 
//     private final String service;
//     private final Sessions sessions;
// 
//     public ServiceMessageHandler(String service, Sessions sessions) {
//         this.service = service;
//         this.sessions = sessions;
//     }
// 
//     @Override
//     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
//         logger.info("Handling message: " + msg);
// 
//         Message message = (Message) msg;
//         long session = message.getSession();
// 
//         switch (message.getType()) {
//             case CHUNK: {
//                 ChunkMessage chunk = (ChunkMessage) msg;
//                 sessions.onChunk(session, chunk.getData());
//                 break;
//             }
//             case CHOKE: {
//                 sessions.onCompleted(session);
//                 break;
//             }
//             case ERROR: {
//                 ErrorMessage error = (ErrorMessage) msg;
//                 sessions.onError(session, new ServiceErrorException(service, error.getMessage(), error.getCode()));
//                 break;
//             }
//             default: {
//                 sessions.onError(session, new UnexpectedServiceMessageException(service, message));
//                 break;
//             }
//         }
//     }
// 
//     @Override
//     public void channelInactive(ChannelHandlerContext ctx) throws Exception {
//         sessions.onCompleted();
//     }
// }


// Path: cocaine-client/src/main/java/cocaine/Service.java
import java.io.IOException;
import java.net.SocketAddress;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import cocaine.netty.ServiceMessageHandler;
import com.google.common.base.Joiner;
import com.google.common.base.Supplier;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import org.apache.log4j.Logger;
import org.msgpack.MessagePackable;
import org.msgpack.packer.Packer;
import org.msgpack.unpacker.Unpacker;
import rx.Observable;

        return new Service(name, api, bootstrap, endpoint);
    }

    public Observable<byte[]> invoke(String method, Object... args) {
        return invoke(method, Arrays.asList(args));
    }

    public Observable<byte[]> invoke(String method, List<Object> args) {
        logger.debug("Invoking " + method + "(" + Joiner.on(", ").join(args) + ") asynchronously");

        Sessions.Session session = sessions.create();
        int requestedMethod = api.getMethod(method);
        channel.write(new InvocationRequest(requestedMethod, session.getId(), args));

        return session.getInput();
    }

    @Override
    public void close() throws Exception {
        if (closed.compareAndSet(false, true)) {
            channel.close();
        }
    }

    @Override
    public String toString() {
        return name + "/" + channel.remoteAddress();
    }

    private void connect(final Bootstrap bootstrap, final Supplier<SocketAddress> endpoint,
            final ServiceMessageHandler handler)