Pipe { public static <S extends MessageSchema<S>> void addUTF8(CharSequence source, Pipe<S> rb) { addUTF8(source, null==source? -1 : source.length(), rb); } Pipe(PipeConfig<T> config);  Pipe(PipeConfig<T> config, boolean usingHighLevelAPI); static void addPubListener(Pipe p, PipePublishListener listener); static void removePubListener(Pipe p, PipePublishListener listener); static void structRegistry(Pipe p, StructRegistry recordTypeData); static StructRegistry structRegistry(Pipe p); static void setThreadCallerLookup(ThreadBasedCallerLookup callerLookup); static long totalWrittenFragments(Pipe<?> p); static RunningStdDev totalWrittenFragmentStdDevperPass(Pipe<?> p); static void sumWrittenFragments(Pipe<?> p, long sum); void creationStack(); static boolean isInBlobFieldWrite(Pipe<S> pipe); long totalBlobBytesRead(); void openBlobFieldWrite(); void closeBlobFieldWrite(); static boolean isRateLimitedConsumer(Pipe<S> pipe); static boolean isRateLimitedProducer(Pipe<S> pipe); static long computeRateLimitConsumerDelay(Pipe<S> pipe); static long computeRateLimitProducerDelay(Pipe<S> pipe); static boolean isForSchema(Pipe<S> pipe, T schema); static boolean isForSchema(Pipe<S> pipe, Class<T> schema); static boolean isForSameSchema(Pipe<S> pipeA, Pipe<T> pipeB); static boolean isForDynamicSchema(Pipe<S> pipe); static long estBytesAllocated(Pipe<S> pipe); static String schemaName(Pipe<S> pipe); static void replayUnReleased(Pipe<S> pipe); static boolean isReplaying(Pipe<S> pipe); static void cancelReplay(Pipe<S> pipe); static int blobIndexBasePosition(Pipe<S> pipe); static void batchAllReleases(Pipe<S> pipe); static void setReleaseBatchSize(Pipe<S> pipe, int size); static void batchFollowingPublishes(Pipe<S> pipe, int count); static void setPublishBatchSize(Pipe<S> pipe, int size); static int getPublishBatchSize(Pipe<S> pipe); static int getReleaseBatchSize(Pipe<S> pipe); static void setMaxPublishBatchSize(Pipe<S> pipe); static void setMaxReleaseBatchSize(Pipe<S> pipe); static int bytesWriteBase(Pipe<S> pipe); static void markBytesWriteBase(Pipe<S> pipe); static int bytesReadBase(Pipe<S> pipe); static void markBytesReadBase(Pipe<S> pipe, int bytesConsumed); static void markBytesReadBase(Pipe<S> pipe); String toString(); PipeConfig<T> config(); static int totalPipes(); Pipe<T> initBuffers(); static void setConsumerRegulation(Pipe<S> pipe, int msgPerMs, int msgSize); static void setProducerRegulation(Pipe<S> pipe, int msgPerMs, int msgSize); static boolean isInit(Pipe<S> pipe); static boolean validateVarLength(Pipe<S> pipe, int length); void reset(); void reset(int structuredPos, int unstructuredPos); static void releaseReadsBatched(Pipe<MessageSchemaDynamic> pipe); static int copyFragment(
    		Pipe<S> source, 
    		Pipe<S> target); static Pipe[] buildPipes(PipeConfig[] configs); static Pipe<S>[] buildPipes(Pipe<S>[] pipes); static Pipe<S>[] buildPipes(int count, PipeConfig<S> commonConfig); static boolean validatePipeBlobHasDataToRead(Pipe<S> pipe, int blobPos, int length); static int addIntAsASCII(Pipe<S> output, int value); static int addRationalAsASCII(Pipe<S> pipe, long numerator, long denominator); static int addLongAsASCII(Pipe<S> output, long value); static int addLongAsUTF8(Pipe<S> digitBuffer, long length); static int addLongAsUTF8(Pipe<S> digitBuffer, int length); static Pipe<S>[][] splitPipes(int pipeCount, Pipe<S>[] pipes); static int[] splitGroups(final int groups, final int fullLen); static void writeFieldToOutputStream(Pipe<S> pipe, OutputStream out); static boolean readFieldFromInputStream(Pipe pipe, InputStream inputStream, final int byteCount); static ByteBuffer[] wrappedWritingBuffers(Pipe<S> output); static ByteBuffer[] wrappedWritingBuffers(int originalBlobPosition, Pipe<S> output); static ByteBuffer[] wrappedWritingBuffers(int originalBlobPosition,
			 				Pipe<S> output, int maxLen); static void moveBlobPointerAndRecordPosAndLength(int len, Pipe<S> output); static void moveBlobPointerAndRecordPosAndLength(int originalBlobPosition, int len, Pipe<S> output); static ByteBuffer wrappedBlobReadingRingA(Pipe<S> pipe, int meta, int len); static ByteBuffer wrappedBlobReadingRingB(Pipe<S> pipe, int meta, int len); static ByteBuffer[] wrappedReadingBuffers(Pipe<S> pipe, int meta, int len); static ByteBuffer[] wrappedReadingDirectBuffers(Pipe<S> pipe, int meta, int len); static int convertToUTF8(final char[] charSeq, final int charSeqOff, final int charSeqLength, 
    		                        final byte[] targetBuf, final int targetIdx, final int targetMask); static int convertToUTF8(final CharSequence charSeq, final int charSeqOff, final int charSeqLength, 
    		                        final byte[] targetBuf, final int targetIdx, final int targetMask); static void appendFragment(Pipe<S> input, Appendable target, int fragIdx); static ByteBuffer readBytes(Pipe<S> pipe, ByteBuffer target, int meta, int len); static DataOutputBlobWriter<?> readBytes(Pipe<S> pipe, 
    		                                                                     DataOutputBlobWriter<?> target, 
    		                                                                     int meta, int len); static DataOutputBlobWriter<?> readBytes(Pipe<S> pipe, 
    		                                                                     DataOutputBlobWriter<?> target); static void readBytes(Pipe<S> pipe, byte[] target, int targetIdx, int targetMask, int meta, int len); static A readASCII(Pipe<S> pipe, A target, int meta, int len); static A readOptionalASCII(Pipe<S> pipe, A target,
    		                                                                              int meta, int len); static void skipNextFragment(Pipe<S> pipe); static void skipNextFragment(Pipe<S> pipe, int msgIdx); static boolean isEqual(Pipe<S> pipe, CharSequence charSeq, int meta, int len); static boolean isEqual(Pipe<S> pipe, byte[] expected, int expectedPos, int meta, int len); static boolean isEqual(byte[] aBack, int aPos, int aMask, 
			                         byte[] bBack, int bPos, int bMask, int len); static A readUTF8(Pipe<S> pipe, A target, int meta, int len); static Appendable readOptionalUTF8(Pipe<S> pipe, Appendable target, int meta, int len); static void addDecimalAsASCII(
			                     int readDecimalExponent,	
			                     long readDecimalMantissa, 
			                     Pipe<S> outputRing); static int safeBlobPosAdd(int pos, long value); static int bytesOfContent(Pipe<S> ringBuffer); static void validateBatchSize(Pipe<S> pipe, int size); static int computeMaxBatchSize(Pipe<S> pipe); static int computeMaxBatchSize(Pipe<S> pipe, int mustFit); static boolean isEndOfPipe(Pipe<S> pipe, long tailPosition); static void publishEOF(Pipe<?>[] pipe); static void publishEOF(Pipe<S> pipe); static void copyBytesFromToRing(byte[] source, int sourceloc, int sourceMask, byte[] target, int targetloc, int targetMask, int length); static void copyIntsFromToRing(int[] source, int sourceloc, int sourceMask, int[] target, int targetloc, int targetMask, int length); static void copyBytesFromArrayToRing(byte[] source, final int sourceloc, byte[] target, int targetloc, int targetMask, int length); static void copyBytesFromInputStreamToRing(InputStream source, byte[] target, int targetloc, int targetMask, int length); static void addByteArray(Pipe<S> source, Pipe<T> target); @Deprecated //use the Appendables methods, Delete this static int leftConvertIntToASCII(
			Pipe<S> pipe, 
			int value, 
			int idx); @Deprecated //use the Appendables methods static int leftConvertLongToASCII(Pipe<S> pipe, long value,	int idx); static int readInt(final int[] buffer, final int mask, final long index); static int readIntSecure(int[] buffer, int mask, long index, int clearValue); static long readLong(int[] buffer, int mask, long index); static long decodeUTF8Fast(byte[] source, long posAndChar, int mask); static int copyASCIIToBytes(CharSequence source, Pipe<S> pipe); static void addASCII(CharSequence source, Pipe<S> rb); static void addASCII(
			CharSequence source, int sourceIdx, 
			int sourceCharCount, Pipe<S> pipe); static void addASCII(char[] source, int sourceIdx, 
			                                               int sourceCharCount, Pipe<S> pipe); static int copyASCIIToBytes(CharSequence source, int sourceIdx, final int sourceLen, Pipe<S> pipe); static int copyASCIIToBytes(char[] source, 
    		                      int sourceIdx, final int sourceLen, Pipe<S> pipe); static void addUTF8(CharSequence source, Pipe<S> rb); static void addUTF8(CharSequence source, int sourceCharCount, Pipe<S> pipe); static void addUTF8(char[] source, int sourceCharCount, Pipe<S> rb); static int copyUTF8ToByte(CharSequence source, int sourceOffset, int sourceCharCount, Pipe<S> pipe); static int copyUTF8ToByte(CharSequence source, int sourceIdx, byte[] target, int targetMask, int targetIdx, int charCount); static void xorRandomToBytes(Random r, byte[] target, int targetIdx, int count, int targetMask); static void xorBytesToBytes(byte[] source, int sourceIdx, int sourceMask,
			                            byte[] target, int targetIdx, int targetMask, 
			                            int count); static int copyUTF8ToByte(char[] source, int sourceCharCount, 
			                                                      Pipe<S> pipe); static int copyUTF8ToByte(char[] source, int sourceOffset, int sourceCharCount, Pipe<S> pipe); static int encodeSingleChar(int c, byte[] buffer,int mask, int pos); static void addByteBuffer(ByteBuffer source, Pipe<S> pipe); static void addByteBuffer(ByteBuffer source, int length, Pipe<S> pipe); static void copyByteBuffer(ByteBuffer source, int length, 
			                                                       Pipe<S> pipe); static void addByteArrayWithMask(final Pipe<S> outputRing, 
												int mask, int len, byte[] data, int offset); static void setByteArrayWithMask(final Pipe<S> outputRing,
    		int mask, int len, byte[] data, int offset, long slabPosition); static int peek(int[] buf, long pos, int mask); static long peekLong(int[] buf, long pos, int mask); static boolean isShutdown(Pipe<S> pipe); static void shutdown(Pipe<S> pipe); static void addByteArray(byte[] source, Pipe<S> pipe); static void addByteArray(byte[] source, int sourceIdx, int sourceLen, Pipe<S> pipe); static void addByteArray(byte[] source, int sourceIdx, int sourceLen, int sourceMask, Pipe<S> pipe); static void addNullByteArray(Pipe<S> pipe); static void addIntValue(int value, Pipe<S> pipe); static void setIntValue(int value, Pipe<S> pipe, long position); static void orIntValue(int value, Pipe<S> pipe, long position); static int addMsgIdx(final Pipe<S> pipe, int msgIdx); static void setValue(int[] buffer, int mask, long offset, int value); static void orValue(int[] buffer, int mask, long offset, int value); static void addBytePosAndLen(Pipe<S> pipe, int position, int length); static void addBytePosAndLenSpecial(Pipe<S> targetOutput, final int startBytePos, int bytesLength); static int bytePosition(int meta, Pipe<S> pipe, int len); static int convertToPosition(int meta, Pipe<S> pipe); static void addValue(int[] buffer, int mask, PaddedLong headCache, int value1, int value2, int value3); static void addDecimal(int exponent, long mantissa, Pipe<S> pipe); static void addLongValue(long value, Pipe<S> pipe); static void addLongValue(
    		int[] buffer, int mask, PaddedLong headCache, long value); static int readByteArrayLength(int idx, Pipe<S> pipe); static int takeByteArrayLength(Pipe<S> pipe); @Deprecated static int takeRingByteLen(Pipe<S> pipe); static byte[] byteBackingArray(int meta, Pipe<S> pipe); @Deprecated static int readRingByteMetaData(int pos, Pipe<S> pipe); static int readByteArraMetaData(int pos, Pipe<S> pipe); @Deprecated static int takeRingByteMetaData(Pipe<S> pipe); static int takeByteArrayMetaData(Pipe<S> pipe); static int readIntValue(int idx, Pipe<S> pipe); @Deprecated static int readValue(int idx, Pipe<S> pipe); static int takeInt(Pipe<S> pipe); @Deprecated //use takeInt static int takeValue(Pipe<S> pipe); static int readIntValue(final int[] backing, final int mask, final long position); @Deprecated static int readValue(int[] rbB, int rbMask, long rbPos); static Integer takeOptionalValue(Pipe<S> pipe); static Integer takeOptionalValue(Pipe<S> pipe, int absent32Value); static long takeLong(Pipe<S> pipe); static Long takeOptionalLong(Pipe<S> pipe); static Long takeOptionalLong(Pipe<S> pipe, long absent64Value); static long readLong(int idx, Pipe<S> pipe); static int takeMsgIdx(final Pipe<S> pipe); static boolean peekMsg(Pipe<S> pipe, int expected); static boolean peekNotMsg(Pipe<S> pipe, int expected); static boolean peekMsg(Pipe<S> pipe, int expected1, int expected2); static boolean peekMsg(Pipe<S> pipe, int expected1, int expected2, int expected3); static boolean peekMsg(Pipe<S> pipe, int expected1, int expected2, int expected3, int expected4); static int peekInt(Pipe<S> pipe); static int peekInt(Pipe<S> pipe, int offset); static long peekLong(Pipe<S> pipe, int offset); static A peekUTF8(Pipe<S> pipe, int offset, A target); static int contentRemaining(Pipe<S> pipe); static boolean isEmpty(Pipe<?> pipe); static int releaseReadLock(Pipe<S> pipe); static int readNextWithoutReleasingReadLock(Pipe<S> pipe); @Deprecated //inline and use releaseReadLock(pipe) static int releaseReads(Pipe<S> pipe); static void batchedReleasePublish(Pipe<S> pipe, int blobTail, long slabTail); static void releaseAllBatchedReads(Pipe<S> pipe); static void releaseBatchedReadReleasesUpToThisPosition(Pipe<S> pipe); static void releaseBatchedReadReleasesUpToPosition(Pipe<S> pipe, long newTailToPublish,  int newTailBytesToPublish); static int publishWrites(Pipe<S> pipe); static ByteBuffer[] immutableDirectBuffer(Pipe<S> pipe); static int publishWritesDirect(Pipe<S> pipe); static int writeTrailingCountOfBytesConsumed(Pipe<S> pipe); static int publishWrites(Pipe<S> pipe, int optionalHiddenTrailingBytes); static void notifyPubListener(Pipe<S> pipe); static void publishWritesBatched(Pipe<S> pipe); static boolean validateFieldCount(Pipe<S> pipe); static void publishAllBatchedWrites(Pipe<S> pipe); static void publishHeadPositions(Pipe<S> pipe); static void storeUnpublishedWrites(Pipe<S> pipe); static void abandonWrites(Pipe<S> pipe); static void spinWork(Pipe<S> pipe); static int blobMask(Pipe<S> pipe); static int slabMask(Pipe<S> pipe); static long getSlabHeadPosition(Pipe<S> pipe); static long headPosition(Pipe<S> pipe); static long workingHeadPosition(Pipe<S> pipe); static void setWorkingHead(Pipe<S> pipe, long value); static long addAndGetWorkingHead(Pipe<S> pipe, int inc); static long getWorkingTailPosition(Pipe<S> pipe); static void setWorkingTailPosition(Pipe<S> pipe, long value); static long addAndGetWorkingTail(Pipe<S> pipe, int inc); static boolean tryReplication(Pipe<S> pipe, 
            final long historicSlabPosition, 
            final int historicBlobPosition); static void publishWorkingHeadPosition(Pipe<S> pipe, long workingHeadPos); static long tailPosition(Pipe<S> pipe); static void publishWorkingTailPosition(Pipe<S> pipe, long workingTailPos); static void publishBlobWorkingTailPosition(Pipe<S> pipe, int blobWorkingTailPos); @Deprecated static int primarySize(Pipe<S> pipe); static FieldReferenceOffsetManager from(Pipe<S> pipe); static int computeCountOfBytesConsumed(Pipe<S> pipe); static IntBuffer wrappedSlabRing(Pipe<S> pipe); static ByteBuffer wrappedBlobRingA(Pipe<S> pipe); static ByteBuffer wrappedBlobRingB(Pipe<S> pipe); static ByteBuffer wrappedBlobConstBuffer(Pipe<S> pipe); static DataOutputBlobWriter<S> outputStream(Pipe<S> pipe); static DataOutputBlobWriter<S> openOutputStream(Pipe<S> pipe); static DataInputBlobReader<S> inputStream(Pipe<S> pipe); static DataInputBlobReader<S> openInputStream(Pipe<S> pipe); static DataInputBlobReader<S> peekInputStream(Pipe<S> pipe, int offset); static boolean hasRoomForWrite(Pipe<S> pipe, int size); static void presumeRoomForWrite(Pipe<S> pipe); static boolean hasRoomForWrite(Pipe<S> pipe); static long confirmLowLevelWrite(Pipe<S> output, int size); static long confirmLowLevelWriteUnchecked(Pipe<S> output, int size); static long confirmLowLevelWrite(Pipe<S> output); static int lastConfirmedWritePosition(Pipe<S> output); static boolean hasContentToRead(Pipe<S> pipe, int size); static boolean hasContentToRead(Pipe<?> pipe); static long confirmLowLevelRead(Pipe<S> pipe, long size); static int getBlobTailPosition(Pipe<S> pipe); @Deprecated static int getBlobRingTailPosition(Pipe<S> pipe); static void setBlobTailPosition(Pipe<S> pipe, int value); @Deprecated static void setBytesTail(Pipe<S> pipe, int value); static int getBlobHeadPosition(Pipe<S> pipe); static void setBlobHeadPosition(Pipe<S> pipe, int value); static int addAndGetBlobHeadPosition(Pipe<S> pipe, int inc); static int getWorkingBlobTailPosition(Pipe<S> pipe); static int addAndGetBlobWorkingTailPosition(Pipe<S> pipe, int inc); static void setBlobWorkingTail(Pipe<S> pipe, int value); static int getWorkingBlobHeadPosition(Pipe<S> pipe); @Deprecated static int getBlobWorkingHeadPosition(Pipe<S> pipe); static int addAndGetBlobWorkingHeadPosition(Pipe<S> pipe, int inc); static void setBlobWorkingHead(Pipe<S> pipe, int value); static int decBatchRelease(Pipe<S> pipe); static int decBatchPublish(Pipe<S> pipe); static void beginNewReleaseBatch(Pipe<S> pipe); static void beginNewPublishBatch(Pipe<S> pipe); static byte[] blob(Pipe<S> pipe); static int[] slab(Pipe<S> pipe); @Deprecated static byte[] byteBuffer(Pipe<S> pipe); @Deprecated static int[] primaryBuffer(Pipe<S> pipe); static PaddedLong getWorkingTailPositionObject(Pipe<S> pipe); static PaddedLong getWorkingHeadPositionObject(Pipe<S> pipe); static int sizeOf(Pipe<S> pipe, int msgIdx); static int sizeOf(S schema, int msgIdx); static void releasePendingReadLock(Pipe<S> pipe); static void releasePendingAsReadLock(Pipe<S> pipe, int consumed); static int releasePendingCount(Pipe<S> pipe); static int releasePendingByteCount(Pipe<S> pipe); static void releaseAllPendingReadLock(Pipe<S> pipe); static void markHead(Pipe pipe); static void resetHead(Pipe pipe); static void markTail(Pipe pipe); static void resetTail(Pipe pipe); static int storeBlobWorkingHeadPosition(Pipe<?> target); static int unstoreBlobWorkingHeadPosition(Pipe<?> target); static void markConsumerPassDone(Pipe<?> target); static int showPipesCreatedLargerThan; static final int RELATIVE_POS_MASK; static final int STRUCTURED_POS_MASK; static final int BYTES_WRAP_MASK; static final int EOF_SIZE; final int id; final int sizeOfSlabRing; final int sizeOfBlobRing; final int slabMask; final int blobMask; final byte bitsOfSlabRing; final byte bitsOfBlogRing; final int maxVarLen; public byte[] blobRing; public int lastMsgIdx; }
@Test public void addUTF8test() { Pipe<TestDataSchema> p = TestDataSchema.instance.newPipe(5, 100); p.initBuffers(); int val = Pipe.bytesOfContent(p); assertEquals(0, val); int msgsize = Pipe.addMsgIdx(p, TestDataSchema.MSG_CHUNKEDSTREAM_10); assertEquals(Pipe.sizeOf(TestDataSchema.instance, TestDataSchema.MSG_CHUNKEDSTREAM_10), msgsize); CharSequence x = "#$ԅԔxxx"; Pipe.addUTF8(x, p); Pipe.confirmLowLevelWrite(p, msgsize); Pipe.publishWrites(p); byte[] arr = { 1, 2, 3 }; int msgidx = Pipe.takeMsgIdx(p); assertEquals(msgidx, TestDataSchema.MSG_CHUNKEDSTREAM_10); StringBuilder str = new StringBuilder(); int meta = Pipe.takeByteArrayMetaData(p); int length = Pipe.takeByteArrayLength(p); Pipe.readUTF8(p, str, meta, length); assertEquals("#$ԅԔxxx", str.toString()); Pipe.confirmLowLevelRead(p, msgsize); Pipe.releaseReadLock(p); }