【问题标题】:How to pass results from one source stream to another如何将结果从一个源流传递到另一个源流
【发布时间】:2018-12-16 12:43:31
【问题描述】:

我有一个处理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]

我怎样才能在我的modifiedMethod 中返回Flow[ByteString, MyValidation[MyClass],就像originalMethod 一样

【问题讨论】:

    标签: scala akka akka-stream


    【解决方案1】:
      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(或者当你调用计算时(然后你不需要for)或者在它之后。通常,这不是方法使用期货

    【讨论】:

    • 我明白了,您是在建议该方法返回 Future[Flow[ByteString, MyValidation[MyClass], NotUsed]] 而不是 Flow[ByteString, MyValidation[MyClass], NotUsed]
    • 如果我将calculate 包裹在onComplete 中,然后运行处于产量中的部分,是否保证在onComplete 完成后运行?
    • modifiedMethod 是从这样的地方调用的:rawData(id).take(1).via(modifiedMethod(checkType, mySource).zip(Source.fromIterator() =&gt; Iterator.from(1))).collect{...}。如果我返回 Future 调用该方法的部分会抱怨 expected Graph[FlowShape[ByteString, NotInferedT], NotInferedMat2] actual: Future[Flow[ByteString, MyValidation[MyClass], NotUsed] 有没有办法解决这个问题?
    • 您可以通过返回 a 来避免 Future 将计算作为流程中的另一个步骤的一部分(例如,使用 Flow[Seq[T]].mapAsync 或类似的东西执行它 - 尽管您可能应该在另一个问题中提出更多细节)
    • 我有一个单独的问题想看看如何从这个方法返回 Future。 stackoverflow.com/questions/53799829/…
    猜你喜欢
    • 2010-12-24
    • 2021-05-26
    • 1970-01-01
    • 1970-01-01
    • 1970-01-01
    • 1970-01-01
    • 1970-01-01
    • 2011-06-03
    • 2023-02-01
    相关资源
    最近更新 更多