Alright, I now have this version that compiles:
import akka.stream._
import akka.stream.scaladsl.GraphDSL.Implicits
import akka.stream.scaladsl.{Flow, GraphDSL, Keep, Merge, Sink, Source}
val routeFlow: Flow[HttpRequest, HttpResponse, Any] = route
val awesomeFlow =
Flow[HttpRequest]
.groupBy(ServerSettings(system).maxConnections, req ⇒ {
req.uri
})
.throttle(1, FiniteDuration(1, TimeUnit.SECONDS), 1, akka.stream.
ThrottleMode.Shaping)
.mergeSubstreams
.via(routeFlow)
Http()
.bindAndHandle(awesomeFlow, "0.0.0.0", 8080)
.onSuccess {
case _ => logger.info("Server running on :8080")
}
Unfortunately when I make 6 requests from chrome they are all fulfileld
instantly
Array.from(new Array(6), (x,i) => i).forEach(x => fetch("
http://localhost:8080/1/k";))
I even tried this minimal setup:
val awesomeFlow =
Flow[HttpRequest]
.throttle(1, 1 second, 1, ThrottleMode.Enforcing)
.via(routeFlow)
Clearly I am missing something fundamental about reactive streams and would
really appreciate any pointers you could give. Even just general guidance
on how to test or think about this.
On Sunday, February 19, 2017 at 12:58:24 PM UTC+1, Søren Bramer Schmidt
wrote:
>
> Hi
>
> I am using Akka Http in a multi tenant setup. I am trying to implement
> per-project rate limiting. The projectId is in the url, which should
> simplify the task. I have very little experience with reactive streams, so
> my conceptual understanding might be wrong.
>
> My thinking is to take an incoming flow of requests, use groupBy to
> dynamically create sub streams for each projectId and then apply throttle.
>
> Where I am stuck is that it seems like IncommingConnection has a stream of
> requests, so actually I am dealing with a stream of streams. Is this
> uderstanding correct?
>
> I am trying to use flatMapMerge to combine the request streams, but this
> doesn't compile:
>
> Http().bind(interface, port)
> .flatMapMerge(ServerSettings(system).maxConnections, conn => conn.flow) //
> this doesn't compile
> .groupBy(1000, x => extractProjectId(x))
> .buffer(1000, OverflowStrategy.dropNew)
> .throttle(100, FiniteDuration(1, TimeUnit.SECONDS), 100,
> akka.stream.ThrottleMode.Shaping)
> .mergeSubstreams
>
>
>
--
>>>>>>>>>> Read the docs: http://akka.io/docs/
>>>>>>>>>> Check the FAQ:
>>>>>>>>>> http://doc.akka.io/docs/akka/current/additional/faq.html
>>>>>>>>>> Search the archives: https://groups.google.com/group/akka-user
---
You received this message because you are subscribed to the Google Groups "Akka
User List" group.
To unsubscribe from this group and stop receiving emails from it, send an email
to [email protected].
To post to this group, send email to [email protected].
Visit this group at https://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.
