scala - 使用简单的 SQL 将记录插入事务中的 3 个表时出现异常

Question

我的情况如下:

我的 PG DB 中有 3 个表(用户、安全、密码)。

当最终用户尝试将 self 注册到我的数据库中时，代码将检查用户表中是否存在该用户。

如果存在，只返回一个 None。否则，在同一事务中将记录插入(用户、安全、密码)中。

以下是我的代码:

def selectUserCountQuery1(user: User): DBIO[Int] = {
  sql"""SELECT COUNT(user_id) FROM users WHERE provider_id = ${user.providerId} AND phone_num = ${user.phoneNumber}""".as[Int].head
}


def insertUserQuery(user: User): DBIO[Int] = {
    sqlu"""INSERT INTO users VALUES (${user.userId}, ${user.deviceToken},${user.phoneNumber},${user.providerId},
        ${user.gender},${user.email},${user.createTime},${user.loginName},${user.nickName},${user.currentPlatform},${user.roles},${user.avatarUrl})"""
}

def insertPasswordQuery(passwordInfo: PasswordInfo): DBIO[Int] = {
  sqlu"""INSERT INTO password_info VALUES (${passwordInfo.userId}, ${passwordInfo.hasher},
         ${passwordInfo.passwordHash}, ${passwordInfo.salt})"""
}

def insertSecurityTokenQuery(accessToken: UserSecurityToken): DBIO[Int] = {
  sqlu"""INSERT INTO user_security_token VALUES (${accessToken.userId}, ${accessToken.platform},
         ${accessToken.token}, ${accessToken.token_time})"""
}


def registerUser (user: User, password: String): Future[Option[UserProfile]] = {
  val passwordHash = MD5PasswordGenerator.generateMD5WithEncoder(password)
  val passwordInfo : PasswordInfo = PasswordInfo(user.userId, Some("MD5"), passwordHash, Some(""))
  val tokenStr: String = JwtTokenSecurity.jwtTokenString(user.userId, user.currentPlatform)
  val accessToken : UserSecurityToken = UserSecurityToken(user.userId, user.currentPlatform, tokenStr, 1)

  val actions = for {
    count <- selectUserCountQuery1(user) if count == 0
    status1 <- insertUserQuery(user)
    status2 <- insertPasswordQuery(passwordInfo)
    status3 <- insertSecurityTokenQuery(accessToken)
  } yield(count)

  db.run(actions.transactionally).map {
    count  => count match {
      case 0 => Some(UserProfile.user2UserProfile(user, passwordInfo, accessToken))
      case _ => None
    }
  }
}

当用户第一次注册信息时，该代码可以工作，但在第二次时会抛出错误。 (应该返回 None 以服务)。

错误是:

--- (RELOAD) ---

