/**
 * Free all direct buffers if allocated, mainly used for testing.
 */
@VisibleForTesting
public void clean() {
  while (!buffers.isEmpty()) {
    ByteBuffer b = buffers.poll();
    if (b instanceof DirectBuffer) {
      DirectBuffer db = (DirectBuffer) b;
      if (db.cleaner() != null) {
        db.cleaner().clean();
      }
    }
  }
  this.usedBufCount.set(0);
  this.maxPoolSizeInfoLevelLogged = false;
  this.poolAllocationBytes.reset();
  this.heapAllocationBytes.reset();
  this.lastPoolAllocationBytes = 0;
  this.lastHeapAllocationBytes = 0;
}
 

/**
 * Handle error with body result
 * @return 
 */
@Override
public boolean handleServerResult(final BBoxDBConnection bBoxDBConnection, 
		final ByteBuffer encodedPackage, final NetworkOperationFuture future)
		throws PackageEncodeException {
	
	final AbstractBodyResponse result = ErrorResponse.decodePackage(encodedPackage);
	
	if(logger.isDebugEnabled()) {
		logger.debug("Handle error package (seq={}, message={})", result.getSequenceNumber(), result.getBody());
	}
	
	if(future != null) {
		future.setMessage(result.getBody());
		future.setFailedState();
		future.fireCompleteEvent();
	}
	
	return true;
}
 

public ByteBuffer get(int index) {
    int offset;
    if (index == 0) {
        offset = 0;
    } else {
        dataBuffer.position(indexStartOffset + ((index - 1) << 2));
        offset = dataBuffer.getInt();
    }

    ByteBuffer resultBuffer = dataBuffer.asReadOnlyBuffer();
    int startOffset = dataStartOffset + offset;
    resultBuffer.position(startOffset);
    int length = resultBuffer.getInt();
    resultBuffer.limit(startOffset + 4 + length);
    return resultBuffer;
}
 

@Test
public void testEncodeAndDecode_EmptyBody() {
    System.setProperty(RemotingCommand.REMOTING_VERSION_KEY, "2333");

    int code = 103; //org.apache.rocketmq.common.protocol.RequestCode.REGISTER_BROKER
    CommandCustomHeader header = new SampleCommandCustomHeader();
    RemotingCommand cmd = RemotingCommand.createRequestCommand(code, header);

    ByteBuffer buffer = cmd.encode();

    //Simulate buffer being read in NettyDecoder
    buffer.getInt();
    byte[] bytes = new byte[buffer.limit() - 4];
    buffer.get(bytes, 0, buffer.limit() - 4);
    buffer = ByteBuffer.wrap(bytes);

    RemotingCommand decodedCommand = RemotingCommand.decode(buffer);

    assertThat(decodedCommand.getSerializeTypeCurrentRPC()).isEqualTo(SerializeType.JSON);
    assertThat(decodedCommand.getBody()).isNull();
}
 

@Test
public void testClosedChannel() throws Throwable {
  RegularFile file = regularFile(15);
  ExecutorService executor = Executors.newSingleThreadExecutor();

  try {
    JimfsAsynchronousFileChannel channel = channel(file, executor, READ, WRITE);
    channel.close();

    assertClosed(channel.read(ByteBuffer.allocate(10), 0));
    assertClosed(channel.write(ByteBuffer.allocate(10), 15));
    assertClosed(channel.lock());
    assertClosed(channel.lock(0, 10, true));
  } finally {
    executor.shutdown();
  }
}
 

