more performant boundary check (#799)

* more performant boundary check

* try to fix tests

* Update BodyPartParser.scala

* Update BodyPartParser.scala

* Update BodyPartParser.scala

* Update BodyPartParser.scala

* add HttpConstants

Author: pjfanning

http-core/src/main/scala/org/apache/pekko/http/impl/engine/parsing/BodyPartParser.scala

@@ -16,18 +16,17 @@ package org.apache.pekko.http.impl.engine.parsing
 import org.apache.pekko
 import pekko.NotUsed
 import pekko.annotation.InternalApi
-
-import scala.annotation.tailrec
 import pekko.event.LoggingAdapter
-import org.parboiled2.CharPredicate
+import pekko.http.impl.util._
+import pekko.http.scaladsl.model._
+import pekko.http.scaladsl.model.headers._
+import pekko.stream.{ Attributes, FlowShape, Inlet, Outlet }
 import pekko.stream.scaladsl.Source
 import pekko.stream.stage._
 import pekko.util.ByteString
-import pekko.http.scaladsl.model._
-import pekko.http.impl.util._
-import pekko.stream.{ Attributes, FlowShape, Inlet, Outlet }
-import pekko.http.scaladsl.model.headers._
+import org.parboiled2.CharPredicate
 
+import scala.annotation.tailrec
 import scala.collection.mutable.ListBuffer
 
 /**
@@ -355,15 +354,38 @@ private[http] object BodyPartParser {
   }
 
   case class UndefinedEndOfLineConfiguration(boundary: String) extends EndOfLineConfiguration {
+    import HttpConstants._
+
     override def eol: String = "\r\n"
 
     override def defineOnce(byteString: ByteString): EndOfLineConfiguration = {
       // Hypothesis: There is either CRLF or LF as EOL, no mix possible
-      val crLfNeedle = ByteString(s"$boundary\r\n")
-      val lfNeedle = ByteString(s"$boundary\n")
-      if (byteString.containsSlice(crLfNeedle)) DefinedEndOfLineConfiguration("\r\n", boundary)
-      else if (byteString.containsSlice(lfNeedle)) DefinedEndOfLineConfiguration("\n", boundary)
-      else this
+      checkForBoundary(byteString) match {
+        case CR_BYTE => DefinedEndOfLineConfiguration("\r\n", boundary)
+        case LF_BYTE => DefinedEndOfLineConfiguration("\n", boundary)
+        case _       => this
+      }
+    }
+
+    // returns CR for CRLF, LF for LF, 0 otherwise
+    private def checkForBoundary(byteString: ByteString): Byte = {
+      val check = ByteString(boundary)
+      @tailrec def findBoundary(offset: Int): Byte = {
+        val index = byteString.indexOfSlice(check, offset)
+        if (index != -1) {
+          val newIndex = index + boundary.length
+          byteAt(byteString, newIndex) match {
+            case CR_BYTE =>
+              if (byteAt(byteString, newIndex + 1) == LF_BYTE) CR_BYTE else findBoundary(index + 1)
+            case LF_BYTE => LF_BYTE
+            case _       => findBoundary(index + 1)
+          }
+        } else 0
+      }
+      try findBoundary(0)
+      catch {
+        case NotEnoughDataException => 0
+      }
     }
   }
 }

http-core/src/main/scala/org/apache/pekko/http/impl/util/HttpConstants.scala

@@ -0,0 +1,32 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pekko.http.impl.util
+
+import org.apache.pekko.annotation.InternalApi
+
+/**
+ * INTERNAL API
+ *
+ * This object contains HTTP related constants that are used in various places.
+ * It is not intended to be used outside of the HTTP implementation.
+ */
+@InternalApi
+private[http] object HttpConstants {
+  final val CR_BYTE: Byte = 13
+  final val LF_BYTE: Byte = 10
+}