我需要调用cassandra来获取日期,然后将获取的日期传递给另一个要将数据插入数据库的流。
def fetchDate: Future[Done] =
readJournal(persistenceKey)
.drop(3)
.take(1)
.map(l => l.mydate)
.runWith(Sink.ignore)
def insertRowsToDb: Future[Done] =
readJournal(somePersistenceKey)
.drop(4)
.take(1)
.map(data => MyClass(data))
.mapAsync(1) { myData =>
for {
insert <- myRepository.insert(data.id, fetchDate) //error here because fetchDate is unavailable
}
}
class MyRepository(tableName: String) {
def insert(id: String, fetchedDate: Long): Future[Int] =
config.db.run {
sqlu"""INSERT INTO #${tableName}
VALUES (
${id},
${fetchedDate}
)
"""
}
题
fetchDate
,然后将结果传递给myRepository.insert
线?你有两个问题。首先是fetchDate
甚至没有约会!修复这意味着运行Sink
不仅仅是“忽略”:
def fetchDateFut: Future[Date] =
readJournal(persistenceKey)
.drop(3)
.take(1)
.map(l => l.mydate)
.runWith(Sink.last)
然后,你需要flatMap
你的Future
进入该日期的范围:
def insertRowsToDb: Future[Done] = fetchDateFut.flatMap { fetchDate: Date =>
readJournal(somePersistenceKey)
.drop(4)
.take(1)
.map(data => MyClass(data))
.mapAsync(1) { myData =>
for {
insert <- myRepository.insert(data.id, fetchDate)
}
}
}