private int createTexture(ByteBuffer buf) {
    // Create a new OpenGL texture
    int textureId = glGenTextures();
    // Bind the texture
    glBindTexture(GL_TEXTURE_2D, textureId);

    // Tell OpenGL how to unpack the RGBA bytes. Each component is 1 byte size
    glPixelStorei(GL_UNPACK_ALIGNMENT, 1);

    glTexParameteri(GL_TEXTURE_2D, GL_TEXTURE_MIN_FILTER, GL_NEAREST);
    glTexParameteri(GL_TEXTURE_2D, GL_TEXTURE_MAG_FILTER, GL_NEAREST);

    // Upload the texture data
    glTexImage2D(GL_TEXTURE_2D, 0, GL_RGBA, width, height, 0,
        GL_RGBA, GL_UNSIGNED_BYTE, buf);
    // Generate Mip Map
    glGenerateMipmap(GL_TEXTURE_2D);

    return textureId;
}
 

private static List<ByteBuffer> fetchCycleOrder(List<ManagedVerifier> managedVerifiers) {

        List<ByteBuffer> cycleOrder = new CopyOnWriteArrayList<>();
        for (int i = 0; i < managedVerifiers.size() && cycleOrder.isEmpty(); i++) {
            Message meshRequest = new Message(MessageType.BootstrapRequestV2_35, new BootstrapRequest());
            ManagedVerifier verifier = managedVerifiers.get(i);
            AtomicBoolean processedResponse = new AtomicBoolean(false);
            Message.fetchTcp(verifier.getHost(), verifier.getPort(), meshRequest, new MessageCallback() {
                @Override
                public void responseReceived(Message message) {
                    if (message != null && (message.getContent() instanceof BootstrapResponseV2)) {
                        BootstrapResponseV2 response = (BootstrapResponseV2) message.getContent();
                        cycleOrder.addAll(response.getCycleVerifiers());
                    }
                    processedResponse.set(true);
                }
            });

            // Wait up to 2 seconds for the response to be processed.
            for (int j = 0; j < 10 && !processedResponse.get(); j++) {
                ThreadUtil.sleep(200L);
            }
        }

        return cycleOrder;
    }
 

public static void copy(InputStream aIS, File aTargetFile) throws IOException
{
    aTargetFile.getParentFile().mkdirs();
    
    try (
            ReadableByteChannel in = newChannel(aIS);
            WritableByteChannel out = newChannel(new FileOutputStream(aTargetFile))
    ) {
        final ByteBuffer buffer = allocateDirect(8192);
        while (in.read(buffer) != -1) {
            buffer.flip();
            out.write(buffer);
            buffer.compact();
        }
        buffer.flip();
        while (buffer.hasRemaining()) {
            out.write(buffer);
        }
    }
}
 

public static void main(String[] args) throws IOException {
    ReadableByteChannel channel = new ReadableByteChannel() {
        public int read(ByteBuffer dst) {
            dst.put((byte) 129);
            return 1;
        }

        public boolean isOpen() {
            return true;
        }

        public void close() {
        }
    };

    InputStream in = Channels.newInputStream(channel);
    int data = in.read();
    if (data < 0)
        throw new RuntimeException(
            "InputStream.read() spec'd to return 0-255");
}
 

private static Object parseAnnotationArray(int length,
                                           Class<? extends Annotation> annotationType,
                                           ByteBuffer buf,
                                           ConstantPool constPool,
                                           Class<?> container) {
    Object[] result = (Object[]) Array.newInstance(annotationType, length);
    boolean typeMismatch = false;
    int tag = 0;

    for (int i = 0; i < length; i++) {
        tag = buf.get();
        if (tag == '@') {
            result[i] = parseAnnotation(buf, constPool, container, true);
        } else {
            skipMemberValue(tag, buf);
            typeMismatch = true;
        }
    }
    return typeMismatch ? exceptionProxy(tag) : result;
}
 

/**
 * {@inheritDoc}.
 */
@Override
public FieldValue parse(final ParadoxTable table, final ByteBuffer buffer, final ParadoxField field) {
    final ByteBuffer bytes = ByteBuffer.allocate(field.getSize());

    // Track for NULL values.
    boolean allZeroes = true;
    for (int chars = 0; chars < field.getSize(); chars++) {
        byte value = buffer.get();
        bytes.put(value);

        if (value != 0) {
            allZeroes = false;
        }
    }

    if (allZeroes) {
        return NULL;
    }

    return new FieldValue(bytes.array(), ParadoxFieldType.BYTES.getSQLType());
}
 

