// 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;
package cocaine;



/**
 * @author Anton Bobukh <abobukh@yandex-team.ru>
 */
public class Service implements AutoCloseable {

    private static final Logger logger = Logger.getLogger(Service.class);

    private final String name;
    private final ServiceApi api;
    private final Sessions sessions;

    private AtomicBoolean closed;
    private Channel channel;

    private Service(String name, ServiceApi api, Bootstrap bootstrap, Supplier<SocketAddress> endpoint) {
        this.name = name;
        this.sessions = new Sessions(name);
        this.api = api;
        this.closed = new AtomicBoolean(false);
// 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;

package cocaine;



/**
 * @author Anton Bobukh <abobukh@yandex-team.ru>
 */
public class Service implements AutoCloseable {

    private static final Logger logger = Logger.getLogger(Service.class);

    private final String name;
    private final ServiceApi api;
    private final Sessions sessions;

    private AtomicBoolean closed;
    private Channel channel;

    private Service(String name, ServiceApi api, Bootstrap bootstrap, Supplier<SocketAddress> endpoint) {
        this.name = name;
        this.sessions = new Sessions(name);
        this.api = api;
        this.closed = new AtomicBoolean(false);
        connect(bootstrap, endpoint, new ServiceMessageHandler(name, sessions));