Set BLOB via ScatteringByteChannel (first attempt)

Author: mst-appear

mysql-async/README.md

@@ -91,8 +91,10 @@ String | string
 Array[Byte] | blob
 java.nio.ByteBuffer | blob
 io.netty.buffer.ByteBuf | blob
+java.nio.channels.ScatteringByteChannel | blob
 
-The maximum size of a blob is 2^24-9 bytes (almost 16 MiB).
+The maximum size of a blob you set via Array[Byte], ByteBuffer or ByteBuf is 2^24-9 bytes (almost 16 MiB).
+Blobs set via a Channel can be larger, the Channel is read until EOF and then closed.
 
 You don't have to match exact values when sending parameters for your prepared statements, MySQL is usually smart
 enough to understand that if you have sent an Int to `smallint` column it has to truncate the 4 bytes into 2.

mysql-async/src/main/scala/com/github/mauricio/async/db/mysql/binary/BinaryRowEncoder.scala

@@ -16,6 +16,8 @@
 
 package com.github.mauricio.async.db.mysql.binary
 
+import java.nio.channels.ScatteringByteChannel
+
 import io.netty.buffer.ByteBuf
 import java.nio.ByteBuffer
 import java.nio.charset.Charset
@@ -88,6 +90,7 @@ class BinaryRowEncoder( charset : Charset ) {
           case v : java.util.Date => JavaDateEncoder
           case v : ByteBuffer => ByteBufferEncoder
           case v : ByteBuf => ByteBufEncoder
+          case v : ScatteringByteChannel => DummyBlobEncoder
         }
       }
     }

mysql-async/src/main/scala/com/github/mauricio/async/db/mysql/binary/encoder/DummyBlobEncoder.scala

@@ -0,0 +1,14 @@
+package com.github.mauricio.async.db.mysql.binary.encoder
+
+import com.github.mauricio.async.db.mysql.column.ColumnTypes
+import io.netty.buffer.ByteBuf
+
+object DummyBlobEncoder extends BinaryEncoder {
+
+  def encode(value: Any, buffer: ByteBuf): Unit = {
+    throw new UnsupportedOperationException()
+  }
+
+  def encodesTo: Int = ColumnTypes.FIELD_TYPE_BLOB
+
+ }

mysql-async/src/main/scala/com/github/mauricio/async/db/mysql/codec/MySQLConnectionHandler.scala

@@ -17,6 +17,7 @@
 package com.github.mauricio.async.db.mysql.codec
 
 import java.nio.ByteBuffer
+import java.nio.channels.ScatteringByteChannel
 
 import com.github.mauricio.async.db.Configuration
 import com.github.mauricio.async.db.general.MutableResultSet
@@ -260,6 +261,7 @@ class MySQLConnectionHandler(
       case v : Array[Byte] => v.length > SendLongDataEncoder.LONG_THRESHOLD
       case v : ByteBuffer => v.remaining() > SendLongDataEncoder.LONG_THRESHOLD
       case v : ByteBuf => v.readableBytes() > SendLongDataEncoder.LONG_THRESHOLD
+      case _ : ScatteringByteChannel => true
 
       case _ => false
     }
@@ -275,9 +277,27 @@ class MySQLConnectionHandler(
 
       case v : ByteBuf =>
         sendBuffer(v, statementId, index)
+
+      case channel : ScatteringByteChannel =>
+        sendChannel(channel, statementId, index)
     }
   }
 
+  // TODO this is blocking
+  private def sendChannel(channel: ScatteringByteChannel, statementId: Array[Byte], paramId: Int) {
+    var bytesWritten = 0
+    do {
+      val dataBuffer = Unpooled.directBuffer(SendLongDataEncoder.INITIAL_BUFFER_SIZE, SendLongDataEncoder.MAX_BUFFER_SIZE)
+      do {
+        bytesWritten = dataBuffer.writeBytes(channel, SendLongDataEncoder.MAX_BUFFER_SIZE)
+      } while (bytesWritten == 0)
+      if (bytesWritten > 0) {
+        sendBuffer(dataBuffer, statementId, paramId)
+      }
+    } while (bytesWritten > -1)
+    channel.close()
+  }
+
   private def sendBuffer(buffer: ByteBuf, statementId: Array[Byte], paramId: Int) {
     writeAndHandleError(new SendLongDataMessage(statementId, buffer, paramId))
   }

mysql-async/src/main/scala/com/github/mauricio/async/db/mysql/codec/SendLongDataEncoder.scala