public int write(ByteBuffer src, long position) throws IOException {
    if (src == null)
        throw new NullPointerException();
    if (position < 0)
        throw new IllegalArgumentException("Negative position");
    if (!writable)
        throw new NonWritableChannelException();
    ensureOpen();
    if (nd.needsPositionLock()) {
        synchronized (positionLock) {
            return writeInternal(src, position);
        }
    } else {
        return writeInternal(src, position);
    }
}
 

public int write(ByteBuffer src) throws IOException {
    int len = src.remaining();
    int totalWritten = 0;
    synchronized (writeLock) {
        while (totalWritten < len) {
            int bytesToWrite = Math.min((len - totalWritten),
                                        TRANSFER_SIZE);
            if (buf.length < bytesToWrite)
                buf = new byte[bytesToWrite];
            src.get(buf, 0, bytesToWrite);
            try {
                begin();
                out.write(buf, 0, bytesToWrite);
            } finally {
                end(bytesToWrite > 0);
            }
            totalWritten += bytesToWrite;
        }
        return totalWritten;
    }
}
 

/***************************************
 * {@inheritDoc}
 */
@Override
protected void performBlockingOperation(
	AsynchronousFileChannel aChannel,
	ByteBuffer				rData) throws InterruptedException,
										  ExecutionException
{
	long nPosition = 0;

	while (rData.hasRemaining())
	{
		nPosition += aChannel.write(rData, nPosition).get();
	}

	rData.clear();
}
 

private static HadoopFsRecoverable deserializeV1(byte[] serialized) throws IOException {
	final ByteBuffer bb = ByteBuffer.wrap(serialized).order(ByteOrder.LITTLE_ENDIAN);

	if (bb.getInt() != MAGIC_NUMBER) {
		throw new IOException("Corrupt data: Unexpected magic number.");
	}

	final long offset = bb.getLong();
	final byte[] targetFileBytes = new byte[bb.getInt()];
	final byte[] tempFileBytes = new byte[bb.getInt()];
	bb.get(targetFileBytes);
	bb.get(tempFileBytes);

	final String targetPath = new String(targetFileBytes, CHARSET);
	final String tempPath = new String(tempFileBytes, CHARSET);

	return new HadoopFsRecoverable(new Path(targetPath), new Path(tempPath), offset);

}
 

static boolean compare(ByteBuffer bb, String str) {
    try{
        return Util.compare(bb, str.getBytes("ISO-8859-1"));
    } catch (UnsupportedEncodingException unsupported) {
        throw new AssertionError(unsupported);
    }
}
 

private ByteBuffer processAsciiCommand(ByteBuffer buffer, Cache cache) {
  CharBuffer flb = getFirstLineBuffer();
  getAsciiDecoder().reset();
  getAsciiDecoder().decode(buffer, flb, false);
  flb.flip();
  String firstLine = getFirstLine();
  String[] firstLineElements = firstLine.split(" ");
  
  assert "flush_all".equals(stripNewline(firstLineElements[0]));
  boolean noReply = false;
  int delay = 0;
  if (firstLineElements.length == 2) {
    if ("noreply".equals(stripNewline(firstLineElements[1]))) {
      noReply = true;
    } else {
      delay = Integer.parseInt(stripNewline(firstLineElements[1]));
    }
  } else if (firstLineElements.length == 3) {
    delay = Integer.parseInt(stripNewline(firstLineElements[1]));
    noReply = true;
  }
  
  final Region<Object, ValueWrapper> r = getMemcachedRegion(cache);
  if (delay == 0) {
    r.destroyRegion();
  } else {
    StorageCommand.getExpiryExecutor().schedule(new Runnable() {
      public void run() {
        r.destroyRegion();
      }
    }, delay, TimeUnit.SECONDS);
  }
  
  CharBuffer retVal = CharBuffer.wrap(Reply.OK.toString());
  
  return noReply ? null : asciiCharset.encode(retVal);
}
 

