2

我正在尝试创建一个提供 OAuth2 令牌并且还负责刷新过期令牌的 Source。目前我的代码看起来有点像这样

  case class Token(expires: Instant = Instant.now().plus(100, ChronoUnit.MILLIS)){
    def expired = Instant.now().isAfter(expires)
  }

  Source
    .repeat()
    .mapAsync(1){ _ =>
      println("  -> token req")
      // this fakes an async token request to the token service
      Future{
        Thread.sleep(500)
        println("  <- token resp")
        Token()
      }
    }
    .mapAsync(1){ token =>
      println("  -> req with token auth")
      if(token.expired){
        println("!!! Received expired token")
      }
      // this is the actual call that needs the token
      println("making call")
      Future{
        Thread.sleep(2000)
        println("  <- req resp")
        "OK"
      }
    }
    .take(2)
    .runWith(Sink.ignore)
    .recover{case e => ()}
    .flatMap{ _ =>
      system.terminate()
    }

此代码的输出如下所示

root   -> token req
root   <- token resp
root   -> token req
root   -> req with token auth
root making call
root   <- token resp
root   -> token req
root   <- token resp
root   -> token req
root   <- token resp
root   -> token req
root   <- req resp
root   -> req with token auth
root !!! Received expired token
root making call
root   <- token resp
root   -> token req
root   <- token resp
root   -> token req
root   <- token resp
root   <- req resp
root   -> req with token auth
root !!! Received expired token
root making call
root ... finished with exit code 0

显然，这个 mapAsync(1) 在不期望的时候产生了需求（预取？）

有2个问题：

  • 需求导致上游不需要的令牌请求
  • 令牌的预取/缓存是有问题的，因为它们仅在特定的时间内有效

那么如何创建一个行为类似于此函数的真正拉流呢？

def tokenSource: () => Future[Token]

4

2 回答 2

3

如果您有意避免预取和排队，那么我认为scala.collection.immutable.Stream，或 Iterator，是比 akka Stream 更好的解决方案。

下面是一个示例实现，可以避免您在问题中列举的陷阱。（注意：我使用 anActorSystem创建 ExecutionContext，通过dispatcher，以防止应用程序在调用有时间完成之前退出sleep。我正在利用 ActorSystem 不会因为 main 函数到达末尾而关闭的事实的表达式定义。）

import scala.collection.immutable.Stream
import scala.concurrent.Future

object ScalaStreamTest extends App {    
  case class Token(expires: Long = System.currentTimeMillis() + 100){
    def expired = System.currentTimeMillis() > expires
  }

  val actorSystem = akka.actor.ActorSystem()      
  import actorSystem.dispatcher

  def createToken =  Future {
    Thread.sleep(500)
    println("  <- token resp")
    Token()
  }

  def checkExpiration(token : Future[Token]) = token map { t =>
    println("  -> req with token auth")
    if(t.expired){println("!!! Received expired token")}
    t
  }

  def makeCall(token : Future[Token]) = token flatMap { t =>
    println("making call")
    Future {
      Thread.sleep(2000)
      println("  <- req resp")
      "OK"
    }
  }

  val stream = Stream.continually(createToken)
                     .map(checkExpiration)
                     .map(makeCall)
                     .take(2)
                     .force
}//end object ScalaStreamTest

调用是必要的force，因为 Stream 是惰性的，因此强制之前的所有方法调用（即：continuous、map 和 take）也是惰性的。除非调用 reducer 或通过强制明确告知 Stream，否则不会在惰性 Stream 上进行计算。

于 2015-10-30T16:38:23.607 回答
0

Akka Streams 总是预取以保持管道饱和。

为了得到你想要的，我建议你创建一个 Source[Token]，当旧的令牌过期而不是请求时发出新的令牌。然后你用 Tokens 的源压缩你的 Source of data 并使用它的结果。

于 2015-09-23T11:28:20.010 回答