Separate protocol encoder for MySQL SendLongData

Author: mst-appear

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

@@ -52,6 +52,7 @@ class MySQLConnectionHandler(
  private final val connectionPromise = Promise[MySQLConnectionHandler]
  private final val decoder = new MySQLFrameDecoder(configuration.charset, connectionId)
  private final val encoder = new MySQLOneToOneEncoder(configuration.charset, charsetMapper)
+ private final val sendLongDataEncoder = new SendLongDataEncoder(configuration.charset)
  private final val currentParameters = new ArrayBuffer[ColumnDefinitionMessage]()
  private final val currentColumns = new ArrayBuffer[ColumnDefinitionMessage]()
  private final val parsedStatements = new HashMap[String,PreparedStatementHolder]()
@@ -70,6 +71,7 @@ class MySQLConnectionHandler(
  channel.pipeline.addLast(
  decoder,
  encoder,
+ sendLongDataEncoder,
  MySQLConnectionHandler.this)
  }
 
@@ -252,7 +254,7 @@ class MySQLConnectionHandler(
  case Some(v) => v
  case _ => maybeValue
  }
- encoder.isLong(value)
+ sendLongDataEncoder.isLong(value)
  }
  }
 

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

@@ -32,69 +32,57 @@ object MySQLOneToOneEncoder {
  val log = Log.get[MySQLOneToOneEncoder]
 }
 
