Scala slick query with Akka & Postgres

Question

Я пытаюсь написать метод, который просто удаляет строку из базы данных на основе идентификатора.

class PolicyHolderDAO(database: DatabaseDef) extends CRUDActor[PolicyHolder] { 

    private val policyHolders: TableQuery[PolicyHolderTable] = TableQuery[PolicyHolderTable] 
    implicit val system: ActorSystem = ActorSystem("Bitcoin-Insurance") 
    import system.dispatcher 
    implicit val timeout: Timeout = Timeout(5.seconds) 
    private implicit var session = database.createSession 

    override def receive = { 
    case PolicyHolderDAO.Read(id) => sender ! read(id) 
    case PolicyHolderDAO.Create(policyHolder) => sender ! create(policyHolder) 
    case PolicyHolderDAO.Delete(policyHolder) => sender ! delete(policyHolder) 
    } 

    /** 
    * @param policyHolder the policy holder to inserted into the database 
    * @return id 
    */ 
    override def create(policyHolder: PolicyHolder): Future[PolicyHolder] = { 
    future { 
     (policyHolders returning policyHolders.map(_.id) into 
     ((policyHolder, id) => policyHolder.copy(id = Some(id)))) += policyHolder 

    } 
    } 
    /** 
    * @param id the id that corresponds to a policy holder 
    * @return a future value of a policy holder if the policy holder exists in the database, else it returns none 
    */ 
    override def read(id: Future[Long]): Future[Option[PolicyHolder]] = { 
    id.map(i => policyHolders.filter(p => p.id === i).firstOption) 
    } 

    /** 
    * @param policyHolder the policyHolder to be updated 
    * @return policyHolder the policyHolders information now saved in the database 
    */ 
    override def update(policyHolder: Future[PolicyHolder]): Future[Option[PolicyHolder]] = { 
    /* //val policyHolderFromDb = policyHolder.map(p => policyHolders.filter(_.id === p.id.getOrElse(-1))) 
    val policyHolderFromDb = for (p <- policyHolder; q = policyHolders.filter(_.id === p.id)) yield q.update 

    val updatedPolicyHolder: Future[Option[PolicyHolder]] = for (p <- policyHolderFromDb; result = create(p)) yield result 
    updatedPolicyHolder*/ 
    Future(None) 
    } 
    /** 
    * @param policyHolder the policy holder to be deleted from our database 
    * @return affectedRows the number of rows effected by this query 
    */ 
    override def delete(policyHolder: Future[PolicyHolder]): Future[Int] = { 
    val policyHolderId: Future[Long] = policyHolder.map(p => p.id.getOrElse(-1)) 
    val affectedRows = for (id <- policyHolderId; q = policyHolders.filter(_.id === id)) yield q 
    affectedRows.map(q => q.delete) 

    } 
} 

Я пытаюсь проверить это с помощью этого теста, написанного для Scala теста

"A PolicyHolderDAO Actor" must { 
    "be able to delete an existing policy holder from our database" in { 
     val policyHolder = PolicyHolder(None, "Chris", "Stewart") 
     val createdPolicyHolderAny: Future[Any] = policyHolderDAOActor ? PolicyHolderDAO.Create(policyHolder) 
     val createdPolicyHolder: Future[PolicyHolder] = createdPolicyHolderAny.mapTo[Future[PolicyHolder]].flatMap(p => p) 

     policyHolderDAOActor ! PolicyHolderDAO.Delete(createdPolicyHolder) 

     val deletedPolicyHolderAny: Future[Any] = policyHolderDAOActor ? PolicyHolderDAO.Get(createdPolicyHolder.map(_.id)) 
     val deletedPolicyHolder: Future[Option[PolicyHolder]] = deletedPolicyHolderAny.mapTo[Future[Option[PolicyHolder]]].flatMap(p => p) 
     whenReady(deletedPolicyHolder, timeout(10 seconds), interval(5 millis)) { p => 
     val policyHolderExists = p match { 
      case Some(a) => 
      println(a) 
      true 
      case None => false 
     } 
     policyHolderExists must be(false) 
     } 
    } 
    } 

Однако я уверен, не этот тест. Причина в том, что строка НЕ ​​удаляется из нашей базы данных. Я не уверен, почему этот тест, если он не работает. У меня есть соответствующий модульный тест для этого метода, который возвращает 1 затронутую строку в результате удаления, что имеет смысл. Может ли быть что-то, что я не понимаю с Akka/Futures?

Спасибо!

Answer 1

Существует вероятность состояние гонки в тесте здесь:

... 
policyHolderDAOActor ! PolicyHolderDAO.Delete(createdPolicyHolder) 

val deletedPolicyHolderAny: Future[Any] = policyHolderDAOActor ? PolicyHolderDAO.Get(createdPolicyHolder.map(_.id)) 
... 

это очень трудно сказать, так как вы не выставили Actor, который это делает, но, основываясь на том, что ваш метод выше возвращает Future[Int]delete, похоже, что происходит это:

• policyHolderDAOActor передается сообщение об удалении в Телль.
• policyHolderDAOActor затем сразу же запрашивается для получения той же записи по идентификатору.
• policyHolderDAOActor вызывает метод delete, который возвращает Future[Int], поэтому результат, вероятно, будет отброшен, и receive немедленно вернется.
• policyHolderDAOActor затем может обрабатывать следующее сообщение, чтобы получить запись по идентификатору, но удаление еще не завершило обработку, поэтому запись все еще существует.

Вводя ожидания между двумя линиями выше, вероятно сделать тестовый проход, но если это не так, возможно, разделяя некоторые из Actor кода будет более показательным.

Если другие сообщения функционируют аналогичным образом, также возможно, что запись еще не была создана, когда вы пытаетесь ее удалить. Вышеупомянутые пули по-прежнему будут правильными, только с помощью разных методов. Чтобы быть в безопасности, вы должны блокировать каждую асинхронную операцию в своем тесте, чтобы убедиться, что все происходит последовательно.

Answer 2

Я думаю, ваша проблема с прохождением Slick сессии в нужное место:

1. удаления не имеет список 2-й параметр, содержащий (implicit val session: Session)
2. , когда актер готов, сеанс, вероятно, закрыта. Поэтому вам нужно явно получить новый.

Я не уверен в 1), потому что я привык использовать Slick с Play, которая может иметь несколько иной API, но у меня есть некоторый опыт работы с 2):

https://github.com/payola/payola-viz/blob/b5668bcbdefc3bce7e6095d211fa04c2aa697685/src/app/model/services/LDVMServiceImpl.scala#L44

Answer 3

Просто побочное замечание, почему бы не сделать delete проще?

def delete(policyHolder: Future[PolicyHolder]): Future[Int] = { 
    for (p <- policyHolder) yield policyHolders.delete(_.id === p.id) 
}