如何从 Future 回调中保存 & return 数据
How to save & return data from within Future callback
过去几天我一直面临一个关于在 Scala 中保存和处理来自 Futures 的数据的问题。我对这两种语言和概念都不熟悉。 Lagom 关于 Cassandra 的文档说要实现大约 9 个代码文件，我想确保我的数据库代码在将其分散到那么多代码之前可以正常工作。
具体来说，我目前正在尝试实施概念验证以向 to/fromlagom 为您实施的 cassandra 数据库发送数据。到目前为止，我能够发送和检索数据 to/from 数据库，但我在 returning 数据时遇到了麻烦，因为这一切都是异步运行的，而且 returning 数据return编辑成功。
我已经玩了一段时间了；检索代码如下所示：
override def getBucket(logicalBucket: String) = ServiceCall[NotUsed, String] {
request => {
val returnList = ListBuffer[String]()
println("Retrieving Bucket " + logicalBucket)
val readingFromTable = "SELECT * FROM data_access_service_impl.s3buckets;"
//DB query
var rowsFuture: Future[Seq[Row]] = cassandraSession.selectAll(readingFromTable)
println(rowsFuture)
Await.result(rowsFuture, 10 seconds)
rowsFuture onSuccess {
case rows => {
println(rows)
for (row <- rows) println(row.getString("name"))
for (row <- rows) returnList += row.getString("name")
println("ReturnList: " + returnList.mkString)
}
}
rowsFuture onFailure {
case e => println("An error has occured: " + e.getMessage)
Future {"An error has occured: " + e.getMessage}
}
Future.successful("ReturnList: " + returnList.mkString)
}
}
运行时，我在 onSuccess 回调中将预期的数据库值设置为 'println'。但是，我在 return 语句中使用的同一个变量在回调之外打印为空（并且 return 的数据也是空的）。这也发生在我使用的 'insertion' 函数中，它并不总是 return 我在回调函数中设置的变量。
如果我尝试将语句放在回调函数中，则会收到 'returns Unit, expects Future[String]' 错误。所以我被困在回调函数中无法 return 的地方，所以我不能保证我正在 returning 数据）。
我的目标是 return 到 API 的字符串，以便它显示数据库中所有 s3 存储桶名称的列表。这意味着遍历 Future[Seq[Row]] 数据类型，并将数据保存到连接的字符串中。如果有人可以提供帮助，他们将解决我通过 Lagom、Akka、Datastax 和 Cassandra 文档阅读的 2 周问题。我在这一点上大吃一惊（信息过载）并且我没有找到关于此的明确指南。
作为参考，这里是 cassandraSession 文档：
LagomTutorial/Documentation Style Information with their only cassandra-query example
CassandraSession.scala code
关于 Future、（和 Option、Either 和 Try）要理解的关键是您（通常）不会获取值out，你把计算带入。最常见的方法是使用 map 和 flatMap 方法。
在您的情况下，您想要将 Seq[Row] 转换为 String。但是，您的 Seq[Row] 包含在这个名为 Future 的不透明数据结构中，因此您不能像 实际上 那样只 rows.mkString一个Seq[Row]。因此，与其获取值并对其执行计算，不如将计算 rows.mkString 带到数据中：
//DB query
val rowsFuture: Future[Seq[Row]] = cassandraSession.selectAll(readingFromTable)
val rowToString = (row: Row) => row.getString("name")
val computation = (rows: Seq[Row]) => rows.map(rowToString).mkString
// Computation to the data, rather than the other way around
val resultFuture = rowsFuture.map(computation)
现在，当 rowsFuture 完成时，您通过调用 rowsFuture.map 创建的新未来将实现 调用 computation 的结果 Seq[Row] 你实际上 关心。
那时你可以 return resultFuture 并且一切都会按预期工作，因为调用 getBucket 的代码期待 Future 并将处理它视情况而定。
为什么 Future 不透明？
原因很简单，因为它代表了一个当前可能不存在的值。只有当值存在时您才能获得该值，但是当您开始调用时它不存在。代码允许您注册计算（回调，如 onSuccess 和 onFailure）或使用 map 和 flatMap.
更深层次的原因是因为 Future 是一个 Monad 并且 monad 包含计算，但是 do not have an operation to extract that computation out of them
将 select 替换为您的特定 select 和您想要为您的特定 field.The 示例获取的字段仅用于测试，不是架构建议。
package ldg.com.dbmodule.model
/**
* Created by ldipotet on 05/11/17.
*/
import com.datastax.driver.core.{Cluster, ResultSet, ResultSetFuture}
import scala.util.{Failure, Success, Try}
import java.util.concurrent.TimeUnit
import scala.collection.JavaConversions._
//Use Implicit Global Context is strongly discouraged! we must create
//our OWN execution CONTEXT !
import scala.concurrent.ExecutionContext.Implicits.global
import scala.concurrent.{Future, _}
import scala.concurrent.duration._
object CassandraDataStaxClient {
//We create here a CallBack in Scala with the DataStax api
implicit def resultSetFutureToScala(rf: ResultSetFuture):
Future[ResultSet] = {
val promiseResult = Promise[ResultSet]()
val producer = Future {
getResultSet(rf) match {
//we write a promise with an specific value
case Success(resultset) => promiseResult success resultset
case Failure(e) => promiseResult failure (new
IllegalStateException)
}
}
promiseResult.future
}
def getResultSet: ResultSetFuture => Try[ResultSet] = rsetFuture => {
Try(
// Other choice can be:
// getUninterruptibly(long timeout, TimeUnit unit) throws
TimeoutException
// for an specific time
//can deal an IOException
rsetFuture.getUninterruptibly
)
}
def main(args: Array[String]) {
def defaultFutureUnit() = TimeUnit.SECONDS
val time = 20 seconds
//Better use addContactPoints and adds more tha one contact point
val cluster = Cluster.builder().addContactPoint("127.0.0.1").build()
val session = cluster.connect("myOwnKeySpace")
//session.executeAsync es asyncronous so we'll have here a
//ResultSetFuture
//converted to a resulset due to Implicitconversion
val future: Future[ResultSet] = session.executeAsync("SELECT * FROM
myOwnTable")
//blocking on a future is strongly discouraged!! next is an specific
//case
//to make sure that all of the futures have been completed
val results = Await.result(future,time).all()
results.foreach(row=>println(row.getString("any_String_Field"))
session.close()
cluster.close()
}
}
过去几天我一直面临一个关于在 Scala 中保存和处理来自 Futures 的数据的问题。我对这两种语言和概念都不熟悉。 Lagom 关于 Cassandra 的文档说要实现大约 9 个代码文件，我想确保我的数据库代码在将其分散到那么多代码之前可以正常工作。
具体来说，我目前正在尝试实施概念验证以向 to/fromlagom 为您实施的 cassandra 数据库发送数据。到目前为止，我能够发送和检索数据 to/from 数据库，但我在 returning 数据时遇到了麻烦，因为这一切都是异步运行的，而且 returning 数据return编辑成功。
我已经玩了一段时间了；检索代码如下所示：
override def getBucket(logicalBucket: String) = ServiceCall[NotUsed, String] {
request => {
val returnList = ListBuffer[String]()
println("Retrieving Bucket " + logicalBucket)
val readingFromTable = "SELECT * FROM data_access_service_impl.s3buckets;"
//DB query
var rowsFuture: Future[Seq[Row]] = cassandraSession.selectAll(readingFromTable)
println(rowsFuture)
Await.result(rowsFuture, 10 seconds)
rowsFuture onSuccess {
case rows => {
println(rows)
for (row <- rows) println(row.getString("name"))
for (row <- rows) returnList += row.getString("name")
println("ReturnList: " + returnList.mkString)
}
}
rowsFuture onFailure {
case e => println("An error has occured: " + e.getMessage)
Future {"An error has occured: " + e.getMessage}
}
Future.successful("ReturnList: " + returnList.mkString)
}
}
运行时，我在 onSuccess 回调中将预期的数据库值设置为 'println'。但是，我在 return 语句中使用的同一个变量在回调之外打印为空（并且 return 的数据也是空的）。这也发生在我使用的 'insertion' 函数中，它并不总是 return 我在回调函数中设置的变量。
如果我尝试将语句放在回调函数中，则会收到 'returns Unit, expects Future[String]' 错误。所以我被困在回调函数中无法 return 的地方，所以我不能保证我正在 returning 数据）。
我的目标是 return 到 API 的字符串，以便它显示数据库中所有 s3 存储桶名称的列表。这意味着遍历 Future[Seq[Row]] 数据类型，并将数据保存到连接的字符串中。如果有人可以提供帮助，他们将解决我通过 Lagom、Akka、Datastax 和 Cassandra 文档阅读的 2 周问题。我在这一点上大吃一惊（信息过载）并且我没有找到关于此的明确指南。
作为参考，这里是 cassandraSession 文档：
LagomTutorial/Documentation Style Information with their only cassandra-query example CassandraSession.scala code
关于 Future、（和 Option、Either 和 Try）要理解的关键是您（通常）不会获取值out，你把计算带入。最常见的方法是使用 map 和 flatMap 方法。
在您的情况下，您想要将 Seq[Row] 转换为 String。但是，您的 Seq[Row] 包含在这个名为 Future 的不透明数据结构中，因此您不能像 实际上 那样只 rows.mkString一个Seq[Row]。因此，与其获取值并对其执行计算，不如将计算 rows.mkString 带到数据中：
//DB query
val rowsFuture: Future[Seq[Row]] = cassandraSession.selectAll(readingFromTable)
val rowToString = (row: Row) => row.getString("name")
val computation = (rows: Seq[Row]) => rows.map(rowToString).mkString
// Computation to the data, rather than the other way around
val resultFuture = rowsFuture.map(computation)
现在，当 rowsFuture 完成时，您通过调用 rowsFuture.map 创建的新未来将实现 调用 computation 的结果 Seq[Row] 你实际上 关心。
那时你可以 return resultFuture 并且一切都会按预期工作，因为调用 getBucket 的代码期待 Future 并将处理它视情况而定。
为什么 Future 不透明？
原因很简单，因为它代表了一个当前可能不存在的值。只有当值存在时您才能获得该值，但是当您开始调用时它不存在。代码允许您注册计算（回调，如 onSuccess 和 onFailure）或使用 map 和 flatMap.
更深层次的原因是因为 Future 是一个 Monad 并且 monad 包含计算，但是 do not have an operation to extract that computation out of them
将 select 替换为您的特定 select 和您想要为您的特定 field.The 示例获取的字段仅用于测试，不是架构建议。
package ldg.com.dbmodule.model
/**
* Created by ldipotet on 05/11/17.
*/
import com.datastax.driver.core.{Cluster, ResultSet, ResultSetFuture}
import scala.util.{Failure, Success, Try}
import java.util.concurrent.TimeUnit
import scala.collection.JavaConversions._
//Use Implicit Global Context is strongly discouraged! we must create
//our OWN execution CONTEXT !
import scala.concurrent.ExecutionContext.Implicits.global
import scala.concurrent.{Future, _}
import scala.concurrent.duration._
object CassandraDataStaxClient {
//We create here a CallBack in Scala with the DataStax api
implicit def resultSetFutureToScala(rf: ResultSetFuture):
Future[ResultSet] = {
val promiseResult = Promise[ResultSet]()
val producer = Future {
getResultSet(rf) match {
//we write a promise with an specific value
case Success(resultset) => promiseResult success resultset
case Failure(e) => promiseResult failure (new
IllegalStateException)
}
}
promiseResult.future
}
def getResultSet: ResultSetFuture => Try[ResultSet] = rsetFuture => {
Try(
// Other choice can be:
// getUninterruptibly(long timeout, TimeUnit unit) throws
TimeoutException
// for an specific time
//can deal an IOException
rsetFuture.getUninterruptibly
)
}
def main(args: Array[String]) {
def defaultFutureUnit() = TimeUnit.SECONDS
val time = 20 seconds
//Better use addContactPoints and adds more tha one contact point
val cluster = Cluster.builder().addContactPoint("127.0.0.1").build()
val session = cluster.connect("myOwnKeySpace")
//session.executeAsync es asyncronous so we'll have here a
//ResultSetFuture
//converted to a resulset due to Implicitconversion
val future: Future[ResultSet] = session.executeAsync("SELECT * FROM
myOwnTable")
//blocking on a future is strongly discouraged!! next is an specific
//case
//to make sure that all of the futures have been completed
val results = Await.result(future,time).all()
results.foreach(row=>println(row.getString("any_String_Field"))
session.close()
cluster.close()
}
}