代码之家 › 专栏 › 技术社区 › Anthony

如何先从一个流获取数据,然后再传递到另一个流

akka-stream akka scala

0

Anthony · 技术社区 · 7 年前

我需要调用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 线

0 回复 | 直到 7 年前

1

0

Alec 7 年前

你有两个问题。首先是 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)
      }
    }
}