如何将结果从一个源流传递到另一个 [英] How to pass results from one source stream to another

问题描述

我有一个处理 Source 并返回的方法。我正在尝试对其进行修改，但似乎无法返回同一内容：

I have a method that processes a Source and returns. I am trying to modify it but can't seem to be able to return the same thing:

原始

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
      }
}

我正在使用其他来源进行修改，并将其结果传递给原始来源并且返回相同的内容：

I am modifying by using another source and pass result of that to the original source and yet return the same thing:

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
  }
}

但是我遇到编译错误 Future [Nothing]类型的表达式不符合现有的Flow [ByteString，MyValidation [MyClass]

如何在我的 modifiedMethod Flow [ByteString，MyValidation [MyClass] $ c>就像 originalMethod 是

How can I return Flow[ByteString, MyValidation[MyClass] in my modifiedMethod just like the originalMethod was

推荐答案

  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]，未使用]] 而不是 Future [Nothing]
，但是如果要删除未来，您需要在某个地方等待（在您调用calculate时（然后您不需要 for ）或之后。通常，这不是使用期货的方法

would give you a Future[Flow[ByteString, MyValidation[MyClass], NotUsed]] instead of Future[Nothing] but if you want to remove the Future you'd need to Await somewhere for it (either when you call calculate (and then you don't need the for) or after it. Usually, that's not the way to use Futures

这篇关于如何将结果从一个源流传递到另一个的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！