@@ -859,23 +859,17 @@ public void reset() {
859859
860860
861861 /**
862- * An abstract base implementation of {@link NestedMatcher} that looks for
863- * a specific delimiter in a {@link DataBuffer}.
864- * <p>
865- * Uses a thread-local buffer to scan data in chunks, reducing memory
866- * allocations and improving performance when processing large buffers.
867- * </p>
862+ * Base {@link NestedMatcher} implementation that scans a {@link DataBuffer}
863+ * for a specific delimiter.
868864 *
869- * <p>
870- * Each matcher keeps its own match state, so it is intended for
871- * single-threaded use. The thread-local buffer ensures that multiple
872- * threads can run their own matchers independently without interfering.
873- * </p>
865+ * <p>Relies on a per-instance reusable buffer to scan data in chunks,
866+ * minimizing allocations and improving performance for large or streaming data.</p>
874867 *
875- * <p>
876- * Subclasses can extend this class to add custom matching behavior while
877- * reusing the built-in delimiter tracking and scanning logic.
878- * </p>
868+ * <p>Each matcher maintains its own state and buffer, ensuring safe use
869+ * in reactive pipelines where execution may shift across threads.</p>
870+ *
871+ * <p>Subclasses may extend this class to customize matching strategies
872+ * while reusing the built-in delimiter tracking and scanning logic.</p>
879873 *
880874 * @see NestedMatcher
881875 * @see DataBuffer
@@ -886,8 +880,7 @@ private abstract static class AbstractNestedMatcher implements NestedMatcher {
886880
887881 private int matches = 0 ;
888882
889- // Thread-local chunk buffer to avoid per-call allocations
890- private static final ThreadLocal <byte []> LOCAL_BUFFER = ThreadLocal .withInitial (() -> new byte [8 * 1024 ]);
883+ private final byte [] localBuffer = new byte [8 * 1024 ]; // Reusable buffer per matcher instance
891884
892885 protected AbstractNestedMatcher (byte [] delimiter ) {
893886 this .delimiter = delimiter ;
@@ -901,33 +894,29 @@ protected int getMatches() {
901894 return this .matches ;
902895 }
903896
904- protected static void releaseLocalBuffer () {
905- LOCAL_BUFFER .remove ();
906- }
907- 908897 @ Override
909898 public int match (DataBuffer dataBuffer ) {
910899 final int readPos = dataBuffer .readPosition ();
911900 final int writePos = dataBuffer .writePosition ();
912901 final int length = writePos - readPos ;
913902
914- final byte [] delimiter0 = this .delimiter ;
915- final int delimiterLen = delimiter0 .length ;
916- final byte delimiter1 = delimiter0 [0 ];
903+ final byte [] delimiterBytes = this .delimiter ;
904+ final int delimiterLength = delimiterBytes .length ;
905+ final byte delimiterFirstByte = delimiterBytes [0 ];
917906
918- int matchIndex = this .matches ;
919- 920- final byte [] chunk = LOCAL_BUFFER .get ();
907+ final byte [] chunk = localBuffer ;
921908 final int chunkSize = Math .min (chunk .length , length );
922909
910+ int matchIndex = this .matches ;
911+ 923912 try {
924913 for (int offset = 0 ; offset < length ; offset += chunkSize ) {
925914 int currentChunkSize = Math .min (chunkSize , length - offset );
926915
927916 dataBuffer .readPosition (readPos + offset );
928917 dataBuffer .read (chunk , 0 , currentChunkSize );
929918
930- matchIndex = processChunk (chunk , currentChunkSize , delimiter0 , delimiterLen , delimiter1 , matchIndex , readPos , offset );
919+ matchIndex = processChunk (chunk , currentChunkSize , delimiterBytes , delimiterLength , delimiterFirstByte , matchIndex , readPos , offset );
931920 if (matchIndex < 0 ) {
932921 return -(matchIndex + 1 ); // found, returning actual position
933922 }
@@ -938,21 +927,20 @@ public int match(DataBuffer dataBuffer) {
938927 }
939928 finally {
940929 dataBuffer .readPosition (readPos ); // restore original position
941- releaseLocalBuffer ();
942930 }
943931 }
944932
945- private int processChunk (byte [] chunk , int currentChunkSize , byte [] delimiter0 , int delimiterLen , byte delimiter1 , int matchIndex , int readPos , int offset ) {
933+ private int processChunk (byte [] chunk , int currentChunkSize , byte [] delimiterBytes , int delimiterLen , byte delimiterFirstByte , int matchIndex , int readPos , int offset ) {
946934 int i = 0 ;
947935 while (i < currentChunkSize ) {
948936 if (matchIndex == 0 ) {
949- i = findNextCandidate (chunk , i , currentChunkSize , delimiter1 );
937+ i = findNextCandidate (chunk , i , currentChunkSize , delimiterFirstByte );
950938 if (i >= currentChunkSize ) {
951939 return matchIndex ; // no candidate in this chunk
952940 }
953941 }
954942
955- matchIndex = updateMatchIndex (chunk [i ], delimiter0 , delimiterLen , delimiter1 , matchIndex );
943+ matchIndex = updateMatchIndex (chunk [i ], delimiterBytes , delimiterLen , delimiterFirstByte , matchIndex );
956944 if (matchIndex == -1 ) {
957945 return -(readPos + offset + i + 1 ); // return found delimiter position (encoded as negative)
958946 }
@@ -961,24 +949,24 @@ private int processChunk(byte[] chunk, int currentChunkSize, byte[] delimiter0,
961949 return matchIndex ;
962950 }
963951
964- private int findNextCandidate (byte [] chunk , int start , int limit , byte delimiter1 ) {
952+ private int findNextCandidate (byte [] chunk , int start , int limit , byte delimiterFirstByte ) {
965953 int j = start ;
966- while (j < limit && chunk [j ] != delimiter1 ) {
954+ while (j < limit && chunk [j ] != delimiterFirstByte ) {
967955 j ++;
968956 }
969957 return j ;
970958 }
971959
972- private int updateMatchIndex (byte b , byte [] delimiter0 , int delimiterLen , byte delimiter1 , int matchIndex ) {
973- if (b == delimiter0 [matchIndex ]) {
960+ private int updateMatchIndex (byte b , byte [] delimiterBytes , int delimiterLen , byte delimiterFirstByte , int matchIndex ) {
961+ if (b == delimiterBytes [matchIndex ]) {
974962 matchIndex ++;
975963 if (matchIndex == delimiterLen ) {
976964 reset ();
977965 return -1 ;
978966 }
979967 }
980968 else {
981- matchIndex = (b == delimiter1 ) ? 1 : 0 ;
969+ matchIndex = (b == delimiterFirstByte ) ? 1 : 0 ;
982970 }
983971 return matchIndex ;
984972 }
0 commit comments