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

如何将结果从一个源流传递到另一个源流

akka-stream akka scala

1

Omnipresent · 技术社区 · 7 年前

我有一个方法可以处理 Source 然后返回。我正在尝试修改它,但似乎无法返回相同的内容:

原件

def originalMethod[as: AS, mat: MAT, ec: EC](checkType: String) 
: Flow[ByteString, MyValidation[MyClass], NotUsed]{
      collectStuff
      .map { ts =>
        val errors = MyEngine.checkAll(ts.code)
        (ts, errors)
      }
      .map { x =>
        x._2
          .leftMap(xs => {
            addInformation(x._1, xs.toList)
          })
          .toEither
      }
}

我正在使用另一个源进行修改,并将结果传递给原始源,但返回相同的内容:

def calculate[T: AS: MAT](source: Source[T, NotUsed]): Future[Seq[T]] = 
{
 source.runWith(Sink.seq)
}


def modifiedMethod[as: AS, mat: MAT, ec: EC](checkType: String, mySource: Source[LoanApplicationRegister, NotUsed]) 
: Flow[ByteString, MyValidation[MyClass], NotUsed]{
  for {
    calc <- calculate(mySource)
    orig <-  collectStuff
        .map { ts =>
          val errors = MyEngine.checkAll(ts.code, calc)
          (ts, errors)
        }
        .map { x =>
          x._2
            .leftMap(xs => {
              addInformation(x._1, xs.toList)
            })
            .toEither
        }
  }
  yield {
    orig
  }
}

但我有编译错误 Expression of type Future[Nothing] doesn't conform to existing type Flow[ByteString, MyValidation[MyClass]

我怎么回去 Flow[ByteString, MyValidation[MyClass] 在我的 modifiedMethod 就像 originalMethod 是

1 回复 | 直到 7 年前

1

0

Arnon Rotem-Gal-Oz 7 年前

  for { calc <- calculate(mySource)}
  yield {
    collectStuff
        .map { ts =>
          val errors = MyEngine.checkAll(ts.code, calc)
          (ts, errors)
        }
        .map { x =>
          x._2
            .leftMap(xs => {
              addInformation(x._1, xs.toList)
            })
            .toEither
        }
  }

会给你一个 Future[Flow[ByteString, MyValidation[MyClass], NotUsed]] 而不是 Future[Nothing] 但是如果你想移除 Future 你需要 Await 在某个地方(当你调用calculate(然后你不需要 for )或之后。通常,这不是利用期货的方法