Splits a ByteBuf into ByteBuf chunks.

@param whole the ByteBuf to split
@param chunkSize the maximum size of each ByteBuf chunk
@return A stream that emits chunks of the original whole ByteBuf
328210
public static Flux<ByteBuf> split(final ByteBuf whole, final int chunkSize) {
        return Flux.generate(whole::readerIndex, (readFromIndex, synchronousSync) -> {
            final int writerIndex = whole.writerIndex();
            //
            if (readFromIndex >= writerIndex) {
                synchronousSync.complete();
                return writerIndex;
            } else {
                int readSize = Math.min(writerIndex - readFromIndex, chunkSize);
                // Netty slice operation will not increment the ref count.
                //
                // Here we invoke 'retain' on each slice, since
                // consumer of the returned Flux stream is responsible for
                // releasing each chunk as it gets consumed.
                //
                synchronousSync.next(whole.slice(readFromIndex, readSize).retain());
                return readFromIndex + readSize;
            }
        });
    }
706830
public static final Bytes of(ByteBuffer bb) {
    Objects.requireNonNull(bb);
    if (bb.remaining() == 0) {
      return EMPTY;
    }
    byte[] data;
    if (bb.hasArray()) {
      data = Arrays.copyOfRange(bb.array(), bb.position() + bb.arrayOffset(),
          bb.limit() + bb.arrayOffset());
    } else {
      data = new byte[bb.remaining()];
      // duplicate so that it does not change position
      bb.duplicate().get(data);
    }
    return new Bytes(data);
  }
456236
public ByteBuffer readToByteBuffer(int max) throws IOException {
        Validate.isTrue(max >= 0, "maxSize must be 0 (unlimited) or larger");
        final boolean localCapped = max > 0; // still possibly capped in total stream
        final int bufferSize = localCapped && max < DefaultSize ? max : DefaultSize;
        final byte[] readBuffer = new byte[bufferSize];
        final ByteArrayOutputStream outStream = new ByteArrayOutputStream(bufferSize);

        int read;
        int remaining = max;

        while (true) {
            read = read(readBuffer);
            if (read == -1) break;
            if (localCapped) { // this local byteBuffer cap may be smaller than the overall maxSize (like when reading first bytes)
                if (read >= remaining) {
                    outStream.write(readBuffer, 0, remaining);
                    break;
                }
                remaining -= read;
            }
            outStream.write(readBuffer, 0, read);
        }
        return ByteBuffer.wrap(outStream.toByteArray());
    }
69096
private static byte[] toBytes(Blob b) {
        try {
            InputStream is = b.getBinaryStream();
            byte[] result = IOUtils.toByteArray(is);
            is.close();
            b.free();
            return result;
        } catch (IOException e) {
            throw new RuntimeException(e);
        } catch (SQLException e) {
            throw new SQLRuntimeException(e);
        }

    }
187936
private void sendChunkSize() {
        if (chunkSizeSent.compareAndSet(false, true)) {
            log.debug("Sending chunk size: {}", chunkSize);
            ChunkSize chunkSizeMsg = new ChunkSize(chunkSize);
            conn.getChannel((byte) 2).write(chunkSizeMsg);
        }
    }
1068378
public static Object deserialize(byte[] bytes, boolean gzipOnSerialize) {
        try {
            InputStream is = new ByteArrayInputStream(bytes);
            if (gzipOnSerialize) {
                is = new GZIPInputStream(is);
            }
            ObjectInputStream in = new ObjectInputStream(is);
            Object inObj = in.readObject();
            in.close();
            return inObj;
        } catch (IOException e) {
            throw new RuntimeException(e);
        } catch (ClassNotFoundException e) {
            throw new RuntimeException(e);
        }
    }
686192
def StreamSignedBinaryContents(blob_iterator,
                               chunk_size = 1024
                              ):
  
  all_blobs_read = False
  byte_buffer = io.BytesIO()
  while not all_blobs_read or byte_buffer.getvalue():
    while not all_blobs_read and byte_buffer.tell() < chunk_size:
      try:
        blob = next(blob_iterator)
      except StopIteration:
        all_blobs_read = True
        break
      byte_buffer.write(blob.data)
    if byte_buffer.tell() > 0:
      # Yield a chunk of the signed binary and reset the buffer to contain
      # only data that hasn't been sent yet.
      byte_buffer.seek(0)
      yield byte_buffer.read(chunk_size)
      byte_buffer = io.BytesIO(byte_buffer.read())
      byte_buffer.seek(0, io.SEEK_END)
