// Path: influent-transport/src/main/java/influent/exception/InfluentIOException.java
// public final class InfluentIOException extends RuntimeException {
//   /**
//    * Constructs a new {@code InfluentIOException}.
//    *
//    * @param message the error message
//    * @param cause the cause
//    */
//   public InfluentIOException(final String message, final Throwable cause) {
//     super(message, cause);
//   }
// 
//   /**
//    * Constructs a new {@code InfluentIOException}.
//    *
//    * @param message the error message
//    */
//   public InfluentIOException(final String message) {
//     super(message);
//   }
// 
//   /** Constructs a new {@code InfluentIOException}. */
//   public InfluentIOException() {
//     super();
//   }
// }
// 
// Path: influent-transport/src/main/java/influent/internal/nio/NioTcpChannel.java
// public interface NioTcpChannel extends AutoCloseable {
//   enum Op {
//     /** OP_READ */
//     READ(SelectionKey.OP_READ),
//     /** OP_WRITE */
//     WRITE(SelectionKey.OP_WRITE);
// 
//     private final int bit;
// 
//     Op(final int bit) {
//       this.bit = bit;
//     }
// 
//     int getBit() {
//       return bit;
//     }
// 
//     static int bits(final Set<Op> ops) {
//       return ops.stream().mapToInt(Op::getBit).reduce(0, (x, y) -> x | y);
//     }
//   }
// 
//   /**
//    * Writes bytes to the socket buffer.
//    *
//    * <p>This method is not guaranteed to be thread-safe.
//    *
//    * @param src the buffer
//    * @return true when some bytes are written
//    * @throws InfluentIOException if some IO error occurs
//    */
//   boolean write(final ByteBuffer src);
// 
//   /**
//    * Reads bytes from the socket buffer.
//    *
//    * <p>This method is not guaranteed to be thread-safe.
//    *
//    * @param dst the buffer
//    * @return true when some bytes are read
//    * @throws InfluentIOException if some IO error occurs
//    */
//   boolean read(final ByteBuffer dst);
// 
//   /**
//    * Registers the this channel to the given {@code NioEventLoop}.
//    *
//    * <p>This method is thread-safe.
//    *
//    * @param ops the operations to be enabled
//    * @param attachment the {@code NioAttachment}
//    */
//   void register(final Set<Op> ops, final NioAttachment attachment);
// 
//   /**
//    * Enables the given operation.
//    *
//    * <p>This method is thread-safe and operations are done asynchronously.
//    *
//    * @param op the operation to be enabled
//    */
//   void enable(final Op op);
// 
//   /**
//    * Disables the given operation.
//    *
//    * <p>This method is thread-safe and operations are done asynchronously.
//    *
//    * @param op the operation to be disabled
//    */
//   void disable(final Op op);
// 
//   /**
//    * Closes the {@code SocketChannel}.
//    *
//    * <p>This method is not guaranteed to be thread-safe.
//    */
//   @Override
//   void close();
// 
//   /**
//    * Returns true if this channel is open
//    *
//    * <p>This method is not guaranteed to be thread-safe.
//    */
//   boolean isOpen();
// 
//   /**
//    * Returns the remote address
//    *
//    * <p>This method is thread-safe.
//    */
//   SocketAddress getRemoteAddress();
// }


import java.nio.ByteBuffer;
import java.util.LinkedList;
import java.util.NoSuchElementException;
import java.util.Queue;
import java.util.function.Supplier;
import org.msgpack.value.ImmutableValue;
import influent.exception.InfluentIOException;
import influent.internal.nio.NioTcpChannel;
/*
 * Copyright 2016 okumin
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package influent.internal.msgpack;


/**
 * An unpacker for a MessagePack stream.
 *
 * <p>This is expected to be used in only Influent project.
 *
 * <p>{@code MsgpackStreamUnpacker} is not thread-safe.
 */
public final class MsgpackStreamUnpacker {
  private final InfluentByteBuffer buffer;
  private final long chunkSizeLimit;

  private final Queue<ImmutableValue> unpackedValues = new LinkedList<>();
  private MsgpackIncrementalUnpacker currentUnpacker = FormatUnpacker.getInstance();
  private long currentChunkSize = 0;

  /**
   * Constructs a new {@code MsgpackStreamUnpacker}.
   *
   * @param chunkSizeLimit the allowable chunk size {@code feed} fails when the size of reading
   *     chunk exceeds the limit
   */
  public MsgpackStreamUnpacker(final long chunkSizeLimit) {
    this.buffer = new InfluentByteBuffer(chunkSizeLimit);
    this.chunkSizeLimit = chunkSizeLimit;
  }

  /**
   * Reads buffers from a {@code ReadableByteChannel}.
   *
   * @param supplier supplier to produce ByteBuffer
   * @param channel channel
   * @throws InfluentIOException when it fails reading from the channel or the chunk size exceeds
   *     the limit
   */
  public void feed(final Supplier<ByteBuffer> supplier, final NioTcpChannel channel) {
    boolean toBeContinued = true;
    while (toBeContinued) {
      toBeContinued = buffer.feed(supplier);
      unpack(channel);
    }
  }

