finagle-mysql: Split Request and Protocol Messages

Problem

Finagle models `Service`s as a function from a `Request` to a
`Future` of a `Response`. For each particular protocol, this definition
of a `Service` is publicly accessible, and for finagle-mysql it follows.
However, finagle-mysql's `Request` class is a mixture of both protocol
level messages and user run commands. This can make APIs awkward since
`Request` is public, but the protocol level messages should not be. Also
since `Request` is sealed, all the extensions need to be contained within
the `Request` file.

Solution

Extract a higher trait from `Request` named `ProtocolMessage` which has
a sequence number and can be converted to a `Packet`. This allows protocol
level messages to extend that, and not `Request`. This then leaves user
level commands as `Request`s, and allows us to remove the synthetic
`Command.COM_NO_OP` which is intended to work around the fact that protocol
level messages are not commands.

JIRA Issues: CSL-7990

Differential Revision: https://phabricator.twitter.biz/D327554

CHANGELOG.rst

@@ -90,6 +90,14 @@ Bug Fixes
 * finagle-mysql: Don't log `c.t.f.ChannelClosedException` when rolling back a transaction
   fails. ``PHAB_ID=D327111``
 
+Breaking API Changes
+~~~~~~~~~~~~~~~~~~~~
+
+* finagle-mysql: The structure of `c.t.f.mysql.Request` has changed. It is now based on
+  a higher level `c.t.f.mysql.ProtocolMessage` and the `cmd` field must contain a value.
+  Additionally, the synthetic `Command.COM_NO_OP` has been removed, as due to the
+  restructuring it was no longer necessary. ``PHAB_ID=D327554``
+
 19.5.1
 ------
 

finagle-mysql/src/main/scala/com/twitter/finagle/mysql/Handshake.scala

@@ -64,8 +64,8 @@ private[mysql] abstract class Handshake(
       LostSyncException.const(isCompatibleCharset(handshakeInit)).map(_ => handshakeInit)
     }
 
-  protected final def simpleDispatch(req: Request): Future[Result] =
-    transport.write(req.toPacket).flatMap(_ => transport.read().flatMap(decodeSimpleResult))
+  protected final def messageDispatch(msg: ProtocolMessage): Future[Result] =
+    transport.write(msg.toPacket).flatMap(_ => transport.read().flatMap(decodeSimpleResult))
 
   protected final def decodeSimpleResult(packet: Packet): Future[Result] =
     MysqlBuf.peek(packet.body) match {

finagle-mysql/src/main/scala/com/twitter/finagle/mysql/PlainHandshake.scala

@@ -26,7 +26,7 @@ private[mysql] final class PlainHandshake(
   def connectionPhase(): Future[Result] =
     readHandshakeInit()
       .map(makePlainHandshakeResponse)
-      .flatMap(simpleDispatch)
+      .flatMap(messageDispatch)
       .onFailure(_ => transport.close())
 
 }

finagle-mysql/src/main/scala/com/twitter/finagle/mysql/Request.scala

@@ -7,7 +7,6 @@ import java.util.logging.Logger
 
 object Command {
   val COM_POISON_CONN: Byte = (-2).toByte // used internally to close an underlying connection
-  val COM_NO_OP: Byte = (-1).toByte // used internally by this client
   val COM_SLEEP: Byte = 0x00.toByte // internal thread state
   val COM_QUIT: Byte = 0x01.toByte // mysql_close
   val COM_INIT_DB: Byte = 0x02.toByte // mysql_select_db
@@ -39,12 +38,26 @@ object Command {
   val COM_STMT_FETCH: Byte = 0x1C.toByte // mysql_stmt_fetch
 }
 
-sealed trait Request {
+/**
+ * A `ProtocolMessage` is an outgoing message sent by the
+ * client as part of the MySQL protocol. It contains a
+ * sequence number and is able to be converted to a MySQL
+ * packet to be sent across the wire.
+ */
+private[mysql] trait ProtocolMessage {
   def seq: Short
-  def cmd: Byte = Command.COM_NO_OP
   def toPacket: Packet
 }
 
+/**
+ * A `Request` is an outgoing message sent by the client
+ * as part of the MySQL protocol. It is packet-based and
+ * based around a MySQL command.
+ */
+sealed trait Request extends ProtocolMessage {
+  def cmd: Byte
+}
+
 /**
  * Contains the SQL for a [[Request]].
  */
@@ -54,15 +67,15 @@ sealed trait WithSql {
 
 private[finagle] object PoisonConnectionRequest extends Request {
   def seq: Short = 0
-  override def cmd: Byte = Command.COM_POISON_CONN
+  def cmd: Byte = Command.COM_POISON_CONN
   def toPacket: Packet = ???
 }
 
 /**
  * A command request is a request initiated by the client
  * and has a cmd byte associated with it.
  */
-abstract class CommandRequest(override val cmd: Byte) extends Request {
+abstract class CommandRequest(val cmd: Byte) extends Request {
   def seq: Short = 0
 }
 
@@ -129,12 +142,12 @@ private[mysql] case class SslConnectionRequest(
   clientCap: Capability,
   charset: Short,
   maxPacketSize: Int)
-    extends Request {
+    extends ProtocolMessage {
   require(
     clientCap.has(Capability.SSL),
     "Using SslConnectionRequest requires having the SSL capability")
 
-  override def seq: Short = 1
+  def seq: Short = 1
 
   def toPacket: Packet = {
     val packetBodySize = 32
@@ -162,7 +175,7 @@ private[mysql] sealed abstract class HandshakeResponse(
   serverCap: Capability,
   charset: Short,
   maxPacketSize: Int)
-    extends Request {
+    extends ProtocolMessage {
 
   lazy val hashPassword: Array[Byte] = password match {
     case Some(p) => encryptPassword(p, salt)
@@ -230,7 +243,7 @@ private[mysql] case class PlainHandshakeResponse(
       charset,
       maxPacketSize) {
 
-  override def seq: Short = 1
+  def seq: Short = 1
 }
 
 /**
@@ -266,7 +279,7 @@ private[mysql] case class SecureHandshakeResponse(
     clientCap.has(Capability.SSL),
     "clientCap must contain Capability.SSL to send a SecureHandshakeResponse")
 
-  override def seq: Short = 2
+  def seq: Short = 2
 }
 
 class FetchRequest(val prepareOK: PrepareOK, val numRows: Int)