402350
public Collection<VoidChunk> split(@NonNull VoidMessage message, int maxBytes) throws IOException {
        if (maxBytes <= 0)
            throw new ND4JIllegalStateException("MaxBytes must be > 0");

        val tempFile = ND4JFileUtils.createTempFile("messageSplitter","temp");
        val result = new ArrayList<VoidChunk>();

        try (val fos = new FileOutputStream(tempFile); val bos = new BufferedOutputStream(fos)) {
            // serializing original message to disc
            SerializationUtils.serialize(message, fos);

            val length = tempFile.length();
            int numChunks = (int) (length /  maxBytes + (length % maxBytes > 0 ? 1 : 0));
            try (val fis = new FileInputStream(tempFile); val bis = new BufferedInputStream(fis)) {
                // now we'll be reading serialized message into
                val bytes = new byte[maxBytes];
                int cnt = 0;
                int id = 0;

                while (cnt < length) {
                    val c = bis.read(bytes);

                    val tmp = Arrays.copyOf(bytes, c);

                    // FIXME: we don't really want UUID used here, it's just a placeholder for now
                    val msg = VoidChunk.builder()
                            .messageId(java.util.UUID.randomUUID().toString())
                            .originalId(message.getMessageId())
                            .chunkId(id++)
                            .numberOfChunks(numChunks)
                            .splitSize(maxBytes)
                            .payload(tmp)
                            .totalSize(length)
                            .build();

                    result.add(msg);
                    cnt += c;
                }
            }
        }

        tempFile.delete();
        return result;
    }
281660
public byte[] toBytes(final int padding) {
    // special case a single entry
    if (padding == 0 && segments.size() == 1) {
      BufferSegment seg = segments.get(0);
      if (seg.offset == 0 && seg.len == seg.buf.length) {
        return seg.buf;
      }
      return Arrays.copyOfRange(seg.buf, seg.offset, seg.offset + seg.len);
    }
    byte[] result = new byte[total_length + padding];
    int ofs = 0;
    for (BufferSegment seg : segments) {
      System.arraycopy(seg.buf, seg.offset, result, ofs, seg.len);
      ofs += seg.len;
    }
    return result;
  }
849417
public <T> Stream<T> stream( Publisher pub, int batchSize ) {
        return StreamSupport.stream(Spliterators.spliteratorUnknownSize(iterator(pub, batchSize), Spliterator.IMMUTABLE | Spliterator.NONNULL), false);
    }
