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 a5f65b6a7e7..b1a40c8cb7a 100644
--- a/akka-actor-tests/src/test/scala/akka/util/ByteStringSpec.scala
+++ b/akka-actor-tests/src/test/scala/akka/util/ByteStringSpec.scala
@@ -269,9 +269,19 @@ class ByteStringSpec extends WordSpec with MustMatchers with Checkers {
             (b.compact eq b)
         }
       }
+
+      "asByteBuffers" in {
+        check { (a: ByteString) ⇒ if (a.isCompact) a.asByteBuffers.size == 1 && a.asByteBuffers.head == a.asByteBuffer else a.asByteBuffers.size > 0 }
+        check { (a: ByteString) ⇒ a.asByteBuffers.foldLeft(ByteString.empty) { (bs, bb) ⇒ bs ++ ByteString(bb) } == a }
+        check { (a: ByteString) ⇒ a.asByteBuffers.forall(_.isReadOnly) }
+        check { (a: ByteString) ⇒
+                import scala.collection.JavaConverters.iterableAsScalaIterableConverter;
+                a.asByteBuffers.zip(a.getByteBuffers().asScala).forall(x ⇒ x._1 == x._2)
+        }
+      }
     }
     "behave like a Vector" when {
-      "concatenating" in { check { (a: ByteString, b: ByteString) ⇒ likeVectors(a, b) { (a, b) ⇒ (a ++ b) } } }
+      "concatenating" in { check { (a: ByteString, b: ByteString) ⇒ likeVectors(a, b) { _ ++ _ } } }
 
       "calling apply" in {
         check { slice: ByteStringSlice ⇒
diff --git a/akka-actor/src/main/scala/akka/util/ByteString.scala b/akka-actor/src/main/scala/akka/util/ByteString.scala
index b6945285414..7905a15e42b 100644
--- a/akka-actor/src/main/scala/akka/util/ByteString.scala
+++ b/akka-actor/src/main/scala/akka/util/ByteString.scala
@@ -5,9 +5,11 @@
 package akka.util
 
 import java.nio.{ ByteBuffer, ByteOrder }
+import java.lang.{ Iterable ⇒ JIterable }
 
 import scala.collection.IndexedSeqOptimized
 import scala.collection.mutable.{ Builder, WrappedArray }
+import scala.collection.immutable
 import scala.collection.immutable.{ IndexedSeq, VectorBuilder }
 import scala.collection.generic.CanBuildFrom
 import scala.reflect.ClassTag
@@ -94,10 +96,12 @@ object ByteString {
 
     private[akka] def toByteString1: ByteString1 = ByteString1(bytes)
 
-    def asByteBuffer: ByteBuffer =
-      toByteString1.asByteBuffer
+    def asByteBuffer: ByteBuffer = toByteString1.asByteBuffer
 
-    def decodeString(charset: String): String = new String(bytes, charset)
+    def asByteBuffers: scala.collection.immutable.Iterable[ByteBuffer] = List(asByteBuffer)
+
+    def decodeString(charset: String): String =
+      if (isEmpty) "" else new String(bytes, charset)
 
     def ++(that: ByteString): ByteString =
       if (that.isEmpty) this
@@ -110,9 +114,10 @@ object ByteString {
   }
 
   private[akka] object ByteString1 {
-    def apply(bytes: Array[Byte]): ByteString1 = new ByteString1(bytes)
+    val empty: ByteString1 = new ByteString1(Array.empty[Byte])
+    def apply(bytes: Array[Byte]): ByteString1 = ByteString1(bytes, 0, bytes.length)
     def apply(bytes: Array[Byte], startIndex: Int, length: Int): ByteString1 =
-      new ByteString1(bytes, startIndex, length)
+      if (length == 0) empty else new ByteString1(bytes, startIndex, length)
   }
 
   /**
@@ -145,6 +150,8 @@ object ByteString {
       else buffer
     }
 
+    def asByteBuffers: scala.collection.immutable.Iterable[ByteBuffer] = List(asByteBuffer)
+
     def decodeString(charset: String): String =
       new String(if (length == bytes.length) bytes else toArray, charset)
 
@@ -250,6 +257,8 @@ object ByteString {
 
     def asByteBuffer: ByteBuffer = compact.asByteBuffer
 
+    def asByteBuffers: scala.collection.immutable.Iterable[ByteBuffer] = bytestrings map { _.asByteBuffer }
+
     def decodeString(charset: String): String = compact.decodeString(charset)
   }
 
@@ -345,6 +354,25 @@ sealed abstract class ByteString extends IndexedSeq[Byte] with IndexedSeqOptimiz
    */
   def asByteBuffer: ByteBuffer
 
+  /**
+   * Returns an immutable Iterable of read-only ByteBuffers that directly wraps this ByteStrings
+   * all fragments. Will always have at least one entry.
+   *
+   * Scala API
+   */
+  def asByteBuffers: immutable.Iterable[ByteBuffer]
+
+  /**
+   * Returns an Iterable of read-only ByteBuffers that directly wraps this ByteStrings
+   * all fragments. Will always have at least one entry.
+   *
+   * Java API
+   */
+  def getByteBuffers(): JIterable[ByteBuffer] = {
+    import scala.collection.JavaConverters.asJavaIterableConverter
+    asByteBuffers.asJava
+  }
+
   /**
    * Creates a new ByteBuffer with a copy of all bytes contained in this
    * ByteString.