@@ -10,6 +10,10 @@ object SendLongDataEncoder {
   val log = Log.get[SendLongDataEncoder]
 
   val LONG_THRESHOLD = 1023
+
+  val INITIAL_BUFFER_SIZE = 1024 // 1 KiB
+
+  val MAX_BUFFER_SIZE = 1024*1024 // 1 MiB
 }
 
 class SendLongDataEncoder

mysql-async/src/test/scala/com/github/mauricio/async/db/mysql/BinaryColumnsSpec.scala

@@ -1,6 +1,8 @@
 package com.github.mauricio.async.db.mysql
 
-import org.specs2.mutable.Specification
+import java.io.{FileInputStream, FileOutputStream, BufferedOutputStream, File}
+
+import org.specs2.mutable.{After, Specification}
 import java.util.UUID
 import java.nio.ByteBuffer
 import io.netty.buffer.Unpooled
@@ -9,6 +11,17 @@ import com.github.mauricio.async.db.RowData
 
 class BinaryColumnsSpec extends Specification with ConnectionHelper {
 
+  val createBlobTable =
+    """CREATE TEMPORARY TABLE POSTS (
+      | id INT NOT NULL,
+      | blob_column LONGBLOB,
+      | primary key (id))
+    """.stripMargin
+
+  val insertIntoBlobTable = "INSERT INTO POSTS (id,blob_column) VALUES (?,?)"
+
+  val selectFromBlobTable = "SELECT id,blob_column FROM POSTS ORDER BY id"
+
   "connection" should {
 
     "correctly load fields as byte arrays" in {
@@ -106,35 +119,43 @@ class BinaryColumnsSpec extends Specification with ConnectionHelper {
 
     }
 
-    "support BLOB type with large values" in {
+    "support BLOB type with long values" in {
 
       val bytes = (1 to 2100).map(_.toByte).toArray
 
       testBlob(bytes)
 
     }
 
-  }
+    "support BLOB type with ScatteringByteChannel input" in new BlobFile {
 
-  def testBlob(bytes: Array[Byte]) = {
-    val create =
-      """CREATE TEMPORARY TABLE POSTS (
-        | id INT NOT NULL,
-        | blob_column BLOB,
-        | primary key (id))
-      """.stripMargin
+      withConnection {
+        connection =>
+          executeQuery(connection, createBlobTable)
+
+          val channel = new FileInputStream(blobFile).getChannel
+          executePreparedStatement(connection, insertIntoBlobTable, 1, channel)
+
+          val Some(rows) = executeQuery(connection, selectFromBlobTable).rows
+          rows(0)("id") === 1
+          val retrievedBlob = rows(0)("blob_column").asInstanceOf[Array[Byte]]
+          retrievedBlob.length === BlobSize
+          0 to retrievedBlob.length-1 foreach { n => retrievedBlob(n) === n.toByte }
+      }
 
-    val insert = "INSERT INTO POSTS (id,blob_column) VALUES (?,?)"
-    val select = "SELECT id,blob_column FROM POSTS ORDER BY id"
+    }
 
+  }
+
+  def testBlob(bytes: Array[Byte]) = {
     withConnection {
       connection =>
-        executeQuery(connection, create)
-        executePreparedStatement(connection, insert, 1, Some(bytes))
-        executePreparedStatement(connection, insert, 2, ByteBuffer.wrap(bytes))
-        executePreparedStatement(connection, insert, 3, Unpooled.wrappedBuffer(bytes))
+        executeQuery(connection, createBlobTable)
+        executePreparedStatement(connection, insertIntoBlobTable, 1, Some(bytes))
+        executePreparedStatement(connection, insertIntoBlobTable, 2, ByteBuffer.wrap(bytes))
+        executePreparedStatement(connection, insertIntoBlobTable, 3, Unpooled.wrappedBuffer(bytes))
 
-        val Some(rows) = executeQuery(connection, select).rows
+        val Some(rows) = executeQuery(connection, selectFromBlobTable).rows
         rows(0)("id") === 1
         rows(0)("blob_column") === bytes
         rows(1)("id") === 2
@@ -149,4 +170,20 @@ class BinaryColumnsSpec extends Specification with ConnectionHelper {
   def compareBytes( row : RowData, column : String, expected : String ) =
     row(column) === expected.getBytes(CharsetUtil.UTF_8)
 
-}
+}
+
+trait BlobFile extends After {
+  val BlobSize = (16 * 1024 * 1024)-9
+
+  lazy val blobFile = {
+    val file = File.createTempFile("blob", null)
+    val bos = new BufferedOutputStream(new FileOutputStream(file))
+    0 to BlobSize-1 foreach { n => bos.write(n.toByte) }
+    bos.close()
+    file
+  }
+
+  def after = {
+    blobFile.delete()
+  }
+}