488316
void flattenToSubMessageBuffer(ByteBuffer buf) throws IOException
    {
        // See Serialization Format comment above getSerializedSize().

        assert(m_items != null);
        assert(!m_items.isEmpty());

        buf.put(VoltDbMessageFactory.FRAGMENT_TASK_ID);
        super.flattenToBuffer(buf);

        // Get useful statistics for the header and optional blocks.
        short nInputDepIds = 0;
        short nOutputDepIds = 0;
        short nUnplanned = 0;
        for (FragmentData item : m_items) {
            if (item.m_inputDepId != -1) {
                // Supporting only one input dep id for now.
                nInputDepIds++;
            }
            if (item.m_outputDepId != null) {
                nOutputDepIds++;
            }
            if (item.m_fragmentPlan != null) {
                nUnplanned++;
            }
        }

        // Header block
        buf.putShort((short) m_items.size());
        buf.putShort(nUnplanned);
        buf.put(m_isFinal ? (byte) 1 : (byte) 0);
        buf.put(m_taskType);
        buf.put(m_emptyForRestart ? (byte) 1 : (byte) 0);
        buf.put(m_executedOnPreviousLeader ? (byte) 1 : (byte) 0);
        buf.put(nOutputDepIds > 0 ? (byte) 1 : (byte) 0);
        buf.put(nInputDepIds  > 0 ? (byte) 1 : (byte) 0);
        if (m_procNameToLoad != null) {
            buf.putShort((short) m_procNameToLoad.length);
            buf.put(m_procNameToLoad);
        }
        else {
            buf.putShort((short) -1);
        }
        buf.put(m_perFragmentStatsRecording ? (byte) 1 : (byte) 0);
        buf.put(m_coordinatorTask ? (byte) 1 : (byte) 0);
        // N Partition Transaction PartitionCount
        buf.put(m_nPartTxn ? (byte) 1 : (byte) 0);
        // timestamp for restarted transaction
        buf.putLong(m_restartTimestamp);

        buf.putLong(m_lastSpUniqueId);

        // Plan Hash block
        for (FragmentData item : m_items) {
            buf.put(item.m_planHash);
        }

        for (FragmentData item : m_items) {
            // write statement name
            if (item.m_stmtName == null) {
                buf.putShort((short) -1);
            }
            else {
                assert(item.m_stmtName.length <= Short.MAX_VALUE);
                buf.putShort((short) item.m_stmtName.length);
                buf.put(item.m_stmtName);
            }
        }

        // Parameter set block
        for (FragmentData item : m_items) {
            buf.putInt(item.m_parameterSet.remaining());
            buf.put(item.m_parameterSet.asReadOnlyBuffer());
        }

        // Optional output dependency ID block
        if (nOutputDepIds > 0) {
            for (FragmentData item : m_items) {
                buf.putInt(item.m_outputDepId);
            }
        }

        // Optional input dependency ID block
        if (nInputDepIds > 0) {
            for (FragmentData item : m_items) {
                if (item.m_inputDepId == -1) {
                    buf.putShort((short) 0);
                } else {
                    buf.putShort((short) 1);
                    buf.putInt(item.m_inputDepId);
                }
            }
        }

        // write procedure name
        if (m_procedureName == null) {
            buf.putShort((short) -1);
        }
        else {
            assert(m_procedureName.length <= Short.MAX_VALUE);
            buf.putShort((short) m_procedureName.length);
            buf.put(m_procedureName);
        }

        // ints for batch context
        buf.putInt(m_currentBatchIndex);

        // put byte flag for timeout value and 4 bytes integer value if specified
        if (m_batchTimeout == BatchTimeoutOverrideType.NO_TIMEOUT) {
            buf.put(BatchTimeoutOverrideType.NO_OVERRIDE_FOR_BATCH_TIMEOUT.getValue());
        } else {
            buf.put(BatchTimeoutOverrideType.HAS_OVERRIDE_FOR_BATCH_TIMEOUT.getValue());
            buf.putInt(m_batchTimeout);
        }

        buf.putShort((short) m_involvedPartitions.size());
        for (int pid : m_involvedPartitions) {
            buf.putInt(pid);
        }

        if (m_initiateTaskBuffer != null) {
            ByteBuffer dup = m_initiateTaskBuffer.duplicate();
            buf.putInt(dup.remaining());
            buf.put(dup);
        } else {
            buf.putInt(0);
        }

        // Unplanned item block
        for (short index = 0; index < m_items.size(); index++) {
            // Each unplanned item gets:
            //  - an index (2)
            //  - a size (4)
            //  - buffer for the fragment plan string
            //  - a size(4)
            //  - buffer for the statement text
            FragmentData item = m_items.get(index);
            if (item.m_fragmentPlan != null) {
                buf.putShort(index);
                buf.putInt(item.m_fragmentPlan.length);
                buf.put(item.m_fragmentPlan);

                assert(item.m_stmtText != null);
                buf.putInt(item.m_stmtText.length);
                buf.put(item.m_stmtText);
            }
        }

    }
1357853
public ByteBufMono aggregate() {
		return Mono.using(alloc::compositeBuffer,
				b -> this.reduce(b,
				                 (prev, next) -> {
				                     if (prev.refCnt() > 0) {
				                         return prev.addComponent(true, next.retain());
				                     }
				                     else {
				                         return prev;
				                     }
				                 })
				         .filter(ByteBuf::isReadable),
				b -> {
				    if (b.refCnt() > 0) {
				        b.release();
				    }
				})
				.as(ByteBufMono::new);
	}
1337441
public static ByteBuffer slice(ByteBuffer buf, int off, int len)
	{
		ByteBuffer localBuf=buf.duplicate();	// so we don't mess up the position,etc
		logger.debug("off={} len={}", off, len);
		localBuf.position(off);
		localBuf.limit(off+len);
		logger.debug("pre-slice: localBuf.position()={} localBuf.limit()={}", localBuf.position(), localBuf.limit());
		localBuf = localBuf.slice();
		logger.debug("post-slice: localBuf.position()={} localBuf.limit()={}", localBuf.position(), localBuf.limit());
		return localBuf;
	}
1111536
public static byte[] streamBytes(InputStream stream) throws IOException {

		ByteArrayOutputStream bytes = new ByteArrayOutputStream();

		copyStream(stream, bytes);

		return bytes.toByteArray();
	}
567791
public static byte[] toBytes(ByteBuffer bb) {
    int length = bb.limit();
    byte [] result = new byte[length];
    System.arraycopy(bb.array(), bb.arrayOffset(), result, 0, length);
    return result;
  }
238131
public static ChunkFetchSuccess decode(ByteBuf buf) {
    StreamChunkId streamChunkId = StreamChunkId.decode(buf);
    buf.retain();
    NettyManagedBuffer managedBuf = new NettyManagedBuffer(buf.duplicate());
    return new ChunkFetchSuccess(streamChunkId, managedBuf);
  }
750134
@Override
    public IoBufferEx write(IoBufferEx source, IoBufferAllocatorEx<?> allocator) {
        return writeChunked(source, allocator);
    }