How to use transactions with finagle-mysql

build.sbt

name := "finagle-mysql-transactions"

version := "1.0"

scalaVersion := "2.11.5"

libraryDependencies ++= Seq(
  "com.twitter" %% "twitter-server" % "1.9.0",
  "com.twitter" %% "finagle-mysql" % "6.28.0",
  "mysql" % "mysql-connector-java" % "5.1.34"
).map(_.exclude("com.google.code.findbugs", "jsr305"))

mainClass in (Compile, run) := Some("transactions.Main")

Db.scala

package transactions

import com.twitter.finagle.exp.mysql.{Client, OK, Result}

class Db(mysqlClient: Client) {

    val insertOrder     = "INSERT INTO orders (ref) VALUES(?)"
    val insertOrderItem = "INSERT INTO order_items (order_id, item_id) VALUES(?, ?)"

    def persistOrderWithItems[T](order: Order)(whenDone: (OK, Seq[Result]) => T): Future[T] = {
      mysqlClient.transaction[Result] { tx =>
        for {
          insert <- tx.prepare(insertOrder)(order.ref)
        } yield insert match {
          case orderResult: OK =>
            val orderId = orderResult.insertId

            Future.collect(order.items.map { item =>
              tx.prepare(insertOrderItem)(orderId, item.itemId)
            }).map { results =>
              whenDone(orderResult, results)
            }
          case r =>
            Future.exception(new Exception(s"Wrong result type [${r.getClass.getName}]"))
        }
      }
    }
  }

Main.scala

package transactions

import java.net.InetSocketAddress

import com.twitter.finagle.exp.Mysql
import com.twitter.finagle.{Http, Service}
import com.twitter.finagle.exp.mysql.{Client, OK, Result}
import com.twitter.finagle.http.Response
import com.twitter.logging.Logger
import com.twitter.server.TwitterServer
import com.twitter.util.{Await, Future}
import org.jboss.netty.handler.codec.http.{HttpResponseStatus, HttpResponse, HttpRequest}

case class OrderItem(itemId: Int, quantity: Int)
case class Order(ref: String, items: Seq[OrderItem])

object Main {
  val logger = Logger.get(getClass)

  val db = new Db(Mysql.client.
      withCredentials("root", "root").
      withDatabase("transactions").
      newRichClient("127.0.0.1:3306"))

  val myOrder = Order("My order", Seq(
    OrderItem(1, 4),
    OrderItem(2, 1)
  ))
  
  db.persistOrderWithItems(myOrder) { (orderResult, itemsResult) =>
    logger.info(s"Created new order ${orderResult.insertId} with ${itemsResult.size} items")
  }
}

schema.sql

DROP TABLE IF EXISTS `orders`;
CREATE TABLE `orders` (
    `id`    INT           NOT NULL AUTO_INCREMENT,
    `ref`   VARCHAR(255)  NOT NULL,
    PRIMARY KEY (`id`)
) ENGINE=InnoDB;

DROP TABLE IF EXISTS `order_items`;
CREATE TABLE `order_items` (
    `order_id`  INT      NOT NULL,
    `item_id`   INT      NOT NULL,
    `quantity`  INT      NOT NULL,
    PRIMARY KEY (`order_id`, `item_id`)
) ENGINE=InnoDB;