[info] - play.api.libs.concurrent.ActorSystemProvider - Shutdown application default Akka system: application
[warn] - com.zaxxer.hikari.HikariConfig - validationTimeout is greater than connectionTimeout, setting validationTimeout to connectionTimeout.
[info] - play.api.libs.concurrent.ActorSystemProvider - Starting application default Akka system: application
[info] - play.api.Play - Application started (Dev)
do something before the action is executed
[warn] - com.zaxxer.hikari.HikariConfig - validationTimeout is greater than connectionTimeout, setting validationTimeout to connectionTimeout.
[error] - play.core.server.netty.PlayDefaultUpstreamHandler - Cannot invoke the action
java.util.NoSuchElementException: Action.withFilter failed
  at slick.dbio.DBIOAction$$anonfun$withFilter$1.apply(DBIOAction.scala:94) ~[slick_2.11.jar:na]
  at slick.dbio.DBIOAction$$anonfun$withFilter$1.apply(DBIOAction.scala:94) ~[slick_2.11.jar:na]
  at slick.backend.DatabaseComponent$DatabaseDef$$anonfun$runInContext$1.apply(DatabaseComponent.scala:146) ~[slick_2.11.jar:na]
  at slick.backend.DatabaseComponent$DatabaseDef$$anonfun$runInContext$1.apply(DatabaseComponent.scala:146) ~[slick_2.11.jar:na]
  at scala.concurrent.Future$$anonfun$flatMap$1.apply(Future.scala:251) ~[scala-library-2.11.6.jar:na]
  at scala.concurrent.Future$$anonfun$flatMap$1.apply(Future.scala:249) ~[scala-library-2.11.6.jar:na]
  at scala.concurrent.impl.CallbackRunnable.run(Promise.scala:32) [scala-library-2.11.6.jar:na]
  at akka.dispatch.BatchingExecutor$AbstractBatch.processBatch(BatchingExecutor.scala:55) [akka-actor_2.11.jar:na]
  at akka.dispatch.BatchingExecutor$BlockableBatch$$anonfun$run$1.apply$mcV$sp(BatchingExecutor.scala:91) [akka-actor_2.11.jar:na]
  at akka.dispatch.BatchingExecutor$BlockableBatch$$anonfun$run$1.apply(BatchingExecutor.scala:91) [akka-actor_2.11.jar:na]
  at akka.dispatch.BatchingExecutor$BlockableBatch$$anonfun$run$1.apply(BatchingExecutor.scala:91) [akka-actor_2.11.jar:na]
  at scala.concurrent.BlockContext$.withBlockContext(BlockContext.scala:72) [scala-library-2.11.6.jar:na]
  at akka.dispatch.BatchingExecutor$BlockableBatch.run(BatchingExecutor.scala:90) [akka-actor_2.11.jar:na]
  at akka.dispatch.TaskInvocation.run(AbstractDispatcher.scala:40) [akka-actor_2.11.jar:na]
  at akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(AbstractDispatcher.scala:397) [akka-actor_2.11.jar:na]
  at scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260) [scala-library-2.11.6.jar:na]
  at scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339) [scala-library-2.11.6.jar:na]
  at scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979) [scala-library-2.11.6.jar:na]
  at scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107) [scala-library-2.11.6.jar:na]
[error] - application - 

! @6n1mk1oih - Internal server error, for (POST) [/register] ->

