diff --git a/.gitignore b/.gitignore
index 8e716eb2d4..0d810187b9 100755
--- a/.gitignore
+++ b/.gitignore
@@ -1,4 +1,5 @@
 *#
+*.jfr
 *.iml
 *.ipr
 *.iws
diff --git a/akka-actor-tests/src/test/scala/akka/util/ByteStringSpec.scala b/akka-actor-tests/src/test/scala/akka/util/ByteStringSpec.scala
index 55751aa816..0a2894ccc0 100644
--- a/akka-actor-tests/src/test/scala/akka/util/ByteStringSpec.scala
+++ b/akka-actor-tests/src/test/scala/akka/util/ByteStringSpec.scala
@@ -10,6 +10,7 @@ import java.lang.Float.floatToRawIntBits
 import java.nio.{ ByteBuffer, ByteOrder }
 import java.nio.ByteOrder.{ BIG_ENDIAN, LITTLE_ENDIAN }
 
+import akka.util.ByteString.{ ByteString1, ByteString1C, ByteStrings }
 import org.apache.commons.codec.binary.Hex.encodeHex
 import org.scalacheck.Arbitrary.arbitrary
 import org.scalacheck.{ Arbitrary, Gen }
@@ -20,6 +21,12 @@ import scala.collection.mutable.Builder
 
 class ByteStringSpec extends WordSpec with Matchers with Checkers {
 
+  //  // uncomment when developing locally to get better coverage
+  //  implicit override val generatorDrivenConfig =
+  //    PropertyCheckConfig(
+  //      minSuccessful = 1000,
+  //      minSize = 0, maxSize = 100)
+
   def genSimpleByteString(min: Int, max: Int) = for {
     n ← Gen.choose(min, max)
     b ← Gen.containerOfN[Array, Byte](n, arbitrary[Byte])
@@ -281,10 +288,113 @@ class ByteStringSpec extends WordSpec with Matchers with Checkers {
     reference.toSeq == builder.result
   }
 
+  "ByteString1" must {
+    "drop(0)" in {
+      ByteString1.fromString("").drop(0) should ===(ByteString.empty)
+      ByteString1.fromString("a").drop(0) should ===(ByteString("a"))
+    }
+    "drop(1)" in {
+      ByteString1.fromString("").drop(1) should ===(ByteString(""))
+      ByteString1.fromString("a").drop(1) should ===(ByteString(""))
+      ByteString1.fromString("ab").drop(1) should ===(ByteString("b"))
+      ByteString1.fromString("xaaa").drop(1) should ===(ByteString("aaa"))
+      ByteString1.fromString("xaab").drop(1).take(2) should ===(ByteString("aa"))
+      ByteString1.fromString("0123456789").drop(5).take(4).drop(1).take(2) should ===(ByteString("67"))
+    }
+    "drop(n)" in {
+      ByteString1.fromString("ab").drop(2) should ===(ByteString(""))
+      ByteString1.fromString("ab").drop(3) should ===(ByteString(""))
+    }
+  }
+  "ByteString1C" must {
+    "drop(0)" in {
+      ByteString1C.fromString("").drop(0) should ===(ByteString.empty)
+      ByteString1C.fromString("a").drop(0) should ===(ByteString("a"))
+    }
+    "drop(1)" in {
+      ByteString1C.fromString("").drop(1) should ===(ByteString(""))
+      ByteString1C.fromString("a").drop(1) should ===(ByteString(""))
+      ByteString1C.fromString("ab").drop(1) should ===(ByteString("b"))
+    }
+    "drop(n)" in {
+      ByteString1C.fromString("ab").drop(2) should ===(ByteString(""))
+      ByteString1C.fromString("ab").drop(3) should ===(ByteString(""))
+    }
+    "take" in {
+      ByteString1.fromString("abcdefg").drop(1).take(0) should ===(ByteString(""))
+      ByteString1.fromString("abcdefg").drop(1).take(-1) should ===(ByteString(""))
+      ByteString1.fromString("abcdefg").drop(1).take(-2) should ===(ByteString(""))
+      ByteString1.fromString("abcdefg").drop(2) should ===(ByteString("cdefg"))
+      ByteString1.fromString("abcdefg").drop(2).take(1) should ===(ByteString("c"))
+    }
+  }
+  "ByteStrings" must {
+    "drop(0)" in {
+      ByteStrings(ByteString1.fromString(""), ByteString1.fromString("")).drop(0) should ===(ByteString.empty)
+      ByteStrings(ByteString1.fromString(""), ByteString1.fromString("a")).drop(0) should ===(ByteString("a"))
+      (ByteString1C.fromString("") ++ ByteString1.fromString("a")).drop(0) should ===(ByteString("a"))
+      ByteStrings(ByteString1.fromString("a"), ByteString1.fromString("")).drop(0) should ===(ByteString("a"))
+      ByteStrings(ByteString1.fromString("a"), ByteString1.fromString("a")).drop(0) should ===(ByteString("aa"))
+      ByteStrings(ByteString1.fromString(""), ByteString1.fromString("")).drop(0) should ===(ByteString(""))
+    }
+    "drop(1)" in {
+      ByteStrings(ByteString1.fromString(""), ByteString1.fromString("")).drop(1) should ===(ByteString(""))
+      ByteStrings(ByteString1.fromString("a"), ByteString1.fromString("")).drop(1) should ===(ByteString(""))
+      ByteStrings(ByteString1.fromString(""), ByteString1.fromString("a")).drop(1) should ===(ByteString(""))
+      ByteStrings(ByteString1.fromString("ab"), ByteString1.fromString("cd")).drop(1) should ===(ByteString("bcd"))
+      ByteStrings(Vector(ByteString1.fromString("xaaa"))).drop(1) should ===(ByteString("aaa"))
+    }
+    "drop(n)" in {
+      ByteStrings(ByteString1.fromString("a"), ByteString1.fromString("")).drop(1) should ===(ByteString(""))
+      ByteStrings(ByteString1.fromString(""), ByteString1.fromString("a")).drop(1) should ===(ByteString(""))
+      ByteStrings(ByteString1.fromString("ab"), ByteString1.fromString("cd")).drop(3) should ===(ByteString("d"))
+      ByteStrings(ByteString1.fromString("ab"), ByteString1.fromString("cd")).drop(4) should ===(ByteString(""))
+      ByteStrings(ByteString1.fromString("ab"), ByteString1.fromString("cd")).drop(5) should ===(ByteString(""))
+      ByteStrings(ByteString1.fromString("ab"), ByteString1.fromString("cd")).drop(10) should ===(ByteString(""))
+
+      ByteStrings(ByteString1.fromString("ab"), ByteString1.fromString("cd")).drop(-2) should ===(ByteString("abcd"))
+      ByteStrings(ByteString1.fromString(""), ByteString1.fromString("")).drop(-2) should ===(ByteString(""))
+      ByteStrings(ByteString1.fromString("ab"), ByteString1.fromString("")).drop(Int.MinValue) should ===(ByteString("ab"))
+      ByteStrings(ByteString1.fromString(""), ByteString1.fromString("ab")).dropRight(Int.MinValue) should ===(ByteString("ab"))
+    }
+    "slice" in {
+      ByteStrings(ByteString1.fromString("a"), ByteString1.fromString("")).slice(0, 1) should ===(ByteString("a"))
+      ByteStrings(ByteString1.fromString(""), ByteString1.fromString("a")).slice(1, 1) should ===(ByteString(""))
+      ByteStrings(ByteString1.fromString("ab"), ByteString1.fromString("cd")).slice(2, 2) should ===(ByteString(""))
+      ByteStrings(ByteString1.fromString("ab"), ByteString1.fromString("cd")).slice(2, 3) should ===(ByteString("c"))
+      ByteStrings(ByteString1.fromString("ab"), ByteString1.fromString("cd")).slice(2, 4) should ===(ByteString("cd"))
+      ByteStrings(ByteString1.fromString("ab"), ByteString1.fromString("cd")).slice(3, 4) should ===(ByteString("d"))
+      ByteStrings(ByteString1.fromString("ab"), ByteString1.fromString("cd")).slice(10, 100) should ===(ByteString(""))
+    }
+    "dropRight" in {
+      ByteStrings(ByteString1.fromString("a"), ByteString1.fromString("")).dropRight(0) should ===(ByteString("a"))
+      ByteStrings(ByteString1.fromString("a"), ByteString1.fromString("")).dropRight(-1) should ===(ByteString("a"))
+      ByteStrings(ByteString1.fromString("a"), ByteString1.fromString("")).dropRight(Int.MinValue) should ===(ByteString("a"))
+      ByteStrings(ByteString1.fromString("a"), ByteString1.fromString("")).dropRight(1) should ===(ByteString(""))
+      ByteStrings(ByteString1.fromString("a"), ByteString1.fromString("")).dropRight(Int.MaxValue) should ===(ByteString(""))
+      ByteStrings(ByteString1.fromString("a"), ByteString1.fromString("bc")).dropRight(1) should ===(ByteString("ab"))
+      ByteStrings(ByteString1.fromString("a"), ByteString1.fromString("bc")).dropRight(2) should ===(ByteString("a"))
+      ByteStrings(ByteString1.fromString("a"), ByteString1.fromString("bc")).dropRight(3) should ===(ByteString(""))
+    }
+    "take" in {
+      ByteStrings(ByteString1.fromString("a"), ByteString1.fromString("bc")).drop(1).take(0) should ===(ByteString(""))
+      ByteStrings(ByteString1.fromString("a"), ByteString1.fromString("bc")).drop(1).take(-1) should ===(ByteString(""))
+      ByteStrings(ByteString1.fromString("a"), ByteString1.fromString("bc")).drop(1).take(-2) should ===(ByteString(""))
+      (ByteStrings(ByteString1.fromString("a"), ByteString1.fromString("bc")) ++ ByteString1.fromString("defg")).drop(2) should ===(ByteString("cdefg"))
+      ByteStrings(ByteString1.fromString("a"), ByteString1.fromString("bc")).drop(2).take(1) should ===(ByteString("c"))
+      ByteStrings(ByteString1.fromString("a"), ByteString1.fromString("bc")).take(100) should ===(ByteString("abc"))
+      ByteStrings(ByteString1.fromString("a"), ByteString1.fromString("bc")).drop(1).take(100) should ===(ByteString("bc"))
+    }
+  }
+
   "A ByteString" must {
     "have correct size" when {
       "concatenating" in { check((a: ByteString, b: ByteString) ⇒ (a ++ b).size == a.size + b.size) }
       "dropping" in { check((a: ByteString, b: ByteString) ⇒ (a ++ b).drop(b.size).size == a.size) }
+      "taking" in { check((a: ByteString, b: ByteString) ⇒ (a ++ b).take(a.size) == a) }
+      "takingRight" in { check((a: ByteString, b: ByteString) ⇒ (a ++ b).takeRight(b.size) == b) }
+      "droppnig then taking" in { check((a: ByteString, b: ByteString) ⇒ (b ++ a ++ b).drop(b.size).take(a.size) == a) }
+      "droppingRight" in { check((a: ByteString, b: ByteString) ⇒ (b ++ a ++ b).drop(b.size).dropRight(b.size) == a) }
     }
 
     "be sequential" when {
@@ -301,6 +411,21 @@ class ByteStringSpec extends WordSpec with Matchers with Checkers {
           (a ++ b ++ c) == xs
         }
       }
+      def excerciseRecombining(xs: ByteString, from: Int, until: Int) = {
+        val (tmp, c) = xs.splitAt(until)
+        val (a, b) = tmp.splitAt(from)
+        (a ++ b ++ c) should ===(xs) 
+      }
+      "recombining - edge cases" in {
+        excerciseRecombining(ByteStrings(Vector(ByteString1(Array[Byte](1)), ByteString1(Array[Byte](2)))), -2147483648, 112121212)
+        excerciseRecombining(ByteStrings(Vector(ByteString1(Array[Byte](100)))), 0, 2)
+        excerciseRecombining(ByteStrings(Vector(ByteString1(Array[Byte](100)))), -2147483648, 2)
+        excerciseRecombining(ByteStrings(Vector(ByteString1.fromString("ab"), ByteString1.fromString("cd"))), 0, 1)
+        excerciseRecombining(ByteString1.fromString("abc").drop(1).take(1), -324234, 234232)
+        excerciseRecombining(ByteString("a"), 0, 2147483647)
+        excerciseRecombining(ByteStrings(Vector(ByteString1.fromString("ab"), ByteString1.fromString("cd"))).drop(2), 2147483647, 1)
+        excerciseRecombining(ByteString1.fromString("ab").drop1(1), Int.MaxValue, Int.MaxValue)
+      }
     }
 
     "behave as expected" when {
diff --git a/akka-actor/src/main/scala/akka/util/ByteIterator.scala b/akka-actor/src/main/scala/akka/util/ByteIterator.scala
index 3ca0092d49..e3a3ffdbab 100644
--- a/akka-actor/src/main/scala/akka/util/ByteIterator.scala
+++ b/akka-actor/src/main/scala/akka/util/ByteIterator.scala
@@ -234,6 +234,7 @@ object ByteIterator {
       new MultiByteArrayIterator(clonedIterators)
     }
 
+    /** For performance sensitive code, call take() directly on ByteString (it's optimised there) */
     final override def take(n: Int): this.type = {
       var rest = n
       val builder = new ListBuffer[ByteArrayIterator]
@@ -249,7 +250,8 @@ object ByteIterator {
       normalize()
     }
 
-    @tailrec final override def drop(n: Int): this.type =
+    /** For performance sensitive code, call drop() directly on ByteString (it's optimised there) */
+    final override def drop(n: Int): this.type =
       if ((n > 0) && !isEmpty) {
         val nCurrent = math.min(n, current.len)
         current.drop(n)
@@ -341,6 +343,7 @@ object ByteIterator {
     def getDoubles(xs: Array[Double], offset: Int, n: Int)(implicit byteOrder: ByteOrder): this.type =
       getToArray(xs, offset, n, 8) { getDouble(byteOrder) } { current.getDoubles(_, _, _)(byteOrder) }
 
+    /** For performance sensitive code, call copyToBuffer() directly on ByteString (it's optimised there) */
     override def copyToBuffer(buffer: ByteBuffer): Int = {
       // the fold here is better than indexing into the LinearSeq
       val n = iterators.foldLeft(0) { _ + _.copyToBuffer(buffer) }
@@ -636,6 +639,7 @@ abstract class ByteIterator extends BufferedIterator[Byte] {
    * @param buffer a ByteBuffer to copy bytes to
    * @return the number of bytes actually copied
    */
+  /** For performance sensitive code, call take() directly on ByteString (it's optimised there) */
   def copyToBuffer(buffer: ByteBuffer): Int
 
   /**
diff --git a/akka-actor/src/main/scala/akka/util/ByteString.scala b/akka-actor/src/main/scala/akka/util/ByteString.scala
index 89204c7169..1e782f3c09 100644
--- a/akka-actor/src/main/scala/akka/util/ByteString.scala
+++ b/akka-actor/src/main/scala/akka/util/ByteString.scala
@@ -12,10 +12,10 @@ import scala.annotation.{ tailrec, varargs }
 import scala.collection.IndexedSeqOptimized
 import scala.collection.mutable.{ Builder, WrappedArray }
 import scala.collection.immutable
-import scala.collection.immutable.{ IndexedSeq, VectorBuilder }
+import scala.collection.immutable.{ IndexedSeq, VectorBuilder, VectorIterator }
 import scala.collection.generic.CanBuildFrom
 import scala.reflect.ClassTag
-import java.nio.charset.StandardCharsets
+import java.nio.charset.{ Charset, StandardCharsets }
 
 object ByteString {
 
@@ -104,6 +104,7 @@ object ByteString {
     }
 
   private[akka] object ByteString1C extends Companion {
+    def fromString(s: String): ByteString1C = new ByteString1C(s.getBytes)
     def apply(bytes: Array[Byte]): ByteString1C = new ByteString1C(bytes)
     val SerializationIdentity = 1.toByte
 
@@ -124,29 +125,49 @@ object ByteString {
 
     override def length: Int = bytes.length
 
+    // Avoid `iterator` in performance sensitive code, call ops directly on ByteString instead
     override def iterator: ByteIterator.ByteArrayIterator = ByteIterator.ByteArrayIterator(bytes, 0, bytes.length)
 
-    private[akka] def toByteString1: ByteString1 = ByteString1(bytes)
+    /** INTERNAL API */
+    private[akka] def toByteString1: ByteString1 = ByteString1(bytes, 0, bytes.length)
 
+    /** INTERNAL API */
     private[akka] def byteStringCompanion = ByteString1C
 
-    def asByteBuffer: ByteBuffer = toByteString1.asByteBuffer
+    override def asByteBuffer: ByteBuffer = toByteString1.asByteBuffer
 
-    def asByteBuffers: scala.collection.immutable.Iterable[ByteBuffer] = List(asByteBuffer)
+    override def asByteBuffers: scala.collection.immutable.Iterable[ByteBuffer] = List(asByteBuffer)
 
-    def decodeString(charset: String): String =
+    override def decodeString(charset: String): String =
       if (isEmpty) "" else new String(bytes, charset)
 
-    def ++(that: ByteString): ByteString =
+    override def decodeString(charset: Charset): String =
+      if (isEmpty) "" else new String(bytes, charset)
+
+    override def ++(that: ByteString): ByteString = {
       if (that.isEmpty) this
       else if (this.isEmpty) that
       else toByteString1 ++ that
+    }
+
+    override def take(n: Int): ByteString =
+      if (n <= 0) ByteString.empty
+      else toByteString1.take(n)
+
+    override def dropRight(n: Int): ByteString =
+      if (n <= 0) this
+      else toByteString1.dropRight(n)
+
+    override def drop(n: Int): ByteString =
+      if (n <= 0) this
+      else toByteString1.drop(n)
 
     override def slice(from: Int, until: Int): ByteString =
-      if ((from != 0) || (until != length)) toByteString1.slice(from, until)
-      else this
+      if ((from == 0) && (until == length)) this
+      else if (from > length) ByteString.empty
+      else toByteString1.slice(from, until)
 
-    private[akka] def writeToOutputStream(os: ObjectOutputStream): Unit =
+    private[akka] override def writeToOutputStream(os: ObjectOutputStream): Unit =
       toByteString1.writeToOutputStream(os)
 
     override def copyToBuffer(buffer: ByteBuffer): Int =
@@ -154,7 +175,7 @@ object ByteString {
 
     /** INTERNAL API: Specialized for internal use, writing multiple ByteString1C into the same ByteBuffer. */
     private[akka] def writeToBuffer(buffer: ByteBuffer, offset: Int): Int = {
-      val copyLength = math.min(buffer.remaining, offset + length)
+      val copyLength = Math.min(buffer.remaining, offset + length)
       if (copyLength > 0) {
         buffer.put(bytes, offset, copyLength)
         drop(copyLength)
@@ -164,11 +185,14 @@ object ByteString {
 
   }
 
+  /** INTERNAL API: ByteString backed by exactly one array, with start / end markers */
   private[akka] object ByteString1 extends Companion {
     val empty: ByteString1 = new ByteString1(Array.empty[Byte])
-    def apply(bytes: Array[Byte]): ByteString1 = ByteString1(bytes, 0, bytes.length)
+    def fromString(s: String): ByteString1 = apply(s.getBytes)
+    def apply(bytes: Array[Byte]): ByteString1 = apply(bytes, 0, bytes.length)
     def apply(bytes: Array[Byte], startIndex: Int, length: Int): ByteString1 =
-      if (length == 0) empty else new ByteString1(bytes, startIndex, length)
+      if (length == 0) empty
+      else new ByteString1(bytes, Math.max(0, startIndex), Math.max(0, length))
 
     val SerializationIdentity = 0.toByte
 
@@ -185,6 +209,7 @@ object ByteString {
 
     def apply(idx: Int): Byte = bytes(checkRangeConvert(idx))
 
+    // Avoid `iterator` in performance sensitive code, call ops directly on ByteString instead
     override def iterator: ByteIterator.ByteArrayIterator =
       ByteIterator.ByteArrayIterator(bytes, startIndex, startIndex + length)
 
@@ -204,12 +229,41 @@ object ByteString {
 
     private[akka] def byteStringCompanion = ByteString1
 
+    override def dropRight(n: Int): ByteString =
+      dropRight1(n)
+
+    /** INTERNAL API */
+    private[akka] def dropRight1(n: Int): ByteString1 =
+      if (n <= 0) this
+      else if (length - n <= 0) ByteString1.empty
+      else new ByteString1(bytes, startIndex, length - n)
+
+    override def drop(n: Int): ByteString =
+      if (n <= 0) this else drop1(n)
+
+    /** INTERNAL API */
+    private[akka] def drop1(n: Int): ByteString1 = {
+      val nextStartIndex = startIndex + n
+      if (nextStartIndex >= bytes.length) ByteString1.empty
+      else ByteString1(bytes, nextStartIndex, length - n)
+    }
+
+    override def take(n: Int): ByteString =
+      if (n <= 0) ByteString.empty
+      else ByteString1(bytes, startIndex, Math.min(n, length))
+
+    override def slice(from: Int, until: Int): ByteString = {
+      if (from <= 0 && until >= length) this // we can do < / > since we're Compact
+      else if (until <= from) ByteString1.empty
+      else ByteString1(bytes, startIndex + from, until - from)
+    }
+
     override def copyToBuffer(buffer: ByteBuffer): Int =
       writeToBuffer(buffer)
 
     /** INTERNAL API: Specialized for internal use, writing multiple ByteString1C into the same ByteBuffer. */
     private[akka] def writeToBuffer(buffer: ByteBuffer): Int = {
-      val copyLength = math.min(buffer.remaining, length)
+      val copyLength = Math.min(buffer.remaining, length)
       if (copyLength > 0) {
         buffer.put(bytes, startIndex, copyLength)
         drop(copyLength)
@@ -228,7 +282,10 @@ object ByteString {
 
     def asByteBuffers: scala.collection.immutable.Iterable[ByteBuffer] = List(asByteBuffer)
 
-    def decodeString(charset: String): String =
+    override def decodeString(charset: String): String =
+      new String(if (length == bytes.length) bytes else toArray, charset)
+
+    override def decodeString(charset: Charset): String = // avoids Charset.forName lookup in String internals
       new String(if (length == bytes.length) bytes else toArray, charset)
 
     def ++(that: ByteString): ByteString = {
@@ -311,8 +368,9 @@ object ByteString {
    */
   final class ByteStrings private (private[akka] val bytestrings: Vector[ByteString1], val length: Int) extends ByteString with Serializable {
     if (bytestrings.isEmpty) throw new IllegalArgumentException("bytestrings must not be empty")
+    if (bytestrings.head.isEmpty) throw new IllegalArgumentException("bytestrings.head must not be empty")
 
-    def apply(idx: Int): Byte =
+    def apply(idx: Int): Byte = {
       if (0 <= idx && idx < length) {
         var pos = 0
         var seen = 0
@@ -322,7 +380,9 @@ object ByteString {
         }
         bytestrings(pos)(idx - seen)
       } else throw new IndexOutOfBoundsException(idx.toString)
+    }
 
+    // Avoid `iterator` in performance sensitive code, call ops directly on ByteString instead
     override def iterator: ByteIterator.MultiByteArrayIterator =
       ByteIterator.MultiByteArrayIterator(bytestrings.toStream map { _.iterator })
 
@@ -367,11 +427,83 @@ object ByteString {
 
     def decodeString(charset: String): String = compact.decodeString(charset)
 
+    def decodeString(charset: Charset): String =
+      compact.decodeString(charset)
+
     private[akka] def writeToOutputStream(os: ObjectOutputStream): Unit = {
       os.writeInt(bytestrings.length)
       bytestrings.foreach(_.writeToOutputStream(os))
     }
 
+    override def take(n: Int): ByteString = {
+      @tailrec def take0(n: Int, b: ByteStringBuilder, bs: Vector[ByteString1]): ByteString =
+        if (bs.isEmpty || n <= 0) b.result
+        else {
+          val head = bs.head
+          if (n <= head.length) b.append(head.take(n)).result
+          else take0(n - head.length, b.append(head), bs.tail)
+        }
+
+      if (n <= 0) ByteString.empty
+      else if (n >= length) this
+      else take0(n, ByteString.newBuilder, bytestrings)
+    }
+
+    override def dropRight(n: Int): ByteString =
+      if (n <= 0) this
+      else {
+        val last = bytestrings.last
+        if (n < last.length) new ByteStrings(bytestrings.init :+ last.dropRight1(n), length - n)
+        else {
+          val remaining = bytestrings.init
+          if (remaining.isEmpty) ByteString.empty
+          else {
+            val s = new ByteStrings(remaining, length - last.length)
+            val remainingToBeDropped = n - last.length
+            s.dropRight(remainingToBeDropped)
+          }
+        }
+      }
+
+    override def slice(from: Int, until: Int): ByteString =
+      if ((from == 0) && (until == length)) this
+      else if (from > length || until <= from) ByteString.empty
+      else drop(from).dropRight(length - until)
+
+    override def drop(n: Int): ByteString =
+      if (n <= 0) this
+      else if (n > length) ByteString.empty
+      else drop0(n)
+
+    private def drop0(n: Int): ByteString = {
+      var continue = true
+      var fullDrops = 0
+      var remainingToDrop = n
+      do {
+        // impl note: could be optimised a bit by using VectorIterator instead, 
+        //            however then we're forced to call .toVector which halfs performance
+        //            We can work around that, as there's a Scala private method "remainingVector" which is fast, 
+        //            but let's not go into calling private APIs here just yet.
+        val currentLength = bytestrings(fullDrops).length
+        if (remainingToDrop >= currentLength) {
+          fullDrops += 1
+          remainingToDrop -= currentLength
+        } else continue = false
+      } while (remainingToDrop > 0 && continue)
+
+      val remainingByteStrings = bytestrings.drop(fullDrops)
+      if (remainingByteStrings.isEmpty) ByteString.empty
+      else if (remainingToDrop > 0) {
+        val h: ByteString1 = remainingByteStrings.head.drop1(remainingToDrop)
+        val bs = remainingByteStrings.tail
+
+        if (h.isEmpty)
+          if (bs.isEmpty) ByteString.empty
+          else new ByteStrings(bs, length - n)
+        else new ByteStrings(h +: bs, length - n)
+      } else ByteStrings(remainingByteStrings, length - n)
+    }
+
     protected def writeReplace(): AnyRef = new SerializationProxy(this)
   }
 
@@ -422,6 +554,8 @@ sealed abstract class ByteString extends IndexedSeq[Byte] with IndexedSeqOptimiz
   // *must* be overridden by derived classes. This construction is necessary
   // to specialize the return type, as the method is already implemented in
   // a parent trait.
+  // 
+  // Avoid `iterator` in performance sensitive code, call ops directly on ByteString instead
   override def iterator: ByteIterator = throw new UnsupportedOperationException("Method iterator is not implemented in ByteString")
 
   override def head: Byte = apply(0)
@@ -429,14 +563,19 @@ sealed abstract class ByteString extends IndexedSeq[Byte] with IndexedSeqOptimiz
   override def last: Byte = apply(length - 1)
   override def init: ByteString = dropRight(1)
 
-  override def slice(from: Int, until: Int): ByteString =
-    if ((from == 0) && (until == length)) this
-    else iterator.slice(from, until).toByteString
-
-  override def take(n: Int): ByteString = slice(0, n)
+  // *must* be overridden by derived classes.
+  override def take(n: Int): ByteString = throw new UnsupportedOperationException("Method slice is not implemented in ByteString")
   override def takeRight(n: Int): ByteString = slice(length - n, length)
-  override def drop(n: Int): ByteString = slice(n, length)
-  override def dropRight(n: Int): ByteString = slice(0, length - n)
+
+  // these methods are optimized in derived classes utilising the maximum knowlage about data layout available to them:
+  // *must* be overridden by derived classes.
+  override def slice(from: Int, until: Int): ByteString = throw new UnsupportedOperationException("Method slice is not implemented in ByteString")
+
+  // *must* be overridden by derived classes.
+  override def drop(n: Int): ByteString = throw new UnsupportedOperationException("Method drop is not implemented in ByteString")
+
+  // *must* be overridden by derived classes.
+  override def dropRight(n: Int): ByteString = throw new UnsupportedOperationException("Method dropRight is not implemented in ByteString")
 
   override def takeWhile(p: Byte ⇒ Boolean): ByteString = iterator.takeWhile(p).toByteString
   override def dropWhile(p: Byte ⇒ Boolean): ByteString = iterator.dropWhile(p).toByteString
@@ -461,7 +600,7 @@ sealed abstract class ByteString extends IndexedSeq[Byte] with IndexedSeqOptimiz
    *
    * @return this ByteString copied into a byte array
    */
-  protected[ByteString] def toArray: Array[Byte] = toArray[Byte] // protected[ByteString] == public to Java but hidden to Scala * fnizz *
+  protected[ByteString] def toArray: Array[Byte] = toArray[Byte]
 
   override def toArray[B >: Byte](implicit arg0: ClassTag[B]): Array[B] = iterator.toArray
   override def copyToArray[B >: Byte](xs: Array[B], start: Int, len: Int): Unit =
@@ -488,11 +627,8 @@ sealed abstract class ByteString extends IndexedSeq[Byte] with IndexedSeqOptimiz
    * @param buffer a ByteBuffer to copy bytes to
    * @return the number of bytes actually copied
    */
-  def copyToBuffer(buffer: ByteBuffer): Int = {
-    // TODO: remove this impl, make it an abstract method when possible
-    // specialized versions of this method exist in sub-classes, we keep this impl for binary compatibility, it never is actually invoked
-    iterator.copyToBuffer(buffer)
-  }
+  // *must* be overridden by derived classes. 
+  def copyToBuffer(buffer: ByteBuffer): Int = throw new UnsupportedOperationException("Method copyToBuffer is not implemented in ByteString")
 
   /**
    * Create a new ByteString with all contents compacted into a single,
@@ -544,9 +680,16 @@ sealed abstract class ByteString extends IndexedSeq[Byte] with IndexedSeqOptimiz
 
   /**
    * Decodes this ByteString using a charset to produce a String.
+   * If you have a [[Charset]] instance available, use `decodeString(charset: java.nio.charset.Charset` instead.
    */
   def decodeString(charset: String): String
 
+  /**
+   * Decodes this ByteString using a charset to produce a String.
+   * Avoids Charset.forName lookup in String internals, thus is preferable to `decodeString(charset: String)`.
+   */
+  def decodeString(charset: Charset): String
+
   /**
    * map method that will automatically cast Int back into Byte.
    */
@@ -608,8 +751,8 @@ object CompactByteString {
    * an Array.
    */
   def fromArray(array: Array[Byte], offset: Int, length: Int): CompactByteString = {
-    val copyOffset = math.max(offset, 0)
-    val copyLength = math.max(math.min(array.length - copyOffset, length), 0)
+    val copyOffset = Math.max(offset, 0)
+    val copyLength = Math.max(Math.min(array.length - copyOffset, length), 0)
     if (copyLength == 0) empty
     else {
       val copyArray = new Array[Byte](copyLength)
@@ -706,6 +849,8 @@ final class ByteStringBuilder extends Builder[Byte, ByteString] {
 
   override def ++=(xs: TraversableOnce[Byte]): this.type = {
     xs match {
+      case b: ByteString if b.isEmpty ⇒
+      // do nothing
       case b: ByteString1C ⇒
         clearTemp()
         _builder += b.toByteString1
@@ -748,7 +893,7 @@ final class ByteStringBuilder extends Builder[Byte, ByteString] {
   /**
    * Java API: append a ByteString to this builder.
    */
-  def append(bs: ByteString): this.type = this ++= bs
+  def append(bs: ByteString): this.type = if (bs.isEmpty) this else this ++= bs
 
   /**
    * Add a single Byte to this builder.
@@ -915,7 +1060,7 @@ final class ByteStringBuilder extends Builder[Byte, ByteString] {
     fillByteBuffer(len * 8, byteOrder) { _.asDoubleBuffer.put(array, start, len) }
 
   def clear(): Unit = {
-    _builder.clear
+    _builder.clear()
     _length = 0
     _tempLength = 0
   }
diff --git a/akka-bench-jmh/src/main/scala/akka/util/ByteStringBenchmark.scala b/akka-bench-jmh/src/main/scala/akka/util/ByteString_copyToBuffer_Benchmark.scala
similarity index 96%
rename from akka-bench-jmh/src/main/scala/akka/util/ByteStringBenchmark.scala
rename to akka-bench-jmh/src/main/scala/akka/util/ByteString_copyToBuffer_Benchmark.scala
index ec62572d8c..5868897ec7 100644
--- a/akka-bench-jmh/src/main/scala/akka/util/ByteStringBenchmark.scala
+++ b/akka-bench-jmh/src/main/scala/akka/util/ByteString_copyToBuffer_Benchmark.scala
@@ -12,7 +12,7 @@ import org.openjdk.jmh.infra.Blackhole
 
 @State(Scope.Benchmark)
 @Measurement(timeUnit = TimeUnit.MILLISECONDS)
-class ByteStringBenchmark {
+class ByteString_copyToBuffer_Benchmark {
 
   val _bs_mini = ByteString(Array.ofDim[Byte](128 * 4))
   val _bs_small = ByteString(Array.ofDim[Byte](1024 * 1))
@@ -83,16 +83,10 @@ class ByteStringBenchmark {
     bss_large.copyToBuffer(buf)
   }
 
-  //  /** compact + copy */
-  //  @Benchmark
-  //  def bss_large_c_copyToBuffer: Int =
-  //    bss_large.compact.copyToBuffer(buf)
-
   /** Pre-compacted */
   @Benchmark
   def bss_large_pc_copyToBuffer(): Int = {
     buf.flip()
     bss_pc_large.copyToBuffer(buf)
   }
-
 }
diff --git a/akka-bench-jmh/src/main/scala/akka/util/ByteString_decode_Benchmark.scala b/akka-bench-jmh/src/main/scala/akka/util/ByteString_decode_Benchmark.scala
new file mode 100644
index 0000000000..b606f251fe
--- /dev/null
+++ b/akka-bench-jmh/src/main/scala/akka/util/ByteString_decode_Benchmark.scala
@@ -0,0 +1,64 @@
+/**
+ * Copyright (C) 2014-2016 Lightbend Inc. <http://www.lightbend.com>
+ */
+package akka.util
+
+import java.nio.charset.Charset
+import java.util.concurrent.TimeUnit
+
+import akka.util.ByteString.{ ByteString1C, ByteStrings }
+import org.openjdk.jmh.annotations._
+
+@State(Scope.Benchmark)
+@Measurement(timeUnit = TimeUnit.MILLISECONDS)
+class ByteString_decode_Benchmark {
+
+  val _bs_large = ByteString(Array.ofDim[Byte](1024 * 4))
+
+  val bs_large = ByteString(Array.ofDim[Byte](1024 * 4 * 4))
+
+  val bss_large = ByteStrings(Vector.fill(4)(bs_large.asInstanceOf[ByteString1C].toByteString1), 4 * bs_large.length)
+  val bc_large = bss_large.compact // compacted
+
+  val utf8String = "utf-8"
+  val utf8 = Charset.forName(utf8String)
+
+  /*
+    Using Charset helps a bit, but nothing impressive:
+  
+    [info] ByteString_decode_Benchmark.bc_large_decodeString_stringCharset_utf8        thrpt   20  21 612.293 ±  825.099  ops/s
+      =>
+    [info] ByteString_decode_Benchmark.bc_large_decodeString_charsetCharset_utf8       thrpt   20  22 473.372 ±  851.597  ops/s
+    
+    
+    [info] ByteString_decode_Benchmark.bs_large_decodeString_stringCharset_utf8        thrpt   20  84 443.674 ± 3723.987  ops/s
+      =>
+    [info] ByteString_decode_Benchmark.bs_large_decodeString_charsetCharset_utf8       thrpt   20  93 865.033 ± 2052.476  ops/s
+    
+    
+    [info] ByteString_decode_Benchmark.bss_large_decodeString_stringCharset_utf8       thrpt   20  14 886.553 ±  326.752  ops/s
+      =>
+    [info] ByteString_decode_Benchmark.bss_large_decodeString_charsetCharset_utf8      thrpt   20  16 031.670 ±  474.565  ops/s
+   */
+
+  @Benchmark
+  def bc_large_decodeString_stringCharset_utf8: String =
+    bc_large.decodeString(utf8String)
+  @Benchmark
+  def bs_large_decodeString_stringCharset_utf8: String =
+    bs_large.decodeString(utf8String)
+  @Benchmark
+  def bss_large_decodeString_stringCharset_utf8: String =
+    bss_large.decodeString(utf8String)
+
+  @Benchmark
+  def bc_large_decodeString_charsetCharset_utf8: String =
+    bc_large.decodeString(utf8)
+  @Benchmark
+  def bs_large_decodeString_charsetCharset_utf8: String =
+    bs_large.decodeString(utf8)
+  @Benchmark
+  def bss_large_decodeString_charsetCharset_utf8: String =
+    bss_large.decodeString(utf8)
+
+}
diff --git a/akka-bench-jmh/src/main/scala/akka/util/ByteString_dropSliceTake_Benchmark.scala b/akka-bench-jmh/src/main/scala/akka/util/ByteString_dropSliceTake_Benchmark.scala
new file mode 100644
index 0000000000..90ff47a807
--- /dev/null
+++ b/akka-bench-jmh/src/main/scala/akka/util/ByteString_dropSliceTake_Benchmark.scala
@@ -0,0 +1,156 @@
+/**
+ * Copyright (C) 2014-2016 Lightbend Inc. <http://www.lightbend.com>
+ */
+package akka.util
+
+import java.nio.ByteBuffer
+import java.util.concurrent.TimeUnit
+
+import akka.util.ByteString.{ ByteString1C, ByteStrings }
+import org.openjdk.jmh.annotations._
+
+@State(Scope.Benchmark)
+@Measurement(timeUnit = TimeUnit.MILLISECONDS)
+class ByteString_dropSliceTake_Benchmark {
+
+  val _bs_mini = ByteString(Array.ofDim[Byte](128 * 4))
+  val _bs_small = ByteString(Array.ofDim[Byte](1024 * 1))
+  val _bs_large = ByteString(Array.ofDim[Byte](1024 * 4))
+
+  val bs_mini = ByteString(Array.ofDim[Byte](128 * 4 * 4))
+  val bs_small = ByteString(Array.ofDim[Byte](1024 * 1 * 4))
+  val bs_large = ByteString(Array.ofDim[Byte](1024 * 4 * 4))
+
+  val bss_mini = ByteStrings(Vector.fill(4)(bs_mini.asInstanceOf[ByteString1C].toByteString1), 4 * bs_mini.length)
+  val bss_small = ByteStrings(Vector.fill(4)(bs_small.asInstanceOf[ByteString1C].toByteString1), 4 * bs_small.length)
+  val bss_large = ByteStrings(Vector.fill(4)(bs_large.asInstanceOf[ByteString1C].toByteString1), 4 * bs_large.length)
+  val bss_pc_large = bss_large.compact
+
+  /*
+   --------------------------------- BASELINE -------------------------------------------------------------------- 
+   [info] Benchmark                                                         Mode  Cnt            Score         Error  Units
+   [info] ByteString_dropSliceTake_Benchmark.bs_large_dropRight_100        thrpt   20  111 122 621.983 ± 6172679.160  ops/s
+   [info] ByteString_dropSliceTake_Benchmark.bs_large_dropRight_256        thrpt   20  110 238 003.870 ± 4042572.908  ops/s
+   [info] ByteString_dropSliceTake_Benchmark.bs_large_dropRight_2000       thrpt   20  106 435 449.123 ± 2972282.531  ops/s
+   [info] ByteString_dropSliceTake_Benchmark.bss_large_dropRight_100       thrpt   20    1 155 292.430 ±   23096.219  ops/s
+   [info] ByteString_dropSliceTake_Benchmark.bss_large_dropRight_256       thrpt   20    1 191 713.229 ±   15910.426  ops/s
+   [info] ByteString_dropSliceTake_Benchmark.bss_large_dropRight_2000      thrpt   20    1 201 342.579 ±   21119.392  ops/s
+
+   [info] ByteString_dropSliceTake_Benchmark.bs_large_drop_100             thrpt   20  108 252 561.824 ± 3841392.346  ops/s
+   [info] ByteString_dropSliceTake_Benchmark.bs_large_drop_256             thrpt   20  112 515 936.237 ± 5651549.124  ops/s
+   [info] ByteString_dropSliceTake_Benchmark.bs_large_drop_2000            thrpt   20  110 851 553.706 ± 3327510.108  ops/s
+   [info] ByteString_dropSliceTake_Benchmark.bss_large_drop_18             thrpt   20      983 544.541 ±   46299.808  ops/s
+   [info] ByteString_dropSliceTake_Benchmark.bss_large_drop_100            thrpt   20      875 345.433 ±   44760.533  ops/s
+   [info] ByteString_dropSliceTake_Benchmark.bss_large_drop_256            thrpt   20      864 182.258 ±  111172.303  ops/s
+   [info] ByteString_dropSliceTake_Benchmark.bss_large_drop_2000           thrpt   20      997 459.151 ±   33627.993  ops/s
+
+   [info] ByteString_dropSliceTake_Benchmark.bs_large_slice_80_80          thrpt   20  112 299 538.691 ± 7259114.294  ops/s
+   [info] ByteString_dropSliceTake_Benchmark.bs_large_slice_129_129        thrpt   20  105 640 836.625 ± 9112709.942  ops/s
+   [info] ByteString_dropSliceTake_Benchmark.bss_large_slice_80_80         thrpt   20   10 868 202.262 ±  526537.133  ops/s
+   [info] ByteString_dropSliceTake_Benchmark.bss_large_slice_129_129       thrpt   20    9 429 199.802 ± 1321542.453  ops/s
+   
+   --------------------------------- AFTER -----------------------------------------------------------------------
+   
+   ------ TODAY –––––––
+   [info] Benchmark                                                         Mode  Cnt            Score         Error  Units
+   [info] ByteString_dropSliceTake_Benchmark.bs_large_dropRight_100        thrpt   20  126 091 961.654 ± 2813125.268  ops/s
+   [info] ByteString_dropSliceTake_Benchmark.bs_large_dropRight_256        thrpt   20  118 393 394.350 ± 2934782.759  ops/s
+   [info] ByteString_dropSliceTake_Benchmark.bs_large_dropRight_2000       thrpt   20  119 183 386.004 ± 4445324.298  ops/s
+   [info] ByteString_dropSliceTake_Benchmark.bss_large_dropRight_100       thrpt   20    8 813 065.392 ±  234570.880  ops/s
+   [info] ByteString_dropSliceTake_Benchmark.bss_large_dropRight_256       thrpt   20    9 039 585.934 ±  297168.301  ops/s
+   [info] ByteString_dropSliceTake_Benchmark.bss_large_dropRight_2000      thrpt   20    9 629 458.168 ±  124846.904  ops/s
+   
+   [info] ByteString_dropSliceTake_Benchmark.bs_large_drop_100             thrpt   20  111 666 137.955 ± 4846727.674  ops/s
+   [info] ByteString_dropSliceTake_Benchmark.bs_large_drop_256             thrpt   20  114 405 514.622 ± 4985750.805  ops/s
+   [info] ByteString_dropSliceTake_Benchmark.bs_large_drop_2000            thrpt   20  114 364 716.297 ± 2512280.603  ops/s
+   [info] ByteString_dropSliceTake_Benchmark.bss_large_drop_18             thrpt   20   10 040 457.962 ±  527850.116  ops/s
+   [info] ByteString_dropSliceTake_Benchmark.bss_large_drop_100            thrpt   20    9 184 934.769 ±  549140.840  ops/s
+   [info] ByteString_dropSliceTake_Benchmark.bss_large_drop_256            thrpt   20   10 887 437.121 ±  195606.240  ops/s
+   [info] ByteString_dropSliceTake_Benchmark.bss_large_drop_2000           thrpt   20   10 725 300.292 ±  403470.413  ops/s
+   
+   [info] ByteString_dropSliceTake_Benchmark.bs_large_slice_80_80          thrpt   20  233 017 314.148 ± 7070246.826  ops/s
+   [info] ByteString_dropSliceTake_Benchmark.bs_large_slice_129_129        thrpt   20  275 245 086.247 ± 4969752.048  ops/s
+   [info] ByteString_dropSliceTake_Benchmark.bss_large_slice_80_80         thrpt   20  264 963 420.976 ± 4259289.143  ops/s
+   [info] ByteString_dropSliceTake_Benchmark.bss_large_slice_129_129       thrpt   20  265 477 577.022 ± 4623974.283  ops/s
+   
+   */
+
+  // 18 == "http://example.com", a typical url length 
+
+  @Benchmark
+  def bs_large_drop_0: ByteString =
+    bs_large.drop(0)
+  @Benchmark
+  def bss_large_drop_0: ByteString =
+    bss_large.drop(0)
+
+  @Benchmark
+  def bs_large_drop_18: ByteString =
+    bs_large.drop(18)
+  @Benchmark
+  def bss_large_drop_18: ByteString =
+    bss_large.drop(18)
+
+  @Benchmark
+  def bs_large_drop_100: ByteString =
+    bs_large.drop(100)
+  @Benchmark
+  def bss_large_drop_100: ByteString =
+    bss_large.drop(100)
+
+  @Benchmark
+  def bs_large_drop_256: ByteString =
+    bs_large.drop(256)
+  @Benchmark
+  def bss_large_drop_256: ByteString =
+    bss_large.drop(256)
+
+  @Benchmark
+  def bs_large_drop_2000: ByteString =
+    bs_large.drop(2000)
+  @Benchmark
+  def bss_large_drop_2000: ByteString =
+    bss_large.drop(2000)
+
+  /* these force 2 array drops, and 1 element drop inside the 2nd to first/last; can be considered as "bad case" */
+
+  @Benchmark
+  def bs_large_slice_129_129: ByteString =
+    bs_large.slice(129, 129)
+  @Benchmark
+  def bss_large_slice_129_129: ByteString =
+    bss_large.slice(129, 129)
+
+  /* these only move the indexes, don't drop any arrays "happy case" */
+
+  @Benchmark
+  def bs_large_slice_80_80: ByteString =
+    bs_large.slice(80, 80)
+  @Benchmark
+  def bss_large_slice_80_80: ByteString =
+    bss_large.slice(80, 80)
+
+  // drop right ---
+
+  @Benchmark
+  def bs_large_dropRight_100: ByteString =
+    bs_large.dropRight(100)
+  @Benchmark
+  def bss_large_dropRight_100: ByteString =
+    bss_large.dropRight(100)
+
+  @Benchmark
+  def bs_large_dropRight_256: ByteString =
+    bs_large.dropRight(256)
+  @Benchmark
+  def bss_large_dropRight_256: ByteString =
+    bss_large.dropRight(256)
+
+  @Benchmark
+  def bs_large_dropRight_2000: ByteString =
+    bs_large.dropRight(2000)
+  @Benchmark
+  def bss_large_dropRight_2000: ByteString =
+    bss_large.dropRight(2000)
+
+}
diff --git a/akka-http-core/src/main/scala/akka/http/impl/engine/parsing/HttpHeaderParser.scala b/akka-http-core/src/main/scala/akka/http/impl/engine/parsing/HttpHeaderParser.scala
index fa4b23637a..63588995e8 100644
--- a/akka-http-core/src/main/scala/akka/http/impl/engine/parsing/HttpHeaderParser.scala
+++ b/akka-http-core/src/main/scala/akka/http/impl/engine/parsing/HttpHeaderParser.scala
@@ -473,7 +473,7 @@ private[http] object HttpHeaderParser {
   private[parsing] class ModeledHeaderValueParser(headerName: String, maxHeaderValueLength: Int, maxValueCount: Int, settings: HeaderParser.Settings)
     extends HeaderValueParser(headerName, maxValueCount) {
     def apply(hhp: HttpHeaderParser, input: ByteString, valueStart: Int, onIllegalHeader: ErrorInfo ⇒ Unit): (HttpHeader, Int) = {
-      // TODO: optimize by running the header value parser directly on the input ByteString (rather than an extracted String)
+      // TODO: optimize by running the header value parser directly on the input ByteString (rather than an extracted String); seems done?
       val (headerValue, endIx) = scanHeaderValue(hhp, input, valueStart, valueStart + maxHeaderValueLength + 2)()
       val trimmedHeaderValue = headerValue.trim
       val header = HeaderParser.parseFull(headerName, trimmedHeaderValue, settings) match {
@@ -569,4 +569,4 @@ private[http] object HttpHeaderParser {
     def withValueCountIncreased = copy(valueCount = valueCount + 1)
     def spaceLeft = valueCount < parser.maxValueCount
   }
-}
\ No newline at end of file
+}
diff --git a/akka-http-core/src/main/scala/akka/http/impl/engine/parsing/HttpRequestParser.scala b/akka-http-core/src/main/scala/akka/http/impl/engine/parsing/HttpRequestParser.scala
index 3fe26250ac..cfb40519b5 100644
--- a/akka-http-core/src/main/scala/akka/http/impl/engine/parsing/HttpRequestParser.scala
+++ b/akka-http-core/src/main/scala/akka/http/impl/engine/parsing/HttpRequestParser.scala
@@ -101,8 +101,8 @@ private[http] class HttpRequestParser(
 
     val uriEnd = findUriEnd()
     try {
-      uriBytes = input.iterator.slice(uriStart, uriEnd).toArray[Byte] // TODO: can we reduce allocations here?
-      uri = Uri.parseHttpRequestTarget(uriBytes, mode = uriParsingMode)
+      uriBytes = input.slice(uriStart, uriEnd).toArray[Byte] // TODO: can we reduce allocations here?
+      uri = Uri.parseHttpRequestTarget(uriBytes, mode = uriParsingMode) // TODO ByteStringParserInput?
     } catch {
       case IllegalUriException(info) ⇒ throw new ParsingException(BadRequest, info)
     }
diff --git a/akka-http-tests/src/multi-jvm/scala/akka/http/AkkaHttpServerLatencyMultiNodeSpec.scala b/akka-http-tests/src/multi-jvm/scala/akka/http/AkkaHttpServerLatencyMultiNodeSpec.scala
index 3640a4d0c9..ef5eac35c5 100644
--- a/akka-http-tests/src/multi-jvm/scala/akka/http/AkkaHttpServerLatencyMultiNodeSpec.scala
+++ b/akka-http-tests/src/multi-jvm/scala/akka/http/AkkaHttpServerLatencyMultiNodeSpec.scala
@@ -52,29 +52,28 @@ object AkkaHttpServerLatencyMultiNodeSpec extends MultiNodeConfig {
 
   private var _ifWrk2Available: Option[Boolean] = None
   final def ifWrk2Available(test: ⇒ Unit): Unit =
-    if (isWrk2Available) test else throw new TestPendingException() 
-  final def isWrk2Available: Boolean = 
+    if (isWrk2Available) test else throw new TestPendingException()
+  final def isWrk2Available: Boolean =
     _ifWrk2Available getOrElse {
-        import scala.sys.process._
-        val wrkExitCode = Try("""wrk""".!).getOrElse(-1)
+      import scala.sys.process._
+      val wrkExitCode = Try("""wrk""".!).getOrElse(-1)
 
-        _ifWrk2Available = Some(wrkExitCode == 1) // app found, help displayed
-        isWrk2Available
+      _ifWrk2Available = Some(wrkExitCode == 1) // app found, help displayed
+      isWrk2Available
     }
 
   private var _abAvailable: Option[Boolean] = None
   final def ifAbAvailable(test: ⇒ Unit): Unit =
     if (isAbAvailable) test else throw new TestPendingException()
-  
-  final def isAbAvailable: Boolean = 
+
+  final def isAbAvailable: Boolean =
     _abAvailable getOrElse {
       import scala.sys.process._
       val abExitCode = Try("""ab -h""".!).getOrElse(-1)
       _abAvailable = Some(abExitCode == 22) // app found, help displayed (22 return code is when -h runs in ab, weird but true)
       isAbAvailable
     }
-  
-  
+
   final case class LoadGenCommand(cmd: String)
   final case class LoadGenResults(results: String) {
     def lines = results.split("\n")
@@ -92,13 +91,13 @@ object AkkaHttpServerLatencyMultiNodeSpec extends MultiNodeConfig {
     import scala.sys.process._
     def ready(port: Int): Receive = {
       case LoadGenCommand(cmd) if cmd startsWith "wrk" ⇒
-        val res = 
+        val res =
           if (isWrk2Available) cmd.!! // blocking. DON'T DO THIS AT HOME, KIDS!
           else "=== WRK NOT AVAILABLE ==="
         sender() ! LoadGenResults(res)
-      
+
       case LoadGenCommand(cmd) if cmd startsWith "ab" ⇒
-        val res = 
+        val res =
           if (isAbAvailable) cmd.!! // blocking. DON'T DO THIS AT HOME, KIDS!
           else "=== AB NOT AVAILABLE ==="
         sender() ! LoadGenResults(res)
diff --git a/akka-http/src/main/scala/akka/http/scaladsl/unmarshalling/PredefinedFromEntityUnmarshallers.scala b/akka-http/src/main/scala/akka/http/scaladsl/unmarshalling/PredefinedFromEntityUnmarshallers.scala
index 21f629c050..43f4cbd420 100644
--- a/akka-http/src/main/scala/akka/http/scaladsl/unmarshalling/PredefinedFromEntityUnmarshallers.scala
+++ b/akka-http/src/main/scala/akka/http/scaladsl/unmarshalling/PredefinedFromEntityUnmarshallers.scala
@@ -35,7 +35,7 @@ trait PredefinedFromEntityUnmarshallers extends MultipartUnmarshallers {
   implicit def stringUnmarshaller: FromEntityUnmarshaller[String] =
     byteStringUnmarshaller mapWithInput { (entity, bytes) ⇒
       if (entity.isKnownEmpty) ""
-      else bytes.decodeString(Unmarshaller.bestUnmarshallingCharsetFor(entity).nioCharset.name)
+      else bytes.decodeString(Unmarshaller.bestUnmarshallingCharsetFor(entity).nioCharset)
     }
 
   implicit def defaultUrlEncodedFormDataUnmarshaller: FromEntityUnmarshaller[FormData] =
@@ -53,4 +53,4 @@ trait PredefinedFromEntityUnmarshallers extends MultipartUnmarshallers {
     }
 }
 
-object PredefinedFromEntityUnmarshallers extends PredefinedFromEntityUnmarshallers
\ No newline at end of file
+object PredefinedFromEntityUnmarshallers extends PredefinedFromEntityUnmarshallers
diff --git a/project/MiMa.scala b/project/MiMa.scala
index 43bd291a6f..1081eed72a 100644
--- a/project/MiMa.scala
+++ b/project/MiMa.scala
@@ -904,6 +904,10 @@ object MiMa extends AutoPlugin {
 
         // #20543 GraphStage subtypes should not be private to akka
         ProblemFilters.exclude[DirectAbstractMethodProblem]("akka.stream.ActorMaterializer.actorOf")
+      ),
+      "2.4.9" -> Seq(
+        // #20994 adding new decode method, since we're on JDK7+ now
+        ProblemFilters.exclude[ReversedMissingMethodProblem]("akka.util.ByteString.decodeString")
       )
     )
   }