  // fails when the chunk size exceeds the limit
  private void unpack(final NioTcpChannel channel) {
    while (buffer.hasRemaining()) {
      try {
        currentChunkSize += buffer.remaining();
        final DecodeResult result = currentUnpacker.unpack(buffer);
        currentChunkSize -= buffer.remaining();
        if (result.isCompleted()) {
          unpackedValues.offer(result.value());
          currentUnpacker = FormatUnpacker.getInstance();
          currentChunkSize = 0;
        } else if (currentChunkSize >= chunkSizeLimit) {
          channel.close();
// Path: influent-transport/src/main/java/influent/exception/InfluentIOException.java
// public final class InfluentIOException extends RuntimeException {
//   /**
//    * Constructs a new {@code InfluentIOException}.
//    *
//    * @param message the error message
//    * @param cause the cause
//    */
//   public InfluentIOException(final String message, final Throwable cause) {
//     super(message, cause);
//   }
// 
//   /**
//    * Constructs a new {@code InfluentIOException}.
//    *
//    * @param message the error message
//    */
//   public InfluentIOException(final String message) {
//     super(message);
//   }
// 
//   /** Constructs a new {@code InfluentIOException}. */
//   public InfluentIOException() {
//     super();
//   }
// }
// 
// Path: influent-transport/src/main/java/influent/internal/nio/NioTcpChannel.java
// public interface NioTcpChannel extends AutoCloseable {
//   enum Op {
//     /** OP_READ */
//     READ(SelectionKey.OP_READ),
//     /** OP_WRITE */
//     WRITE(SelectionKey.OP_WRITE);
// 
//     private final int bit;
// 
//     Op(final int bit) {
//       this.bit = bit;
//     }
// 
//     int getBit() {
//       return bit;
//     }
// 
//     static int bits(final Set<Op> ops) {
//       return ops.stream().mapToInt(Op::getBit).reduce(0, (x, y) -> x | y);
//     }
//   }
// 
//   /**
//    * Writes bytes to the socket buffer.
//    *
//    * <p>This method is not guaranteed to be thread-safe.
//    *
//    * @param src the buffer
//    * @return true when some bytes are written
//    * @throws InfluentIOException if some IO error occurs
//    */
//   boolean write(final ByteBuffer src);
// 
//   /**
//    * Reads bytes from the socket buffer.
//    *
//    * <p>This method is not guaranteed to be thread-safe.
//    *
//    * @param dst the buffer
//    * @return true when some bytes are read
//    * @throws InfluentIOException if some IO error occurs
//    */
//   boolean read(final ByteBuffer dst);
// 
//   /**
//    * Registers the this channel to the given {@code NioEventLoop}.
//    *
//    * <p>This method is thread-safe.
//    *
//    * @param ops the operations to be enabled
//    * @param attachment the {@code NioAttachment}
//    */
//   void register(final Set<Op> ops, final NioAttachment attachment);
// 
//   /**
//    * Enables the given operation.
//    *
//    * <p>This method is thread-safe and operations are done asynchronously.
//    *
//    * @param op the operation to be enabled
//    */
//   void enable(final Op op);
// 
//   /**
//    * Disables the given operation.
//    *
//    * <p>This method is thread-safe and operations are done asynchronously.
//    *
//    * @param op the operation to be disabled
//    */
//   void disable(final Op op);
// 
//   /**
//    * Closes the {@code SocketChannel}.
//    *
//    * <p>This method is not guaranteed to be thread-safe.
//    */
//   @Override
//   void close();
// 
//   /**
//    * Returns true if this channel is open
//    *
//    * <p>This method is not guaranteed to be thread-safe.
//    */
//   boolean isOpen();
// 
//   /**
//    * Returns the remote address
//    *
//    * <p>This method is thread-safe.
//    */
//   SocketAddress getRemoteAddress();
// }


// Path: influent-java/src/main/java/influent/internal/msgpack/MsgpackStreamUnpacker.java
import java.nio.ByteBuffer;
import java.util.LinkedList;
import java.util.NoSuchElementException;
import java.util.Queue;
import java.util.function.Supplier;
import org.msgpack.value.ImmutableValue;
import influent.exception.InfluentIOException;
import influent.internal.nio.NioTcpChannel;

/*
 * Copyright 2016 okumin
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package influent.internal.msgpack;


/**
 * An unpacker for a MessagePack stream.
 *
 * <p>This is expected to be used in only Influent project.
 *
 * <p>{@code MsgpackStreamUnpacker} is not thread-safe.
 */
public final class MsgpackStreamUnpacker {
  private final InfluentByteBuffer buffer;
  private final long chunkSizeLimit;

  private final Queue<ImmutableValue> unpackedValues = new LinkedList<>();
  private MsgpackIncrementalUnpacker currentUnpacker = FormatUnpacker.getInstance();
  private long currentChunkSize = 0;

  /**
   * Constructs a new {@code MsgpackStreamUnpacker}.
   *
   * @param chunkSizeLimit the allowable chunk size {@code feed} fails when the size of reading
   *     chunk exceeds the limit
   */
  public MsgpackStreamUnpacker(final long chunkSizeLimit) {
    this.buffer = new InfluentByteBuffer(chunkSizeLimit);
    this.chunkSizeLimit = chunkSizeLimit;
  }

  /**
   * Reads buffers from a {@code ReadableByteChannel}.
   *
   * @param supplier supplier to produce ByteBuffer
   * @param channel channel
   * @throws InfluentIOException when it fails reading from the channel or the chunk size exceeds
   *     the limit
   */
  public void feed(final Supplier<ByteBuffer> supplier, final NioTcpChannel channel) {
    boolean toBeContinued = true;
    while (toBeContinued) {
      toBeContinued = buffer.feed(supplier);
      unpack(channel);
    }
  }

  // fails when the chunk size exceeds the limit
  private void unpack(final NioTcpChannel channel) {
    while (buffer.hasRemaining()) {
      try {
        currentChunkSize += buffer.remaining();
        final DecodeResult result = currentUnpacker.unpack(buffer);
        currentChunkSize -= buffer.remaining();
        if (result.isCompleted()) {
          unpackedValues.offer(result.value());
          currentUnpacker = FormatUnpacker.getInstance();
          currentChunkSize = 0;
        } else if (currentChunkSize >= chunkSizeLimit) {
          channel.close();
          throw new InfluentIOException(