play.api.http.HttpErrorHandlerExceptions$$anon$1: Execution exception[[NoSuchElementException: Action.withFilter failed]]
  at play.api.http.HttpErrorHandlerExceptions$.throwableToUsefulException(HttpErrorHandler.scala:265) ~[play_2.11-2.4.1.jar:2.4.1]
  at play.api.http.DefaultHttpErrorHandler.onServerError(HttpErrorHandler.scala:191) ~[play_2.11-2.4.1.jar:2.4.1]
  at play.api.GlobalSettings$class.onError(GlobalSettings.scala:179) [play_2.11-2.4.1.jar:2.4.1]
  at play.api.mvc.WithFilters.onError(Filters.scala:93) [play_2.11-2.4.1.jar:2.4.1]
  at play.api.http.GlobalSettingsHttpErrorHandler.onServerError(HttpErrorHandler.scala:94) [play_2.11-2.4.1.jar:2.4.1]
  at play.core.server.netty.PlayDefaultUpstreamHandler$$anonfun$3.applyOrElse(PlayDefaultUpstreamHandler.scala:273) [play-netty-server_2.11-2.4.1.jar:2.4.1]
  at play.core.server.netty.PlayDefaultUpstreamHandler$$anonfun$3.applyOrElse(PlayDefaultUpstreamHandler.scala:269) [play-netty-server_2.11-2.4.1.jar:2.4.1]
  at scala.concurrent.Future$$anonfun$recoverWith$1.apply(Future.scala:344) [scala-library-2.11.6.jar:na]
  at scala.concurrent.Future$$anonfun$recoverWith$1.apply(Future.scala:343) [scala-library-2.11.6.jar:na]
  at scala.concurrent.impl.CallbackRunnable.run(Promise.scala:32) [scala-library-2.11.6.jar:na]
  at play.api.libs.iteratee.Execution$trampoline$.executeScheduled(Execution.scala:109) [play-iteratees_2.11-2.4.1.jar:2.4.1]
  at play.api.libs.iteratee.Execution$trampoline$.execute(Execution.scala:71) [play-iteratees_2.11-2.4.1.jar:2.4.1]
  at scala.concurrent.impl.CallbackRunnable.executeWithValue(Promise.scala:40) [scala-library-2.11.6.jar:na]
  at scala.concurrent.impl.Promise$DefaultPromise.tryComplete(Promise.scala:248) [scala-library-2.11.6.jar:na]
  at scala.concurrent.Promise$class.complete(Promise.scala:55) [scala-library-2.11.6.jar:na]
  at scala.concurrent.impl.Promise$DefaultPromise.complete(Promise.scala:153) [scala-library-2.11.6.jar:na]
  at scala.concurrent.Future$$anonfun$map$1.apply(Future.scala:235) [scala-library-2.11.6.jar:na]
  at scala.concurrent.Future$$anonfun$map$1.apply(Future.scala:235) [scala-library-2.11.6.jar:na]
  at scala.concurrent.impl.CallbackRunnable.run(Promise.scala:32) [scala-library-2.11.6.jar:na]
  at akka.dispatch.BatchingExecutor$AbstractBatch.processBatch(BatchingExecutor.scala:55) [akka-actor_2.11.jar:na]
  at akka.dispatch.BatchingExecutor$BlockableBatch$$anonfun$run$1.apply$mcV$sp(BatchingExecutor.scala:91) [akka-actor_2.11.jar:na]
  at akka.dispatch.BatchingExecutor$BlockableBatch$$anonfun$run$1.apply(BatchingExecutor.scala:91) [akka-actor_2.11.jar:na]
  at akka.dispatch.BatchingExecutor$BlockableBatch$$anonfun$run$1.apply(BatchingExecutor.scala:91) [akka-actor_2.11.jar:na]
  at scala.concurrent.BlockContext$.withBlockContext(BlockContext.scala:72) [scala-library-2.11.6.jar:na]
  at akka.dispatch.BatchingExecutor$BlockableBatch.run(BatchingExecutor.scala:90) [akka-actor_2.11.jar:na]
  at akka.dispatch.TaskInvocation.run(AbstractDispatcher.scala:40) [akka-actor_2.11.jar:na]
  at akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(AbstractDispatcher.scala:397) [akka-actor_2.11.jar:na]
  at scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260) [scala-library-2.11.6.jar:na]
  at scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339) [scala-library-2.11.6.jar:na]
  at scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979) [scala-library-2.11.6.jar:na]
  at scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107) [scala-library-2.11.6.jar:na]
Caused by: java.util.NoSuchElementException: Action.withFilter failed
  at slick.dbio.DBIOAction$$anonfun$withFilter$1.apply(DBIOAction.scala:94) ~[slick_2.11.jar:na]
  at slick.dbio.DBIOAction$$anonfun$withFilter$1.apply(DBIOAction.scala:94) ~[slick_2.11.jar:na]
  at slick.backend.DatabaseComponent$DatabaseDef$$anonfun$runInContext$1.apply(DatabaseComponent.scala:146) ~[slick_2.11.jar:na]
  at slick.backend.DatabaseComponent$DatabaseDef$$anonfun$runInContext$1.apply(DatabaseComponent.scala:146) ~[slick_2.11.jar:na]
  at scala.concurrent.Future$$anonfun$flatMap$1.apply(Future.scala:251) ~[scala-library-2.11.6.jar:na]
  at scala.concurrent.Future$$anonfun$flatMap$1.apply(Future.scala:249) ~[scala-library-2.11.6.jar:na]
  … 13 common frames omitted

如果我摆脱了 Guard 条件(计数 ==0)，这太奇怪了，没有异常(exception)。但是用户信息会不断地插入到数据库中。

有没有专家帮帮我？

Answer 1

守卫DBIO.withFilter像 Future.withFilter : 不匹配谓词失败并返回 NoSuchElementException .

相反，您可以使用 flatMap执行你的分支:

val actions = selectUserCountQuery1(user).flatMap {
  case 0 => for {
    status1 <- insertUserQuery(user)
    status2 <- insertPasswordQuery(passwordInfo)
    status3 <- insertSecurityTokenQuery(accessToken)
  } yield 0

  case n => DBIO.successful(n)
}

db.run(actions.transactionally).map {
  case 0 => Some(UserProfile.user2UserProfile(user, passwordInfo, accessToken))
  case _ => None
}