Merge "JERSEY-3015: SSE is able to parse CRLFCRLF chunks."

Author: pavelbucek

core-client/src/main/java/org/glassfish/jersey/client/ChunkedInput.java

@@ -46,8 +46,11 @@
 import java.io.InputStream;
 import java.lang.annotation.Annotation;
 import java.lang.reflect.Type;
+import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
+import java.util.Comparator;
+import java.util.List;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.logging.Level;
 import java.util.logging.Logger;
@@ -107,34 +110,41 @@ public static ChunkParser createParser(final byte[] boundary) {
         return new FixedBoundaryParser(boundary);
     }
 
-    private static class FixedBoundaryParser implements ChunkParser {
-
-        private final byte[] delimiter;
+    /**
+     * Create a new chunk multi-parser that will split the response entity input stream
+     * based on multiple fixed boundary strings.
+     *
+     * @param boundaries chunk boundaries.
+     * @return new fixed boundary string-based chunk parser.
+     */
+    public static ChunkParser createMultiParser(final String... boundaries) {
+        return new FixedMultiBoundaryParser(boundaries);
+    }
 
-        public FixedBoundaryParser(final byte[] boundary) {
-            delimiter = Arrays.copyOf(boundary, boundary.length);
-        }
+    private abstract static class AbstractBoundaryParser implements ChunkParser {
 
         @Override
         public byte[] readChunk(final InputStream in) throws IOException {
             final ByteArrayOutputStream buffer = new ByteArrayOutputStream();
-            byte[] delimiterBuffer = new byte[delimiter.length];
+            byte[] delimiterBuffer = new byte[getDelimiterBufferSize()];
 
             int data;
             int dPos;
             do {
                 dPos = 0;
                 while ((data = in.read()) != -1) {
                     final byte b = (byte) data;
+                    byte[] delimiter = getDelimiter(b, dPos, delimiterBuffer);
 
                     // last read byte is part of the chunk delimiter
-                    if (b == delimiter[dPos]) {
+                    if (delimiter != null && b == delimiter[dPos]) {
                         delimiterBuffer[dPos++] = b;
                         if (dPos == delimiter.length) {
                             // found chunk delimiter
                             break;
                         }
                     } else if (dPos > 0) {
+                        delimiter = getDelimiter(dPos - 1, delimiterBuffer);
                         delimiterBuffer[dPos] = b;
 
                         int matched = matchTail(delimiterBuffer, 1, dPos, delimiter);
@@ -159,14 +169,44 @@ public byte[] readChunk(final InputStream in) throws IOException {
 
             } while (data != -1 && buffer.size() == 0); // skip an empty chunk
 
-            if (dPos > 0 && dPos != delimiter.length) {
+            if (dPos > 0 && dPos != getDelimiter(dPos - 1, delimiterBuffer).length) {
                 // flush the delimiter buffer, if not empty - parsing finished in the middle of a potential delimiter sequence
                 buffer.write(delimiterBuffer, 0, dPos);
             }
 
             return (buffer.size() > 0) ? buffer.toByteArray() : null;
         }
 
+        /**
+         * Selects a delimiter which corresponds to delimiter buffer. Method automatically appends {@code b} param on the
+         * {@code pos} position of {@code delimiterBuffer} array and then starts the selection process with a newly created array.
+         *
+         * @param b               byte which will be added on the {@code pos} position of {@code delimiterBuffer} array
+         * @param pos             number of bytes from the delimiter buffer which will be used in processing
+         * @param delimiterBuffer current content of the delimiter buffer
+         * @return delimiter which corresponds to delimiterBuffer
+         */
+        abstract byte[] getDelimiter(byte b, int pos, byte[] delimiterBuffer);
+
+        /**
+         * Selects a delimiter which corresponds to delimiter buffer.
+         *
+         * @param pos             position of the last read byte
+         * @param delimiterBuffer number of bytes from the delimiter buffer which will be used in processing
+         * @return delimiter which corresponds to delimiterBuffer
+         */
+        abstract byte[] getDelimiter(int pos, byte[] delimiterBuffer);
+
+        /**
+         * Returns a delimiter buffer size depending on the selected strategy.
+         * <p>
+         * If a strategy has multiple registered delimiters, then the delimiter buffer should be a length of the longest
+         * delimiter.
+         *
+         * @return length of the delimiter buffer
+         */
+        abstract int getDelimiterBufferSize();
+
         /**
          * Tries to find an element intersection between two arrays in a way that intersecting elements must be
          * at the tail of the first array and at the beginning of the second array.
@@ -192,6 +232,10 @@ public byte[] readChunk(final InputStream in) throws IOException {
          * any part of the head of the pattern, otherwise returns number of overlapping elements.
          */
         private static int matchTail(byte[] buffer, int offset, int length, byte[] pattern) {
+            if (pattern == null) {
+                return 0;
+            }
+
             outer:
             for (int i = 0; i < length; i++) {
                 final int tailLength = length - i;
@@ -209,6 +253,87 @@ private static int matchTail(byte[] buffer, int offset, int length, byte[] patte
         }
     }
 
+    private static class FixedBoundaryParser extends AbstractBoundaryParser {
+
+        private final byte[] delimiter;
+
+        public FixedBoundaryParser(final byte[] boundary) {
+            delimiter = Arrays.copyOf(boundary, boundary.length);
+        }
+
+        @Override
+        byte[] getDelimiter(byte b, int pos, byte[] delimiterBuffer) {
+            return delimiter;
+        }
+
+        @Override
+        byte[] getDelimiter(int pos, byte[] delimiterBuffer) {
+            return delimiter;
+        }
+
+        @Override
+        int getDelimiterBufferSize() {
+            return delimiter.length;
+        }
+    }
+
+    private static class FixedMultiBoundaryParser extends AbstractBoundaryParser {
+
+        private final List<byte[]> delimiters = new ArrayList<byte[]>();
+
+        private final int longestDelimiterLength;
+
+        public FixedMultiBoundaryParser(String... boundaries) {
+            for (String boundary: boundaries) {
+                byte[] boundaryBytes = boundary.getBytes();
+                delimiters.add(Arrays.copyOf(boundaryBytes, boundaryBytes.length));
+            }
+
+            Collections.sort(delimiters, new Comparator<byte[]>() {
+                @Override
+                public int compare(byte[] o1, byte[] o2) {
+                    return Integer.compare(o1.length, o2.length);
+                }
+            });
+
+            byte[] longestDelimiter = delimiters.get(delimiters.size() - 1);
+            this.longestDelimiterLength = longestDelimiter.length;
+        }
+
+        @Override
+        byte[] getDelimiter(byte b, int pos, byte[] delimiterBuffer) {
+            byte[] buffer = Arrays.copyOf(delimiterBuffer, delimiterBuffer.length);
+            buffer[pos] = b;
+
+            return getDelimiter(pos, buffer);
+        }
+
+        @Override
+        byte[] getDelimiter(int pos, byte[] delimiterBuffer) {
+            outer:
+            for (byte[] delimiter: delimiters) {
+                if (pos > delimiter.length) {
+                    continue;
+                }
+
+                for (int i = 0; i <= pos && i < delimiter.length; i++) {
+                    if (delimiter[i] != delimiterBuffer[i]) {
+                        continue outer;
+                    } else if (pos == i) {
+                        return delimiter;
+                    }
+                }
+            }
+
+            return null;
+        }
+
+        @Override
+        int getDelimiterBufferSize() {
+            return this.longestDelimiterLength;
+        }
+    }
+
     /**
      * Package-private constructor used by the {@link ChunkedInputReader}.
      *

core-client/src/test/java/org/glassfish/jersey/client/ChunkedInputTest.java renamed to core-client/src/test/java/org/glassfish/jersey/client/FixedBoundaryParserTest.java

@@ -52,7 +52,7 @@
  *
  * @author Petr Bouda (petr.bouda at oracle.com)
  **/
-public class ChunkedInputTest {
+public class FixedBoundaryParserTest {
 
     public static final String DELIMITER_4 = "1234";