-class MySQLOneToOneEncoder(charset: Charset, charsetMapper: CharsetMapper) extends MessageToMessageEncoder[Any] {
+class MySQLOneToOneEncoder(charset: Charset, charsetMapper: CharsetMapper)
+ extends MessageToMessageEncoder[ClientMessage](classOf[ClientMessage]) {
 
  import MySQLOneToOneEncoder.log
 
  private final val handshakeResponseEncoder = new HandshakeResponseEncoder(charset, charsetMapper)
  private final val queryEncoder = new QueryMessageEncoder(charset)
  private final val rowEncoder = new BinaryRowEncoder(charset)
  private final val prepareEncoder = new PreparedStatementPrepareEncoder(charset)
- private final val sendLongDataEncoder = new SendLongDataEncoder(rowEncoder)
  private final val executeEncoder = new PreparedStatementExecuteEncoder(rowEncoder)
  private final val authenticationSwitchEncoder = new AuthenticationSwitchResponseEncoder(charset)
 
  private var sequence = 1
 
- def isLong(value: Any): Boolean = rowEncoder.encoderFor(value).isLong(value)
-
- def encode(ctx: ChannelHandlerContext, msg: Any, out: java.util.List[Object]): Unit = {
-
- msg match {
- case message: ClientMessage => {
- val encoder = (message.kind: @switch) match {
- case ClientMessage.ClientProtocolVersion => this.handshakeResponseEncoder
- case ClientMessage.Quit => {
- sequence = 0
- QuitMessageEncoder
- }
- case ClientMessage.Query => {
- sequence = 0
- this.queryEncoder
- }
- case ClientMessage.PreparedStatementExecute => {
- sequence = 0
- this.executeEncoder
- }
- case ClientMessage.PreparedStatementPrepare => {
- sequence = 0
- this.prepareEncoder
- }
- case ClientMessage.PreparedStatementSendLongData => {
- sequence = 0
- this.sendLongDataEncoder
- }
- case ClientMessage.AuthSwitchResponse => {
- sequence += 1
- this.authenticationSwitchEncoder
- }
- case _ => throw new EncoderNotAvailableException(message)
- }
-
- val result = encoder.encode(message)
+ def encode(ctx: ChannelHandlerContext, message: ClientMessage, out: java.util.List[Object]): Unit = {
+ val encoder = (message.kind: @switch) match {
+ case ClientMessage.ClientProtocolVersion => this.handshakeResponseEncoder
+ case ClientMessage.Quit => {
+ sequence = 0
+ QuitMessageEncoder
+ }
+ case ClientMessage.Query => {
+ sequence = 0
+ this.queryEncoder
+ }
+ case ClientMessage.PreparedStatementExecute => {
+ sequence = 0
+ this.executeEncoder
+ }
+ case ClientMessage.PreparedStatementPrepare => {
+ sequence = 0
+ this.prepareEncoder
+ }
+ case ClientMessage.AuthSwitchResponse => {
+ sequence += 1
+ this.authenticationSwitchEncoder
+ }
+ case _ => throw new EncoderNotAvailableException(message)
+ }
 
-  ByteBufferUtils.writePacketLength(result, sequence)
+ val result: ByteBuf = encoder.encode(message)
 
-  sequence += 1
+ ByteBufferUtils.writePacketLength(result, sequence)
 
- if ( log.isTraceEnabled ) {
- log.trace(s"Writing message ${message.getClass.getName} - \n${BufferDumper.dumpAsHex(result)}")
- }
+ sequence += 1
 
-  out.add(result)
- }
+ if ( log.isTraceEnabled ) {
+ log.trace(s"Writing message ${message.getClass.getName} - \n${BufferDumper.dumpAsHex(result)}")
  }
 
+ out.add(result)
  }
 
 }

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

@@ -0,0 +1,59 @@
+package com.github.mauricio.async.db.mysql.codec
+
+import java.nio.charset.Charset
+
+import com.github.mauricio.async.db.mysql.binary.BinaryRowEncoder
+import com.github.mauricio.async.db.mysql.message.client.{ClientMessage, SendLongDataMessage}
+import com.github.mauricio.async.db.util.{Log, ByteBufferUtils}
+import io.netty.buffer.{Unpooled, ByteBuf}
+import io.netty.channel.ChannelHandlerContext
+import io.netty.handler.codec.MessageToMessageEncoder
+
+object SendLongDataEncoder {
+ val log = Log.get[SendLongDataEncoder]
+}
+
+class SendLongDataEncoder(charset: Charset)
+ extends MessageToMessageEncoder[SendLongDataMessage](classOf[SendLongDataMessage]) {
+
+ import com.github.mauricio.async.db.mysql.codec.SendLongDataEncoder.log
+
+ private final val rowEncoder = new BinaryRowEncoder(charset)
+
+ def isLong(value: Any): Boolean = rowEncoder.encoderFor(value).isLong(value)
+
+ def encode(ctx: ChannelHandlerContext, message: SendLongDataMessage, out: java.util.List[Object]): Unit = {
+ val result: ByteBuf = encode(message)
+
+ ByteBufferUtils.writePacketLength(result, 0)
+
+ if ( log.isTraceEnabled ) {
+ log.trace(s"Writing message ${message.toString}")
+ }
+
+ out.add(result)
+ }
+
+ private def encode(message: SendLongDataMessage): ByteBuf = {
+ val buffer = ByteBufferUtils.packetBuffer()
+ buffer.writeByte(ClientMessage.PreparedStatementSendLongData)
+ buffer.writeBytes(message.statementId)
+ buffer.writeShort(message.paramId)
+
+ Unpooled.wrappedBuffer(buffer, encodeValue(message.value))
+ }
+
+ private def encodeValue(maybeValue: Any) : ByteBuf = {
+ if ( maybeValue == null || maybeValue == None ) {
+ throw new UnsupportedOperationException("Cannot encode NULL as long value")
+ } else {
+ val value = maybeValue match {
+ case Some(v) => v
+ case _ => maybeValue
+ }
+ val encoder = rowEncoder.encoderFor(value)
+ encoder.encodeLong(value)
+ }
+ }
+
+}

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

@@ -3,8 +3,6 @@ package com.github.mauricio.async.db.mysql.message.client
 case class SendLongDataMessage (
  statementId : Array[Byte],
  value : Any,
- paramId : Int )
- extends ClientMessage( ClientMessage.PreparedStatementSendLongData ) {
-
+ paramId : Int ) {
  override def toString = "SendLongDataMessage(statementId=" + statementId + ",paramId=" + paramId + ",value.getClass=" + value.getClass.getName +")"
 }