/**
 * Read an integer stored in variable-length format using zig-zag decoding from
 * <a href="http://code.google.com/apis/protocolbuffers/docs/encoding.html"> Google Protocol Buffers</a>.
 *
 * @param buffer The buffer to read from
 * @return The integer read
 *
 * @throws IllegalArgumentException if variable-length value does not terminate after 5 bytes have been read
 */
public static int readVarint(ByteBuffer buffer) {
    int value = 0;
    int i = 0;
    int b;
    while (((b = buffer.get()) & 0x80) != 0) {
        value |= (b & 0x7f) << i;
        i += 7;
        if (i > 28)
            throw illegalVarintException(value);
    }
    value |= b << i;
    return (value >>> 1) ^ -(value & 1);
}
 

public ByteBuffer serialize() {
	DataWriter data = new DataWriter(LENGTH);
	data.add(TYPE);
	data.add(this.peerIdentifier);
	data.add(this.connectionPurpose.toRaw());
	return data.getData();
}
 

/** {@inheritDoc} */
@Override public boolean readFrom(ByteBuffer buf, MessageReader reader) {
    if (!super.readFrom(buf, reader))
        return false;

    if (buf.remaining() < 4)
        return false;

    connIdx = buf.getInt();

    return true;
}
 

/**
 * Calculates the key of a JSON value by Base 64 encoding the first 64 bits
 * of the SHA-256 digest of the JSON's string representation.
 *
 * @param json
 *            the JSON to get a hash of, not <code>null</code>
 * @return the key uniquely identifying the given JSON value
 */
private static String calculateHash(JsonValue json) {
    byte[] digest = MessageDigestUtil.sha256(json.toJson());

    /*
     * Only use first 64 bits to keep id string short (1 in 100 000 000
     * collision risk with 500 000 items). 64 bits base64 -> 11 ASCII chars
     */
    ByteBuffer truncatedDigest = ByteBuffer.wrap(digest, 0, 8);

    ByteBuffer base64Bytes = Base64.getEncoder().encode(truncatedDigest);

    return StandardCharsets.US_ASCII.decode(base64Bytes).toString();
}
 

@Override
public UnboundPredicate<ByteBuffer> project(String name,
                                            BoundPredicate<ByteBuffer> pred) {
  if (pred.op() == NOT_NULL || pred.op() == IS_NULL) {
    return Expressions.predicate(pred.op(), name);
  }
  return ProjectionUtil.truncateArray(name, pred, this);
}
 

static void increaseMajor(Path cfile, int delta) {
   try (RandomAccessFile cls =
        new RandomAccessFile(cfile.toFile(), "rw");
        FileChannel fc = cls.getChannel()) {
       ByteBuffer rbuf = ByteBuffer.allocate(2);
       fc.read(rbuf, 6);
       ByteBuffer wbuf = ByteBuffer.allocate(2);
       wbuf.putShort(0, (short)(rbuf.getShort(0) + delta));
       fc.write(wbuf, 6);
       fc.force(false);
   } catch (Exception e){
       throw new RuntimeException("Failed: unexpected exception");
   }
}
 

public static int EVP_CipherFinal_ex(final PointerByReference context, final ByteBuffer outBuffer,
        final int[] outlen) {
    if (VERSION == VERSION_1_1_X) {
        return OpenSsl11XNativeJna.EVP_CipherFinal_ex(context, outBuffer, outlen);
    }
    return OpenSsl10XNativeJna.EVP_CipherFinal_ex(context, outBuffer, outlen);
}
 

private void deserializeRecord(Record resultRecord, ByteBuffer in) {
    for (int i = 0; i < schema.getDimensionCount(); i++) {
        resultRecord.setDimension(i, BytesUtil.readUTFString(in));
    }
    for (int i = 0; i < schema.getMetricsCount(); i++) {
        resultRecord.setMetric(i, metricsSerializers[i].deserialize(in));
    }
}
 

