Scala WS-Aufrufe in einem bestimmten ExecutionContext versuchen

Question

Ich habe ein Projekt, das HTTP-Aufrufe an zwei separate APIs ausführt. Die Aufrufe an diese beiden APIs müssen separat eingeschränkt werden. Ich habe mit den Aufrufen zu einem der APIs begonnen und versuche, einen benutzerdefinierten ExecutionContext zu verwenden, um dies zu erreichen. Hier ist mein application.conf:

play.modules.enabled += "playtest.PlayTestModule" 

my-context { 
    fork-join-executor { 
    parallelism-min = 10 
    parallelism-max = 10 
    } 
} 

Dies ist die scala Klasse verwende ich zu testen, ob es funktioniert:

@Singleton 
class MyWsClient @Inject() (client: WSClient, akkaSystem: ActorSystem) { 

    val myExecutionContext: ExecutionContext = akkaSystem.dispatchers.lookup("my-context") 
    val i = new AtomicInteger(0) 

    def doThing: Future[Int] = { 
     Future { 
     println(i.incrementAndGet) 
     println("Awaiting") 
     Await.result(client.url("http://localhost:9000/test").get, Duration.Inf) 
     println("Done") 
     i.decrementAndGet 
     1 
     }(myExecutionContext) 
    } 
} 

Doch egal, was ich versuche, die Anzahl paralleler Anrufe übersteigt die Grenzen, die ich in der application.conf gesetzt habe. Aber es kommt noch Fremde, denn wenn ich die Linie

Await.result(client.url("http://localhost:9000/test").get, Duration.Inf) 

mit

Thread.sleep(1000) 

ersetzen die Grenzen respektiert werden und die Rate ist richtig begrenzt.

Was mache ich falsch und wie kann ich es beheben? Wenn es eine andere Möglichkeit der Ratenbegrenzung mit der scala-ws-Bibliothek gibt, würde ich es gerne hören.

Answer 1

Ich verstehe, dass Sie weiter verwenden möchten scala-ws ok, aber was ist mit etwas, das nicht auf spezifische ExecutionContext verwenden?

Wenn Sie damit einverstanden sind, ist hier eine Idee ... Sie erstellen eine RateLimitedWSClient Komponente, die Sie in Ihre Controller anstelle von WSClient injizieren werden. Diese Komponente sollte ein Singleton sein und eine einzige Methode unterstützen def rateLimit[R](rateLimitClass: String)(request: WSClient => Future[R]). Die rateLimitClass soll angeben, welche Rate auf den aktuellen request gelten soll, da Sie die Anforderungen für unterschiedliche API-Raten unterschiedlich festlegen müssen. Die request Funktion sollte offensichtlich sein.

nun für die Umsetzung meines Vorschlag ist, einen einfachen akka-Stream zu verwenden, das wird Rohr Ihrer request s durch die tatsächlichen WSClient während geschwindigkeitsbestimmend mit der throttle Flussstufe (https://doc.akka.io/docs/akka/current/scala/stream/stages-overview.html#throttle):

val client: WSClient = ??? // injected into the component 
// component initialization, for example create one flow per API 
val queue = 
    Source 
    .queue[(Promise[_], WSClient => Future[_])](...) // keep this materialized value 
    .throttle(...) 
    .map { (promise, request) => 
    promise.completeWith(request(client)) 
    } 
    .to(Sink.ignore) 
    .run() // You have to get the materialized queue out of here! 

def rateLimit[R](rateLimitClass: String)(request: WSClient => Future[R]): Future[R] = { 
    val result = Promise.empty[R] 
    // select which queue to use based on rateLimitClass 
    if (rateLimitClass == "API1") 
    queue.offer(result -> request) 
    else ??? 
    result.future 
} 

Die oben ist nur grobe code, ich hoffe du kommst auf die idee. Sie können natürlich etwas anderes als eine Warteschlange wählen, oder wenn Sie die Warteschlange behalten, müssen Sie entscheiden, wie Sie mit Überläufen umgehen ...