@Test
public void testReplaceAtHeadWithUnpinningChain() {
  PinningOffHeapChainMap<Long> pinningOffHeapChainMap = getPinningOffHeapChainMap();

  ByteBuffer buffer = buffer(PUT_IF_ABSENT);
  Chain pinningChain = chainOf(buffer);
  Chain unpinningChain = chainOf(buffer(PUT));

  pinningOffHeapChainMap.append(1L, buffer);
  assertThat(pinningOffHeapChainMap.heads.isPinned(1L), is(true));

  pinningOffHeapChainMap.replaceAtHead(1L, pinningChain, unpinningChain);
  assertThat(pinningOffHeapChainMap.heads.isPinned(1L), is(false));
}
 

static boolean compare(ByteBuffer bb, String str) {
    try{
        return Util.compare(bb, str.getBytes("ISO-8859-1"));
    } catch (UnsupportedEncodingException unsupported) {
        throw new AssertionError(unsupported);
    }
}
 

/**
 * Invoked prior to write to prepare the WSABUF array. Where necessary,
 * it substitutes non-direct buffers with direct buffers.
 */
void prepareBuffers() {
    shadow = new ByteBuffer[numBufs];
    long address = writeBufferArray;
    for (int i=0; i<numBufs; i++) {
        ByteBuffer src = bufs[i];
        int pos = src.position();
        int lim = src.limit();
        assert (pos <= lim);
        int rem = (pos <= lim ? lim - pos : 0);
        long a;
        if (!(src instanceof DirectBuffer)) {
            // substitute with direct buffer
            ByteBuffer bb = Util.getTemporaryDirectBuffer(rem);
            bb.put(src);
            bb.flip();
            src.position(pos);  // leave heap buffer untouched for now
            shadow[i] = bb;
            a = ((DirectBuffer)bb).address();
        } else {
            shadow[i] = src;
            a = ((DirectBuffer)src).address() + pos;
        }
        unsafe.putAddress(address + OFFSETOF_BUF, a);
        unsafe.putInt(address + OFFSETOF_LEN, rem);
        address += SIZEOF_WSABUF;
    }
}
 

public static void main(String[] args) throws Exception {
    final String cont = RandomStringUtils.randomAlphanumeric(8 * 1024 * 1024);
    try (final FileChannel src = new RandomAccessFile(SOURCE, "rw").getChannel();
         final FileChannel dest = new RandomAccessFile(DEST, "rw").getChannel()) {
        src.write(ByteBuffer.wrap(cont.getBytes()));
        long start = System.currentTimeMillis();
        final MappedByteBuffer mmap = dest.map(FileChannel.MapMode.READ_WRITE, 0, src.size());
        src.write(mmap);
        mmap.flip();
        System.out.println("Cost: " + (System.currentTimeMillis() - start) + " ms");
    } finally {
        Files.deleteIfExists(Paths.get(SOURCE));
        Files.deleteIfExists(Paths.get(DEST));
    }
}
 

/** {@inheritDoc} */
@Override public void onMessageReceived(GridNioSession ses, Object msg) throws IgniteCheckedException {
    if (!(msg instanceof ByteBuffer))
        throw new GridNioException("Failed to decode incoming message (incoming message is not a byte buffer, " +
            "is filter properly placed?): " + msg.getClass());

    try {
        ByteBuffer input = (ByteBuffer)msg;

        while (input.hasRemaining()) {
            Object res = parser.decode(ses, input);

            if (res != null)
                proceedMessageReceived(ses, res);
            else {
                if (input.hasRemaining()) {
                    if (directMode)
                        return;

                    LT.warn(log, "Parser returned null but there are still unread data in input buffer (bug in " +
                        "parser code?) [parser=" + parser + ", ses=" + ses + ']');

                    input.position(input.limit());
                }
            }
        }
    }
    catch (IOException e) {
        throw new GridNioException(e);
    }
}