akka / akka-http Goto Github PK
View Code? Open in Web Editor NEWThe Streaming-first HTTP server/module of Akka
Home Page: https://doc.akka.io/docs/akka-http
License: Other
The Streaming-first HTTP server/module of Akka
Home Page: https://doc.akka.io/docs/akka-http
License: Other
Extending the Directives
trait is a killer for binary compatibility.
What if, when going stable in akka/akka-http we say this is not supported and mark the trait as deprecated, instead people should always import Directives._
or better, just the ones they need.
It is very "cool" to just extend Directives, however it comes with many problems as you know...
Once Scala 2.12 is out, we can support it again, because then adding directives to traits IS binary compatible.
Yay or Nay @akka/akka-http-team ?
Issue by alias1
Monday Jul 11, 2016 at 09:53 GMT
Originally opened as akka/akka#20934
I'm trying to put together some API abstractions using the request level client API, and running into this error message:
Cannot determine request scheme and target endpoint as HttpMethod(GET) request to /self/calendar doesn't have an absolute URI
I would have expected this should work since I am setting the host header. Cut down example:
# Initial Setup/Abstractions
val http = Http(system)
val hostHeader = headers.Host("api.meetup.com")
val request = HttpRequest().withDefaultHeaders(hostHeader)
val get = request.withMethod(HttpMethods.GET)
# Individual Call
val uri = Uri().withPath(Path("/self/calendar"))
val r = get.withUri(uri)
http.singleRequest(r)
Printing the request:
HttpRequest(HttpMethod(GET),/self/calendar,List(Host: api.meetup.com),HttpEntity.Strict(none/none,ByteString()),HttpProtocol(HTTP/1.1))
I have also tried using the following, to no avail:
val request = HttpRequest().withHeaders(hostHeader)
val request = HttpRequest().withEffectiveUri(true, hostHeader)
.toRelative
on the uri
chainget.withUri("/self/calendar")
Though if I use val request = get.withUri(uri).withEffectiveUri(true, hostHeader)
then the request works.
Am I missing something? Or is there a better way to build up these abstractions? I would have expected that so long as I had the host/effective URL set somewhere in the chain it should work fine.
We have a fixme for it.
The problem is that it would be binary incompatible, right?
// FIXME: Why don't we extend akka.http.javadsl.model.ws.Message here?
Issue by Joe-Edwards
Tuesday Jun 28, 2016 at 17:14 GMT
Originally opened as akka/akka#20850
A newcomer to akka (and routing) might do something like:
extractRequest { request =>
val strictEntity = Await.result(request.entity.toStrict(5.seconds), 10.seconds)
// do something with the strict entity here
}
This appears to completely deadlock when the request isn't already strict, and eventually just times out the request. (Note that this is much worse than blocking on other things, for which you might get some thread starvation, it just hangs completely)
While the code should clearly be changed to avoid the Await, it's surprisingly difficult to debug what's actually happening if you do something like this. I had some test code which used awaits like this and it took a while to figure out why we were getting random requests hanging.
IMO the outcome should be some subset of:
I can't work out exactly what's going on, I guess the thread handling the route must be locking on something the TCP/HTTP layer needs to stream the entity.
There are not breaking changes in Jackson 2.8.x compared to 2.7.x, only new features, it is even still compatible with Java < 8
That said I see no harm on just bumping to 2.8.x, I have upgraded a POJO project to it -legacy code- and I haven't found any issue so far which with annotated POJOs is even more complicated to migrate.
Issue by kelebra
Sunday Jun 19, 2016 at 04:05 GMT
Originally opened as akka/akka#20802
Hi, guys,
I've been playing around with aka-http and found out that when I have a route like this:
get {
path("something" / Rest) { info =>
println(info)
complete(s"You entered: $info")
}
}
and send get to /something/long%20line
the info
variable is actually equal to long%20line
. Maybe it would be nice to add at least implicit method to replace %20
to actual spaces like info.withSpaces
or it would be make this king of behavior default?
We want to use this to power examples for all directives.
It will make it easier to maintain the directive docs and would also address akka/akka#18660
The idea is to keep examples somewhere where they are compiled as usual. Then we'd put a reference to an example directly into the scaladoc and then have a preprocessor (on sbt or scaladoc/javadoc level) that includes the snippet at that point into the scaladocs.
Example would be:
class ThingDirectives {
/**
* Does things.
*
* @example $root/akka-http-docs/blabla/ExampleDirectivesSpec.scala#example
*/
def example(t: String): Directive0 = ???
}
It should:
<code>
block basically Issue by francisdb
Wednesday Aug 24, 2016 at 08:26 GMT
Originally opened as akka/akka#21272
We got this on akka http 2.4.6
Not sure how a request with invalid uri ends up at the server but I suppose this should not be a 500 error but rather a 400?
Internal server error, sending 500 response
java.lang.IllegalArgumentException: `uri` must have scheme "http", "https" or no scheme
at akka.http.scaladsl.model.HttpRequest$.verifyUri(HttpMessage.scala:292)
at akka.http.scaladsl.model.HttpRequest.<init>(HttpMessage.scala:155)
at akka.http.scaladsl.model.HttpRequest$.apply(HttpMessage.scala:302)
at akka.http.impl.engine.server.HttpServerBluePrint$PrepareRequests$$anon$1.onPush(HttpServerBluePrint.scala:133)
at akka.stream.impl.fusing.GraphInterpreter.processElement$1(GraphInterpreter.scala:587)
at akka.stream.impl.fusing.GraphInterpreter.processEvent(GraphInterpreter.scala:598)
at akka.stream.impl.fusing.GraphInterpreter.execute(GraphInterpreter.scala:539)
at akka.stream.impl.fusing.GraphInterpreterShell.runBatch(ActorGraphInterpreter.scala:472)
at akka.stream.impl.fusing.GraphInterpreterShell.receive(ActorGraphInterpreter.scala:424)
at akka.stream.impl.fusing.ActorGraphInterpreter.akka$stream$impl$fusing$ActorGraphInterpreter$$processEvent(ActorGraphInterpreter.scala:604)
at akka.stream.impl.fusing.ActorGraphInterpreter$$anonfun$receive$1.applyOrElse(ActorGraphInterpreter.scala:619)
at akka.actor.Actor$class.aroundReceive(Actor.scala:482)
at akka.stream.impl.fusing.ActorGraphInterpreter.aroundReceive(ActorGraphInterpreter.scala:530)
at akka.actor.ActorCell.receiveMessage(ActorCell.scala:526)
at akka.actor.ActorCell.invoke(ActorCell.scala:495)
at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:257)
at akka.dispatch.Mailbox.run(Mailbox.scala:224)
at akka.dispatch.Mailbox.exec(Mailbox.scala:234)
at scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)
at scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)
at scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)
at scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)
@2Beaucoup said:
Maintaining the directive docs in their current form will require a lot of effort. Maybe we can enrich the nicely grouped ScalaDoc with code samples, link to it and call it a day?
I think that's a great idea. It would require tooling and we would need it before the first stable release probably...
Would someone step up to impl this?
Issue by levkhomich
Saturday Jul 16, 2016 at 06:40 GMT
Originally opened as akka/akka#20969
The problem is that decodeRequest
and decodeRequestWith
directives map original request entity to HttpEntity.Chunked
which, in turn, fails HttpRequest.copy
's requirement
require(
protocol == HttpProtocols.`HTTP/1.1` || !entity.isInstanceOf[HttpEntity.Chunked],
"HTTP/1.0 responses must not have a chunked entity")
and produces following stack trace:
java.lang.IllegalArgumentException: requirement failed: HTTP/1.0 requests must not have a chunked entity
at scala.Predef$.require(Predef.scala:224)
at akka.http.scaladsl.model.HttpRequest.<init>(HttpMessage.scala:207)
at akka.http.scaladsl.model.HttpRequest.copy(HttpMessage.scala:271)
at akka.http.scaladsl.coding.DataMapper$$anonfun$1.apply(DataMapper.scala:32)
at akka.http.scaladsl.coding.DataMapper$$anonfun$1.apply(DataMapper.scala:28)
at akka.http.scaladsl.coding.DataMapper$$anon$3.transformDataBytes(DataMapper.scala:41)
at akka.http.scaladsl.coding.Decoder$class.decodeData(Decoder.scala:25)
at akka.http.scaladsl.coding.StreamDecoder$$anon$1.decodeData(Decoder.scala:44)
at akka.http.scaladsl.coding.Decoder$class.decode(Decoder.scala:22)
at akka.http.scaladsl.coding.StreamDecoder$$anon$1.decode(Decoder.scala:44)
at akka.http.scaladsl.server.directives.CodingDirectives$$anonfun$applyDecoder$1$1$$anonfun$apply$2.apply(CodingDirectives.scala:86)
at akka.http.scaladsl.server.directives.CodingDirectives$$anonfun$applyDecoder$1$1$$anonfun$apply$2.apply(CodingDirectives.scala:85)
Steps to reproduce. Test case:
val echoRequestContent: Route = { ctx ⇒ ctx.complete(ctx.request.entity.dataBytes.utf8String) }
lazy val helloGzipped = compress("Hello", Gzip)
def compress(input: String, encoder: Encoder): ByteString = {
val compressor = encoder.newCompressor
compressor.compressAndFlush(ByteString(input)) ++ compressor.finish()
}
"decode HTTP/1.0 request with 'gzip' encoding" in {
HttpRequest(POST, Uri("/"), entity = HttpEntity(helloGzipped), protocol = `HTTP/1.0`) ~> `Content-Encoding`(gzip) ~> {
decodeRequestWith(Gzip) { echoRequestContent }
} ~> check { responseAs[String] shouldEqual "Hello" }
}
Issue by jrudolph
Thursday Aug 21, 2014 at 12:03 GMT
Originally opened as akka/akka#15714
E.g. in HttpManager
we react differently on network errors caused by Http.Connect
and Http.Bind
commands:
Http.Connect
is DEBUG-logged and then it's reported to the commander containing the remote address in questionHttp.Bind
is WARNING-logged and the singleton BindFailedException
is reported to the commanderIn both cases, neither the actual exception nor the message are reported back to the commander. This means it isn't possible for the commander to actually take responsibility for handling or reporting the error and thus it's easy to get into the old Java double logging anti-pattern.
I would propose a scheme where errors/exceptions are generally reported to the commander including the underlying message and if they are logged then only with DEBUG-level, to let the commander decide what to do with them.
It's probably not too useful to also send the stacktraces/original exceptions around, but there should be at least some way to log the stacktraces somewhere before they are discarded for debugging purposes.
/cc @sirthias
Issue by mpilquist
Thursday Aug 04, 2016 at 13:08 GMT
Originally opened as akka/akka#21123
The default rejection handlers do not discard request entity bytes before completing requests. For example, if a request is rejected with an AuthorizationFailedRejection
, the default rejection handler will complete the request with a 403
, ignoring the request entity.
I think this will lead to problems but I haven't tried reproducing. I noticed this when updating a code base to properly discard request/response entities on server/client side. Might be worth a mention in the docs on writing custom rejection handlers.
Issue by jrudolph
Thursday Aug 21, 2014 at 12:03 GMT
Originally opened as akka/akka#15714
E.g. in HttpManager
we react differently on network errors caused by Http.Connect
and Http.Bind
commands:
Http.Connect
is DEBUG-logged and then it's reported to the commander containing the remote address in questionHttp.Bind
is WARNING-logged and the singleton BindFailedException
is reported to the commanderIn both cases, neither the actual exception nor the message are reported back to the commander. This means it isn't possible for the commander to actually take responsibility for handling or reporting the error and thus it's easy to get into the old Java double logging anti-pattern.
I would propose a scheme where errors/exceptions are generally reported to the commander including the underlying message and if they are logged then only with DEBUG-level, to let the commander decide what to do with them.
It's probably not too useful to also send the stacktraces/original exceptions around, but there should be at least some way to log the stacktraces somewhere before they are discarded for debugging purposes.
/cc @sirthias
Issue by jrudolph
Monday Sep 05, 2016 at 12:45 GMT
Originally opened as akka/akka#21370
PoolFlow, PoolConductor, and PoolSlot do a sophisticated dance to send state information around. This could probably be much simplified by having just two GraphStages:
Flow[HttpRequest, HttpResponse]
and creates the slots internally, or could have the shape of a flow plus additional holes for all of the slots to be filled from the outside.The same issue around "but I want it strict" surfaces in WebSocket messages as it does with HttpEntities - for entities people can easily toStrict them - for websocket messages where toStricting usually is more safe (smaller msgs... generally speaking), we don't have this helper on the type directly.
I think it would make a nice addition, opinions? Did I miss something?
Issue by mpilquist
Tuesday Aug 02, 2016 at 14:54 GMT
Originally opened as akka/akka#21098
Consider the following program, which incorrectly fails to process response entities:
import $ivy.`com.typesafe.akka::akka-http-experimental:2.4.8`
import scala.concurrent.Future
import akka.actor.ActorSystem
import akka.stream.ActorMaterializer
import akka.stream.scaladsl._
import akka.http.scaladsl._
import akka.http.scaladsl.client.RequestBuilding._
import akka.http.scaladsl.server._
implicit val system = ActorSystem()
implicit val materializer = ActorMaterializer()
implicit val ec = system.dispatcher
// Let's create an HTTP server that responds to all requests with a very large response entity
val serverFlow = Route.handlerFlow(Directives.complete("." * 5000000))
Http().bindAndHandle(serverFlow, "0.0.0.0", 5555).flatMap { serverBinding =>
// Now that server has started, let's make a bunch of requests in succession using the superPool API
val pool = Http().superPool[Unit]()
Future.traverse(0 to 20) { i =>
val response = Source.single(Get("http://localhost:5555") -> ()).
via(pool).
runWith(Sink.head)
response.map { case (tryResponse, _) =>
// We incorrectly fail to drain the response entity here
println(s"Got response $i")
tryResponse
}
}
}
Running this program results in:
Got response 2
Got response 0
Got response 4
Got response 1
4 requests run and then, due to the client failing to process response entities, the program hangs forever. This is a bug in this code -- the response entities must be processed. However, that's not what this ticket is about.
The super pool API could be made much safer by supporting some form of (optional?) timeouts on requests. For example, if the fifth request hasn't been assigned a host connector within X seconds, fail the response with a timeout exception indicating the pool was too busy.
In presence of such a feature, the above program would complete - 4 requests would execute like they did in the above output and then 16 requests would fail with timeout exceptions, indicating the connector pool was too busy. This is a much safer default behavior.
More generally, I'd like a programmatic way to recover from such scenarios -- e.g., if many requests are timing out, throw away the super pool and instantiate a new one. I could build this feature on top of the super pool and Future
API but I suspect many folks will want similar behavior.
For some background, see this Twitter stream: https://twitter.com/wildsebastian/status/758937740410888192
Currently, akka.http.javadsl.testkit.TestRouteResult.assertHeaderExists(expectedHeader)
, when the header has a different value, will simply say "Header $expected was missing."
. That's not very useful if you're trying to diagnose what's wrong with your actual returned header
Much more useful would be "Expected $expected, but instead got $actualHeadersWithThatName"
The current formField
and friends require buffering of the full request, since we don't know when each field is going to come in from the live TCP stream.
Futures won't work very well for this, since not consuming them may cause the source stream to stop in case "bigfile" comes before "smallvalue".
Suggestion: chop up the TCP stream into FormField
elements (or maybe an HMap
-like thingy to make things more type and keep the extractors), where each FormField
contains, at lowest level, a name:String
and contents:Source[ByteString,NotUsed]
. Then require that those sub streams are consumed in order, unless explicitly buffered.
Related: akka/akka#18426
Issue by derjust
Friday Aug 26, 2016 at 18:36 GMT
Originally opened as akka/akka#21304
This is a very simplified example to show the issue. The general process is
Source[HttpRequest]
http.newHostConnectionPool()
to turn it into Flow[Future[HttpResponse]]
Flow.mapAsyncUnordered
The problem appears now as soon as the IdleTimeout
throws its
java.util.concurrent.TimeoutException: No elements passed in the last X.
at akka.stream.impl.Timers$IdleTimeoutBidi$$anon$7.onTimer(Timers.scala:160) ~[akka-stream_2.11-2.4.9.jar:na]
at akka.stream.stage.TimerGraphStageLogic.akka$stream$stage$TimerGraphStageLogic$$onInternalTimer(GraphStage.scala:1152) ~[akka-stream_2.11-2.4.9.jar:na]
at akka.stream.stage.TimerGraphStageLogic$$anonfun$akka$stream$stage$TimerGraphStageLogic$$getTimerAsyncCallback$1.apply(GraphStage.scala:1141) ~[akka-stream_2.11-2.4.9.jar:na]
at akka.stream.stage.TimerGraphStageLogic$$anonfun$akka$stream$stage$TimerGraphStageLogic$$getTimerAsyncCallback$1.apply(GraphStage.scala:1141) ~[akka-stream_2.11-2.4.9.jar:na]
at akka.stream.impl.fusing.GraphInterpreter.runAsyncInput(GraphInterpreter.scala:699) ~[akka-stream_2.11-2.4.9.jar:na]
OR the stream is aborted before all elements of the http()/mapAsync
combo are consumed.
I understand why it is happening in my real setup (the chunked response isn't consumed quickly enough)
The issue is, that this exception seems to keep the overall Flow
- depicted above - in an open state. At least it is not GCed and thus my memory fills up till OOM.
The very same also happens if something behind the http.pool
Flow (i.e. httpResponse2Chunk
) issues a complete/fail stage
To reproduce, use the code below, consume the service per
curl http://localhost:8080/ --limit-rate 1b > /dev/null
Afterwards use the heap dump (jmap
, visualvm
etc.) to search for instances of JustSomeCaseToMakeSearchingInHeapDumpEasier
Those should not exist as the whole stream should be gone (a LastChunk
was sent to the client)
I want to understand why the resources are not removed or what to do to release the resources.
scalaVersion := "2.11.8"
libraryDependencies ++= {
val akkaV = "2.4.9"
Seq(
"com.typesafe.akka" %% "akka-http-core" % akkaV,
"com.typesafe.akka" %% "akka-http-experimental" % akkaV,
"com.typesafe.akka" %% "akka-actor" % akkaV,
"com.typesafe.akka" %% "akka-slf4j" % akkaV,
"org.slf4j" % "slf4j-api" % "1.7.21",
"ch.qos.logback" % "logback-classic" % "1.0.7",
"com.typesafe.akka" %% "akka-http-testkit" % akkaV % "test"
)
}
akka {
http {
host-connection-pool {
#This is the mapAsyncUnordered parallelism
# with max-connections = 1 the issue does not appear at all
max-connections = 8
idle-timeout = 45 s
client = {
idle-timeout = 45 s
}
}
}
}
object Boot extends App {
implicit val actorSystem = ActorSystem()
implicit val ec = actorSystem.dispatcher
implicit val materializer = ActorMaterializer()
val routes = new TestService().downloadService
val serverSource = Http().bindAndHandle(routes, "0.0.0.0", port = 8080)
println("Ready. Use \n\tcurl http://localhost:8080/ --limit-rate 5k > /dev/null\nto cause the issue")
}
final class SomeDownstreamBusiness(data: List[JustSomeCaseToMakeSearchingInHeapDumpEasier]) extends SimpleLinearGraphStage[ChunkStreamPart] {
override def createLogic(inheritedAttributes: Attributes): GraphStageLogic = new GraphStageLogic(shape) {
setHandler(in, new InHandler {
override def onPush(): Unit = push(out, grab(in))
})
setHandler(out, new OutHandler {
override def onPull(): Unit = pull(in)
})
}
}
case class JustSomeCaseToMakeSearchingInHeapDumpEasier(file: String)
class TestService(implicit system: ActorSystem, fm: Materializer) {
def chunkStreamGraphCreator() = {
Source.fromGraph(GraphDSL.create() { implicit builder =>
import GraphDSL.Implicits._
implicit val ec = system.dispatcher
val baseUrl = Uri("http://devimages.apple.com/iphone/samples/bipbop/gear4/")
val data = (for (i <- 0 to 170) yield JustSomeCaseToMakeSearchingInHeapDumpEasier(s"fileSequence$i.ts"))
val in = Source.fromIterator[JustSomeCaseToMakeSearchingInHeapDumpEasier](() => data.iterator)
val createRequest = Flow[JustSomeCaseToMakeSearchingInHeapDumpEasier].map(s => {
val totalUri = Uri(s.file).resolvedAgainst(baseUrl)
println(totalUri)
(HttpRequest(uri = totalUri), s)
})
val httpClient = Http().newHostConnectionPool[JustSomeCaseToMakeSearchingInHeapDumpEasier](baseUrl.authority.host.address, baseUrl.effectivePort)
val httpResponse2Chunk = Flow[(Try[HttpResponse], JustSomeCaseToMakeSearchingInHeapDumpEasier)].mapAsyncUnordered(8)({case (response, downloadedFile) =>
response match {
// No proper error handling here - it's fine to send a 404 error page down, too
case Success(httpResponse) => httpResponse.entity.toStrict(1 minute).map(data => {
ChunkStreamPart(data.data)
})
// Upstream error should let the client finish properly
case _ => Future.successful(LastChunk())
}})
val out = in ~> createRequest ~> httpClient ~> httpResponse2Chunk ~> Flow[ChunkStreamPart].takeWithin(5 seconds)
SourceShape(out.outlet)
})
}
def downloadService = get { complete {
HttpResponse(entity = HttpEntity.Chunked(ContentType(MediaTypes.`application/octet-stream`), chunkStreamGraphCreator()))
}}
}
The transfer of data will stop after 5 seconds (and should complete the Flow) - but it doesn't.
Also after the idle timer went off, you still can see having alive parts of the flow via visualvm
:
GraphInterpreterShell
than beforeJustSomeCaseToMakeSearchingInHeapDumpEasier
instances that should not be thereAccording to my visualvm
, the GC root is:
this - value: akkaHttpMemory.JustSomeCaseToMakeSearchingInHeapDumpEasier #1
<- [31] - class: java.lang.Object[], value: akkaHttpMemory.JustSomeCaseToMakeSearchingInHeapDumpEasier #1
<- display0 - class: scala.collection.immutable.VectorIterator, value: java.lang.Object[] #3054
<- currentIterator - class: akka.stream.impl.fusing.StatefulMapConcat$$anon$19, value: scala.collection.immutable.VectorIterator #5
<- [9] - class: akka.stream.stage.GraphStageLogic[], value: akka.stream.impl.fusing.StatefulMapConcat$$anon$19 #12
<- logics - class: akka.stream.impl.fusing.GraphInterpreter, value: akka.stream.stage.GraphStageLogic[] #10
<- _interpreter - class: akka.stream.impl.fusing.SubSink$$anon$3, value: akka.stream.impl.fusing.GraphInterpreter #11
<- $outer - class: akka.stream.stage.GraphStageLogic$$anon$2, value: akka.stream.impl.fusing.SubSink$$anon$3 #1
<- value - class: java.util.concurrent.atomic.AtomicReference, value: akka.stream.stage.GraphStageLogic$$anon$2 #25
<- akka$stream$impl$fusing$SubSink$$status - class: akka.stream.impl.fusing.SubSink, value: java.util.concurrent.atomic.AtomicReference #40
<- _sink - class: akka.stream.stage.GraphStageLogic$SubSinkInlet, value: akka.stream.impl.fusing.SubSink #1
<- sinkIn$1 - class: akka.http.impl.engine.rendering.HttpResponseRendererFactory$HttpResponseRenderer$$anon$1$$anon$5, value: akka.stream.stage.GraphStageLogic$SubSinkInlet #1
<- outHandler - class: akka.stream.impl.fusing.GraphInterpreter$Connection, value: akka.http.impl.engine.rendering.HttpResponseRendererFactory$HttpResponseRenderer$$anon$1$$anon$5 #1
<- [6] - class: akka.stream.impl.fusing.GraphInterpreter$Connection[], value: akka.stream.impl.fusing.GraphInterpreter$Connection #271
<- connections - class: akka.stream.impl.fusing.GraphInterpreter, value: akka.stream.impl.fusing.GraphInterpreter$Connection[] #245
<- _interpreter - class: akka.stream.impl.io.TcpConnectionStage$TcpStreamLogic, value: akka.stream.impl.fusing.GraphInterpreter #12
<- $outer - class: akka.stream.impl.io.TcpConnectionStage$TcpStreamLogic$$anonfun$preStart$2, value: akka.stream.impl.io.TcpConnectionStage$TcpStreamLogic #9
<- behaviour - class: akka.stream.stage.GraphStageLogic$StageActor, value: akka.stream.impl.io.TcpConnectionStage$TcpStreamLogic$$anonfun$preStart$2 #1
<- $outer - class: akka.stream.stage.GraphStageLogic$StageActor$$anonfun$2, value: akka.stream.stage.GraphStageLogic$StageActor #9
<- f - class: akka.actor.FunctionRef, value: akka.stream.stage.GraphStageLogic$StageActor$$anonfun$2 #9
<- key - class: scala.collection.immutable.HashSet$HashSet1, value: akka.actor.FunctionRef #9
<- akka$actor$dungeon$DeathWatch$$terminatedQueued - class: akka.actor.ActorCell, value: scala.collection.immutable.HashSet$HashSet1 #195
<- actorCell - class: akka.actor.LocalActorRef, value: akka.actor.ActorCell #54
<- attachment - class: sun.nio.ch.SelectionKeyImpl, value: akka.actor.LocalActorRef #17
<- ski - class: sun.nio.ch.KQueueSelectorImpl$MapEntry, value: sun.nio.ch.SelectionKeyImpl #9
<- value - class: java.util.HashMap$Node, value: sun.nio.ch.KQueueSelectorImpl$MapEntry #9
<- [4] - class: java.util.HashMap$Node[], value: java.util.HashMap$Node #3464
<- table - class: java.util.HashMap, value: java.util.HashMap$Node[] #571
<- fdMap (Java frame) - class: sun.nio.ch.KQueueSelectorImpl, value: java.util.HashMap #624
This ticket serves as progress tracker of all the initial tasks:
Issue by jrudolph
Monday Sep 05, 2016 at 12:39 GMT
Originally opened as akka/akka#21368
HTTP pipelining doesn't work any more in the low-level client-side implementation.
We do not seem to have any tests that exercise that pipelining actually works over the wire.
This test (to be added in LowLevelOutgoingConnectionSpec) fails on the last expectWireData
:
"supports two pipelined requests" in new TestSetup {
requestsSub.sendNext(HttpRequest(uri = "/a"))
expectWireData(
"""GET /a HTTP/1.1
|Host: example.com
|User-Agent: akka-http/test
|
|""")
responsesSub.request(10)
requestsSub.sendNext(HttpRequest(uri = "/b"))
expectWireData(
"""GET /b HTTP/1.1
|Host: example.com
|User-Agent: akka-http/test
|
|""")
}
since
Bisecting: 0 revisions left to test after this (roughly 0 steps)
[5afab017b08c1ed97099ade57aef83e6b9490383] add fusing
(which is not surprising since fusing changed all kinds of buffering behavior).
In the commit where it worked the last time, I tried a similar test:
responsesSub.request(10)
def sendOne(id: Int): Unit = {
println(id)
requestsSub.sendNext(HttpRequest(uri = s"/$id"))
expectWireData(
s"""GET /$id HTTP/1.1
|Host: example.com
|User-Agent: akka-http/test
|
|""")
}
(1 to 1000).foreach(sendOne)
which will fail only on the 49th request when backpressure kicks in (but where and why?).
In ConnectionPoolSpec, we should have these additional tests:
"pipeline several requests on one slot if all running requests are idempotent" in pending
"do not pipeline a further request on one slot if one request is non-idempotent" in pending
"retry idempotent request pipelined after a request with a response that closed the connection" in pending
"retry idempotent request pipelined on a connection that was aborted" in pending
"do not retry non-idempotent request pipelined after a request with a response that closed the connection" in pending
"do not retry non-idempotent request pipelined on a connection that was aborted" in pending
When the issue has been fixed, changes from #21316 to PoolSlot need to be revisited to see if everything still works (should also be caught by the new tests).
Issue by stanch
Monday Apr 25, 2016 at 23:10 GMT
Originally opened as akka/akka#20393
I think Test Kit is great and it’s really a shame that we have to dispense with the nice syntax it offers when testing external web services or doing system tests. In those cases there is no Route
, but an HTTP client can be used to provide a function HttpRequest => Future[HttpResponse]
. The problem is that RouteTestResult
, which is needed for the check
DSL to work, can only be initialized within the akka.http.scaladsl.testkit
package. Other than that, it seems straight-forward to initialize RouteTestResult
from an HttpResponse
. Here is what it takes:
implicit def injectIntoPipeline(
implicit timeout: RouteTestTimeout,
defaultHostInfo: DefaultHostInfo,
executionContext: ExecutionContext
): TildeArrow[HttpRequest, Future[HttpResponse]] { type Out = RouteTestResult } =
new TildeArrow[HttpRequest, Future[HttpResponse]] {
type Out = RouteTestResult
def apply(request: HttpRequest, pipeline: HttpRequest => Future[HttpResponse]): Out = {
val routeTestResult = new RouteTestResult(timeout.duration)
pipeline(request) foreach { response =>
routeTestResult.handleResult(RouteResult.Complete(response))
}
routeTestResult
}
}
With this added to akka.http.scaladsl.testkit.RouteTest
, the following can be done:
// define your HTTP pipeline (I think this is what it was called in Spray)
def pipeline(
implicit system: ActorSystem, materializer: Materializer, ec: ExecutionContext
): HttpRequest => Future[HttpResponse] = { request =>
Http().singleRequest {
request
.withUri(???) // add your custom host/port
.withHeaders(???) // add your custom auth
}
}
// test away
val request = Json.obj("foo" -> "bar")
Post("/v1.1/api", request) ~> pipeline ~> check {
// your custom checks
}
Issue by ktoso
Monday Aug 01, 2016 at 10:17 GMT
Originally opened as akka/akka#21085
Given such route
val routes: Route =
path("setting") {
val value = system.settings.config.getString("akka.http.server.transparent-head-requests")
complete(s"akka.http.server.transparent-head-requests = $value")
} ~
path("test") {
head { complete(HttpResponse()) }
}
people are very confused since the HEAD route never gets hit.
This is because of akka.http.server.transparent-head-requests
which mutated it into a GET request and would strip the response entity from the response.
We should make this still work - if someone wants to handle a HEAD direcly they should be able to.
$ curl 127.0.0.1:8080/setting
akka.http.server.transparent-head-requests = on
ktoso @ 月~/code/akka [master*]
$ http -v HEAD 127.0.0.1:8080/test
HEAD /test HTTP/1.1
Accept: */*
Accept-Encoding: gzip, deflate
Connection: keep-alive
Host: 127.0.0.1:8080
User-Agent: HTTPie/0.9.3
HTTP/1.1 405 Method Not Allowed
Allow: HEAD
Content-Length: 48
Content-Type: text/plain; charset=UTF-8
Date: Mon, 01 Aug 2016 10:18:17 GMT
Server: akka-http/2.4-SNAPSHOT
Issue by zifeo
Friday Jun 24, 2016 at 08:18 GMT
Originally opened as akka/akka#20825
RequestBuilding
overloadsRouteTestTimeout
Reference #20798.
zifeo included the following code: https://github.com/akka/akka/pull/20825/commits
We want to move documentation from:
To this repo. Both text and tests of course.
The docs should be generated using https://github.com/lightbend/paradox which will allow us to "aggregate" them together with other akka docs into an "aggregated view" if someone wants that.
Otherwise the docs will be separate so this will allow more focused browsing and searching.
The theme for now can be the plain Akka one, though we'll likely want to differentiate a little bit, maybe just by adding HTTP in the header or sth.
I will attempt to make the first step of this move, and then would like to ask for help migrating all the docs to paradox.
Issue by jrudolph
Wednesday Sep 07, 2016 at 17:16 GMT
Originally opened as akka/akka#21393
Even if the content-type would allow it by having compressible=true
set.
Follow up from #20793.
Issue by Joe-Edwards
Tuesday Jul 05, 2016 at 09:45 GMT
Originally opened as akka/akka#20881
From #20850
A Directive1[HttpEntity.Strict]
which calls toStrict on the HTTP Entity safely with a specified timeout (to avoid the deadlock in the other ticket). N.B. should also replace the entity in the request context.
Issue by leachbj
Tuesday Aug 02, 2016 at 23:29 GMT
Originally opened as akka/akka#21102
Recently I had to test a directive that used the withRequestTimeoutResponse directive but was unable to make it work with a regular ScalatestRouteTest. I found that IntegrationRoutingSpec in the akka-http-tests module however worked great for this scenario. Unfortunately it is private[akka] so I had to copy the code in order to re-use it.
It would be great to have official integration test support like IntegrationRoutingSpec available in akka-http-testkit.
From @agboom on September 2, 2016 12:32
When an Http Request results in an internal server error, the entire request is logged, including body, headers and all. This means that credentials/tokens for authentication (or other sensitive data) is logged on error level.
ERROR [default-akka.actor.default-dispatcher-5746] (akka.actor.ActorSystemImpl): Error during processing of request HttpRequest(
HttpMethod(POST),
http://localhost:9000,
List(Host: localhost, User-Agent: <redacted>, Cache-Control: no-cache, Authorization: <redacted>, Accept: */*, DNT: 1, Accept-Encoding: gzip, deflate, Accept-Language: en-US, en;q=0.8, nl;q=0.6, X-Forwarded-For: <redacted>, Timeout-Access: <function1>),
HttpEntity.Strict(... (4847 bytes total)),
HttpProtocol(HTTP/1.1)
)
This can't be turned off by configuring a different log level. One may be able to work around it using an ExceptionHandler (haven't tested it). A relevant question might be whether it is safe to even log the request by default or if it would be safer to make this an "opt-in" feature, for example by using an ExceptionHandler.
Copied from original issue: akka/akka#21354
Issue by @2m
Friday Sep 02, 2016 at 07:34 GMT
Originally opened as akka/akka#21348
Sep 2, 2016 7:18:01 AM
https://jenkins.akka.io:8498/job/akka-multi-node-nightly2/461/
[akka.http.impl.engine.server.HttpServerBluePrint$ControllerStage@69243a2b]: requirement failed: Cannot push port (requestPrepOut) twice
java.lang.IllegalArgumentException: requirement failed: Cannot push port (requestPrepOut) twice
at scala.Predef$.require(Predef.scala:224)
at akka.stream.stage.GraphStageLogic.push(GraphStage.scala:459)
at akka.http.impl.engine.server.HttpServerBluePrint$ControllerStage$$anon$12$$anon$13.onPush(HttpServerBluePrint.scala:400)
at akka.stream.impl.fusing.GraphInterpreter.processPush(GraphInterpreter.scala:747)
at akka.stream.impl.fusing.GraphInterpreter.execute(GraphInterpreter.scala:649)
at akka.stream.impl.fusing.GraphInterpreterShell.runBatch(ActorGraphInterpreter.scala:471)
at akka.stream.impl.fusing.GraphInterpreterShell.receive(ActorGraphInterpreter.scala:410)
at akka.stream.impl.fusing.ActorGraphInterpreter.akka$stream$impl$fusing$ActorGraphInterpreter$$processEvent(ActorGraphInterpreter.scala:603)
at akka.stream.impl.fusing.ActorGraphInterpreter$$anonfun$receive$1.applyOrElse(ActorGraphInterpreter.scala:618)
at akka.actor.Actor$class.aroundReceive(Actor.scala:484)
at akka.stream.impl.fusing.ActorGraphInterpreter.aroundReceive(ActorGraphInterpreter.scala:529)
[info] - should not cause internal graph failures when consuming a `100 Continue` entity triggers a failure *** FAILED *** (179 milliseconds)
[info] HttpServerBluePrint.ControllerStage: requirement failed: Cannot pull closed port (requestParsingIn),
[info] inside HttpRequest(HttpMethod(POST),http://example.com/,List(Host: example.com, Expect: 100-continue, Timeout-Access: <function1>),HttpEntity.Chunked(application/octet-stream),HttpProtocol(HTTP/1.1)) (HttpServerBug2108Spec.scala:70)
[info] org.scalatest.exceptions.TestFailedException:
[info] at org.scalatest.Assertions$class.newAssertionFailedException(Assertions.scala:528)
[info] at akka.testkit.AkkaSpec.newAssertionFailedException(AkkaSpec.scala:57)
[info] at org.scalatest.Assertions$class.fail(Assertions.scala:1089)
[info] at akka.testkit.AkkaSpec.fail(AkkaSpec.scala:57)
[info] at akka.http.impl.engine.server.HttpServerBug2108Spec$$anonfun$1$$anonfun$apply$mcV$sp$1$$anonfun$apply$1$$anon$1$$anonfun$2.applyOrElse(HttpServerBug2108Spec.scala:70)
[info] at akka.http.impl.engine.server.HttpServerBug2108Spec$$anonfun$1$$anonfun$apply$mcV$sp$1$$anonfun$apply$1$$anon$1$$anonfun$2.applyOrElse(HttpServerBug2108Spec.scala:38)
[info] at scala.runtime.AbstractPartialFunction.apply(AbstractPartialFunction.scala:36)
[info] at org.scalatest.Inside$class.inside(Inside.scala:128)
[info] at akka.http.impl.engine.server.HttpServerBug2108Spec.inside(HttpServerBug2108Spec.scala:18)
[info] at akka.http.impl.engine.server.HttpServerBug2108Spec$$anonfun$1$$anonfun$apply$mcV$sp$1$$anonfun$apply$1$$anon$1.<init>(HttpServerBug2108Spec.scala:38)
[info] at akka.http.impl.engine.server.HttpServerBug2108Spec$$anonfun$1$$anonfun$apply$mcV$sp$1$$anonfun$apply$1.apply(HttpServerBug2108Spec.scala:28)
[info] at akka.http.impl.engine.server.HttpServerBug2108Spec$$anonfun$1$$anonfun$apply$mcV$sp$1$$anonfun$apply$1.apply(HttpServerBug2108Spec.scala:28)
[info] at akka.stream.testkit.Utils$.assertAllStagesStopped(Utils.scala:25)
[info] at akka.http.impl.engine.server.HttpServerBug2108Spec$$anonfun$1$$anonfun$apply$mcV$sp$1.apply(HttpServerBug2108Spec.scala:28)
[info] at akka.http.impl.engine.server.HttpServerBug2108Spec$$anonfun$1$$anonfun$apply$mcV$sp$1.apply(HttpServerBug2108Spec.scala:28)
[info] at org.scalatest.OutcomeOf$class.outcomeOf(OutcomeOf.scala:85)
[info] at org.scalatest.OutcomeOf$.outcomeOf(OutcomeOf.scala:104)
[info] at org.scalatest.Transformer.apply(Transformer.scala:22)
[info] at org.scalatest.Transformer.apply(Transformer.scala:20)
[info] at org.scalatest.WordSpecLike$$anon$1.apply(WordSpecLike.scala:1078)
[info] at org.scalatest.TestSuite$class.withFixture(TestSuite.scala:196)
[info] at akka.testkit.AkkaSpec.withFixture(AkkaSpec.scala:57)
[info] at org.scalatest.WordSpecLike$class.invokeWithFixture$1(WordSpecLike.scala:1075)
[info] at org.scalatest.WordSpecLike$$anonfun$runTest$1.apply(WordSpecLike.scala:1088)
[info] at org.scalatest.WordSpecLike$$anonfun$runTest$1.apply(WordSpecLike.scala:1088)
Issue by monktastic
Tuesday Aug 02, 2016 at 03:36 GMT
Originally opened as akka/akka#21089
This flow should close the websocket when the output stream or input stream finishes, according to the doc:
val flow = Http().webSocketClientFlow(WebSocketRequest("ws://echo.websocket.org"))
But when the output finishes, the input stream is not shut down (nor is the websocket closed). E.g.:
val KeepAliveMsg = TextMessage(JsString("KeepAlive").toString) val src = Source.tick[Message](1.seconds, 1.seconds, KeepAliveMsg) val sink = myPrintFlow[Message].viaMat(KillSwitches.single)(Keep.right).to(Sink.ignore) src.via(flow).runWith(sink).shutdown()
The output is closed, but it keeps sending ticks down the websocket.
Issue by ktoso
Thursday Jul 28, 2016 at 13:51 GMT
Originally opened as akka/akka#21059
Via @johanandren
Ah, nice, just counting them. One idea, maybe for later, I guess one could do an even more imperative loop and keep the index at which each was find to avoid the find-part for when it actually is a websocket request.
As discussed in akka/akka-meta#27
A new Versioning scheme for Akka HTTP
Binary compatibility is very important to us. We’ve shown one of the best track records in the Scala ecosystem on maintaining such compatibility–including efforts that we spent between 2.3.x and 2.4.x that made these binary compatible, which would have–under the previous versioning scheme–been major releases.
Please note that Akka, since the release of 2.4.0 maintains a MAJOR.MINOR.PATCH versioning, as opposed to the previously used EPOCH.MAJOR.MINOR which used to be Scala versioning scheme inspired.
We can not use the 1.0 version to demarcate this new start as it would be very confusing since that number was already used back in July 2015 while we were working on Akka Streams & HTTP under a separate version scheme than Akka "core", which was at 2.3.x and later on continued to become 2.4.x.
Our goal with the HTTP versioning scheme is to signify that while Akka HTTP may need to bump its major version number, e.g. from 10 to 11, it would be still compatible with Akka 2.x–the stable core which will only break compatibility when a huge change is needed.
Action: After careful consideration of all of the above constraints, we propose that the new versioning scheme be different enough not be confused with Akka "core" versions. Specifically, we think using a version number much higher than 2 or 3 (which would be confusable with Akka versions) is the way to go, and suggest to start at 10.0, simultaneously announcing Akka HTTP to be stable (removing the experimental flag).
To decide: Given troublesome situation of the Directives trait being extended by user code, binary compatibility is impossible to maintain when a new directive is added into it. This will change with Scala 2.12, however with 2.11 it still is the case. We propose to keep source compatibility in minor releases, e.g. a new directive can be added but to a new trait that one mixes-in explicitly, and in a major release we can merge those new traits to the existing Directives structure. A major release may then still happen sooner than an Akka 3.0 (which is very far out), and the project can keep a healthy speed - to be determined by the team.
Please comment!
Issue by drewhk
Friday Jul 22, 2016 at 09:04 GMT
Originally opened as akka/akka#21013
https://jenkins.akka.io:8498/job/akka-multi-node-nightly/4009/consoleFull
(but also many failures in other builds)
[JVM-2] akka.actor.ActorInitializationException: akka://MultiNodeSpecSpec/user/$a: failure while creating ActorCell
[JVM-2] at akka.actor.ActorInitializationException$.apply(Actor.scala:174)
[JVM-2] at akka.actor.dungeon.Dispatch$class.initWithFailure(Dispatch.scala:90)
[JVM-2] at akka.actor.ActorCell.initWithFailure(ActorCell.scala:374)
[JVM-2] at akka.actor.RepointableActorRef.point(RepointableActorRef.scala:97)
[JVM-2] at akka.actor.ActorCell.handleSupervise(ActorCell.scala:625)
[JVM-2] at akka.actor.ActorCell.supervise(ActorCell.scala:617)
[JVM-2] at akka.actor.ActorCell.invokeAll$1(ActorCell.scala:468)
[JVM-2] at akka.actor.ActorCell.systemInvoke(ActorCell.scala:483)
[JVM-2] at akka.dispatch.Mailbox.processAllSystemMessages(Mailbox.scala:282)
[JVM-2] at akka.dispatch.Mailbox.run(Mailbox.scala:223)
[JVM-2] at akka.dispatch.Mailbox.exec(Mailbox.scala:234)
[JVM-2] at scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)
[JVM-2] at scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)
[JVM-2] at scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)
[JVM-2] at scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)
[JVM-2] Caused by: java.lang.AssertionError: assertion failed: Don't use anonymous actor classes, actor class for Actor[akka://MultiNodeSpecSpec/user/$a#1670666637] was [akka.actor.Actor]
[JVM-2] at scala.Predef$.assert(Predef.scala:170)
[JVM-2] at akka.stream.testkit.StreamTestDefaultMailbox.create(StreamTestDefaultMailbox.scala:26)
[JVM-2] at akka.dispatch.Dispatcher$$anon$1.<init>(Dispatcher.scala:89)
[JVM-2] at akka.dispatch.Dispatcher.createMailbox(Dispatcher.scala:89)
[JVM-2] at akka.actor.dungeon.Dispatch$class.init(Dispatch.scala:49)
[JVM-2] at akka.actor.ActorCell.init(ActorCell.scala:374)
[JVM-2] at akka.actor.RepointableActorRef.newCell(RepointableActorRef.scala:119)
[JVM-2] at akka.actor.RepointableActorRef.point(RepointableActorRef.scala:93)
[JVM-2] ... 11 more
[JVM-2]
...
[JVM-2] *** RUN ABORTED *** (20 minutes, 1 second)
[JVM-2] java.lang.RuntimeException: failure while attaching new conductor
[JVM-2] at akka.remote.testkit.MultiNodeSpec.attachConductor(MultiNodeSpec.scala:384)
[JVM-2] at akka.remote.testkit.MultiNodeSpec.<init>(MultiNodeSpec.scala:389)
[JVM-2] at akka.remote.testkit.MultiNodeSpec.<init>(MultiNodeSpec.scala:249)
[JVM-2] at akka.http.MultiNodeSpecSpec.<init>(AkkaHttpServerLatencyMultiNodeSpec.scala:78)
[JVM-2] at akka.http.AkkaHttpServerLatencyMultiNodeSpecMultiJvmNode2.<init>(AkkaHttpServerLatencyMultiNodeSpec.scala:76)
[JVM-2] at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)
[JVM-2] at sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:62)
[JVM-2] at sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
[JVM-2] at java.lang.reflect.Constructor.newInstance(Constructor.java:422)
[JVM-2] at java.lang.Class.newInstance(Class.java:442)
[JVM-2] at org.scalatest.tools.Runner$.genSuiteConfig(Runner.scala:2644)
[JVM-2] at org.scalatest.tools.Runner$$anonfun$37.apply(Runner.scala:2461)
[JVM-2] at org.scalatest.tools.Runner$$anonfun$37.apply(Runner.scala:2460)
[JVM-2] at scala.collection.immutable.List.map(List.scala:273)
[JVM-2] at org.scalatest.tools.Runner$.doRunRunRunDaDoRunRun(Runner.scala:2460)
[JVM-2] at org.scalatest.tools.Runner$$anonfun$runOptionallyWithPassFailReporter$2.apply(Runner.scala:1044)
[JVM-2] at org.scalatest.tools.Runner$$anonfun$runOptionallyWithPassFailReporter$2.apply(Runner.scala:1043)
[JVM-2] at org.scalatest.tools.Runner$.withClassLoaderAndDispatchReporter(Runner.scala:2722)
[JVM-2] at org.scalatest.tools.Runner$.runOptionallyWithPassFailReporter(Runner.scala:1043)
[JVM-2] at org.scalatest.tools.Runner$.main(Runner.scala:860)
[JVM-2] at org.scalatest.tools.Runner.main(Runner.scala)
[JVM-2] Cause: java.util.concurrent.TimeoutException: Futures timed out after [1200000 milliseconds]
[JVM-2] at scala.concurrent.impl.Promise$DefaultPromise.ready(Promise.scala:219)
[JVM-2] at scala.concurrent.impl.Promise$DefaultPromise.result(Promise.scala:223)
[JVM-2] at scala.concurrent.Await$$anonfun$result$1.apply(package.scala:190)
[JVM-2] at
Issue by ktoso
Monday Jul 18, 2016 at 17:21 GMT
Originally opened as akka/akka#20986
scaladoc has such filtering feature: https://github.com/scala/scala/blob/2.11.x/src/scaladoc/scala/tools/nsc/doc/Settings.scala#L195-L200
via https://twitter.com/MasseGuillaume/status/755088815547617280
We may want to hide javadsl from scaladoc?
Open questions:
Note to self, the move must preserve authorships of people, don't want to lose that data.
I needed a way to incorporate nested pathPrefix
directives when setting a Location:
header, so its value became nicely relative to the root of the server. Parking the code here, to discuss whether to make an official directive out of it.
Relates to akka/akka#15839 since it does calculate the "matchedPath".
Test:
describe("Directives.addLocationHeader", () -> {
when("nested inside a pathPrefix directive", () -> {
Function<String,Route> route = relativePath ->
pathPrefix("foo", () ->
pathPrefix("bar", () ->
Directives.addLocationHeader(relativePath, () ->
complete("OK"))));
it("should take nested path prefixes into account when building a URL for a relative path segment", () -> {
runRoute(route.apply("hello"), HttpRequest.GET("/foo/bar")).assertHeaderExists(Location.create("/foo/bar/hello"));
runRoute(route.apply("hello"), HttpRequest.GET("/foo/bar/")).assertHeaderExists(Location.create("/foo/bar/hello"));
runRoute(route.apply("hello"), HttpRequest.GET("/foo/bar/baz")).assertHeaderExists(Location.create("/foo/bar/hello"));
});
it("should return the location relative to the root if the relative path segment starts with /", () -> {
runRoute(route.apply("/hello"), HttpRequest.GET("/foo/bar")).assertHeaderExists(Location.create("/hello"));
runRoute(route.apply("/hello"), HttpRequest.GET("/foo/bar/")).assertHeaderExists(Location.create("/hello"));
runRoute(route.apply("/hello"), HttpRequest.GET("/foo/bar/baz")).assertHeaderExists(Location.create("/hello"));
});
it("should return the location relative to the parent if the relative path segment starts with ..", () -> {
runRoute(route.apply("../hello"), HttpRequest.GET("/foo/bar")).assertHeaderExists(Location.create("/foo/hello"));
runRoute(route.apply("../hello"), HttpRequest.GET("/foo/bar/")).assertHeaderExists(Location.create("/foo/hello"));
runRoute(route.apply("../hello"), HttpRequest.GET("/foo/bar/baz")).assertHeaderExists(Location.create("/foo/hello"));
});
});
when("invoked on the root", () -> {
Function<String,Route> route = relativePath ->
Directives.addLocationHeader(relativePath, () ->
complete("OK"));
it("should use the given string as absolute url", () -> {
runRoute(route.apply("hello"), HttpRequest.GET("/")).assertHeaderExists(Location.create("/hello"));
runRoute(route.apply("/hello"), HttpRequest.GET("/")).assertHeaderExists(Location.create("/hello"));
});
});
});
Implementation:
public static Route addLocationHeader(String relativePath, Supplier<Route> inner) {
return extractRequestContext(ctx -> {
String path = ctx.getRequest().getUri().path();
String base = path.substring(0, path.length() - ctx.getUnmatchedPath().length());
if (!base.endsWith("/")) {
base = base + "/";
}
URI uri = URI.create(base).resolve(relativePath);
return respondWithHeader(Location.create(uri.toString()), inner);
});
}
This depends somewhat on #3, as "core" is already a stable module.
We have not, and should not do breaking changes in it.
"Akka HTTP", this repo, shall contain the directives and other high level stuff.
These high level things MAY need to do slightly more binary compatibility trade-offs.
The MAY originates from the the very user-facing-dsl aspect of this DSL.
Akka HTTP Core is already a stable module and should not change.
We have 2 options:
I originally started with the idea 1) however recently started leaning to 2), since if someone wants to file a bug, they'd come over to "the akka http repo" of course so it would be confusing to tell them that they should report "oh that thing" in akka/akka.
I propose we move the akka-http-core code into this repo as well (opposite as I was proposing a long time ago).
Opinions?
Issue by skonto
Saturday Jun 18, 2016 at 14:04 GMT
Originally opened as akka/akka#20800
Example code
val customSink: Sink[Message, Future[Done]] = Sink.foreach(f)
val flow: Flow[Message, Message, Promise[Option[Message]]] =
Flow.fromSinkAndSourceMat(
customSink,
Source.maybe[Message])(Keep.right)
val (upgradeResponse, promise) =
Http().singleWebSocketRequest(
WebSocketRequest(url),
flow.alsoTo(Sink.onComplete {
case Failure(e) => e.printStackTrace()
case Success(v) =>
}))
Nothing happens if i start the without connection, no exception got there.
Issue by ktoso
Tuesday Jun 28, 2016 at 14:27 GMT
Originally opened as akka/akka#20848
Unmarshal must force a poper full draining of the http response body otherwise we might end up stalling connections. Reported by some users via akka/akka#20192
I'm not 100% certain we do the right thing currently, so this ticket is about making sure that if an unmarshaller fails we do actually consume the entire incoming stream.
Issue by ktoso
Sunday Jul 10, 2016 at 14:01 GMT
Originally opened as akka/akka#20929
... and the next path will be attempted to be resolved.
This is different than exceptions inside of a route already (which would render as 500 usually then)
Issue by alias1
Monday Jul 11, 2016 at 09:53 GMT
Originally opened as akka/akka#20934
I'm trying to put together some API abstractions using the request level client API, and running into this error message:
Cannot determine request scheme and target endpoint as HttpMethod(GET) request to /self/calendar doesn't have an absolute URI
I would have expected this should work since I am setting the host header. Cut down example:
# Initial Setup/Abstractions
val http = Http(system)
val hostHeader = headers.Host("api.meetup.com")
val request = HttpRequest().withDefaultHeaders(hostHeader)
val get = request.withMethod(HttpMethods.GET)
# Individual Call
val uri = Uri().withPath(Path("/self/calendar"))
val r = get.withUri(uri)
http.singleRequest(r)
Printing the request:
HttpRequest(HttpMethod(GET),/self/calendar,List(Host: api.meetup.com),HttpEntity.Strict(none/none,ByteString()),HttpProtocol(HTTP/1.1))
I have also tried using the following, to no avail:
val request = HttpRequest().withHeaders(hostHeader)
val request = HttpRequest().withEffectiveUri(true, hostHeader)
.toRelative
on the uri
chainget.withUri("/self/calendar")
Though if I use val request = get.withUri(uri).withEffectiveUri(true, hostHeader)
then the request works.
Am I missing something? Or is there a better way to build up these abstractions? I would have expected that so long as I had the host/effective URL set somewhere in the chain it should work fine.
Issue by stejskal
Wednesday Jul 06, 2016 at 22:17 GMT
Originally opened as akka/akka#20902
opened based on discussion in forum:
https://groups.google.com/forum/#!topic/akka-user/MJJADiJZOrY
Scala version: 2.11.8
akka version: 2.4.6
I have a graph that is leaking http connections from an Akka-Http Superpool when the graph fails. The general shape of the graph is:
A ~> B ~> C ~> D ~> E ~> F
The stage actions are:
A emits HttpRequests
B is the superpool (created by Http().superPool())
C does some validation on the HttpResponse but leaves the ResponseEntity alone
D uses flatMapConcat to emit the ResponseEntity ByteString Sources
E Collects the ResponseEntity ByteStrings into a single chunk
F further processes the ByteString
Which all works fine if the graph completes successfully. However if stage F fails and fails the graph then one connection is never released and eventually a materialization of the graph stalls forever waiting for stage B to process a request.
What I believe is happening is that when stage F fails, the ResponseEntity Source in stage C has not been sinked and so the connection it occupies is not released.
Is there a way to sink that ResponseEntity source on graph failure or alternatively, is there a better way to push a series of HttpRequests through a superPool that is aware of the entityBody and sinks on failure?
Here is a reproducible example of this problem.
import java.util.concurrent.Executors
import akka.NotUsed
import akka.actor.ActorSystem
import akka.http.scaladsl.Http
import akka.http.scaladsl.model._
import akka.stream.ActorMaterializer
import akka.stream.scaladsl.{Flow, Keep, Sink, Source}
import akka.util.ByteString
import scala.collection.immutable.Seq
import scala.concurrent.duration._
import scala.concurrent.{ExecutionContext, Future}
import scala.util.{Failure, Success, Try}
object Example
{
implicit val system = ActorSystem("example")
implicit val ec = ExecutionContext.fromExecutorService(Executors.newSingleThreadExecutor())
implicit val mat = ActorMaterializer()
var requestCount = 0
val pool = Http().superPool[NotUsed]()
val requestSource = Source.repeat((HttpRequest(uri = "https://www.google.com/images/branding/googlelogo/2x/googlelogo_color_272x92dp.png"), NotUsed))
val responseValidationFlow = Flow[(Try[HttpResponse], NotUsed)].collect
{
case (Success(response), _) =>
if (response.status == StatusCodes.OK) Success(response.entity)
else
{
response.entity.dataBytes.runWith(Sink.ignore)
Failure(new Exception("non 200 status"))
}
}.collect
{
case Success(entity) => entity
}
val extractBodyFlowWithStrict = Flow[ResponseEntity].mapAsync(1)(entity => entity.toStrict(2 seconds).map(_.data))
val failableFlow = Flow[ByteString].map
{
imageBytes =>
requestCount += 1
if (requestCount % 2 == 0) throw new Exception("i blew up")
println(imageBytes)
imageBytes
}
val sink: Sink[ByteString, Future[Seq[ByteString]]] = Sink.seq[ByteString]
val runnableGraph = requestSource
.via(pool)
.via(responseValidationFlow)
.via(extractBodyFlowWithStrict)
.via(failableFlow)
.toMat(sink)(Keep.right)
def main(args: Array[String]): Unit =
{
materialize
}
def materialize: Unit =
{
println("starting graph")
runnableGraph.run().recover
{
case t =>
println(s"graph failed ${t.getMessage}")
materialize
}
}
}
thanks,
Spencer
Issue by jrudolph
Monday Sep 05, 2016 at 12:45 GMT
Originally opened as akka/akka#21370
PoolFlow, PoolConductor, and PoolSlot do a sophisticated dance to send state information around. This could probably be much simplified by having just two GraphStages:
Flow[HttpRequest, HttpResponse]
and creates the slots internally, or could have the shape of a flow plus additional holes for all of the slots to be filled from the outside. Issue by huntc
Tuesday Aug 23, 2016 at 04:41 GMT
Originally opened as akka/akka#21251
Http clients such as curl
expect a 100 Continue
response in relation to posting a large amount of data. As of Akka 2.4.7, but also observed with 2.4.9, there appears to be a race condition where sub streams formed as part of prefixAndTail
fail. The problem is hard to reproduce but manifests itself with the following stack trace when it does occur:
2016-08-23T04:26:51Z ChristophersMBP.lan ERROR FileSubscriber [sourceThread=conductr-akka.stream.default-blocking-io-dispatcher-219, akkaTimestamp=04:26:51.172UTC, akkaSource=akka.tcp://[email protected]:9004/user/reaper/control-server/StreamSupervisor-9/flow-1262-1-fileSink, sourceActorSystem=conductr] - Tearing down FileSink(/var/folders/7g/ftd7sqz132zcfj_rlpb6xpw00000gn/T/2f522926-3f58-403e-9b5f-ea7923a6a92f6630142131447228655/default/bundle.conf) due to upstream error
akka.stream.impl.SubscriptionTimeoutException: Substream Source has not been materialized in 5000 milliseconds
at akka.stream.impl.fusing.SubSource.timeout(StreamOfStreams.scala:693)
at akka.stream.stage.GraphStageLogic$SubSourceOutlet.timeout(GraphStage.scala:1035)
at akka.stream.impl.fusing.PrefixAndTail$PrefixAndTailLogic.onTimer(StreamOfStreams.scala:130)
at <snip>
2016-08-23T04:26:51Z ChristophersMBP.lan ERROR RepointableActorRef [sourceThread=conductr-akka.actor.default-dispatcher-5, akkaTimestamp=04:26:51.174UTC, akkaSource=akka://conductr/user/reaper/control-server/StreamSupervisor-9/flow-1253-0-unknown-operation, sourceActorSystem=conductr] - Error in stage [akka.http.impl.engine.server.HttpServerBluePrint$ControllerStage@3d1662e9]: requirement failed: Cannot push port (requestPrepOut) twice
java.lang.IllegalArgumentException: requirement failed: Cannot push port (requestPrepOut) twice
at scala.Predef$.require(Predef.scala:224)
at akka.stream.stage.GraphStageLogic.push(GraphStage.scala:436)
at akka.http.impl.engine.server.HttpServerBluePrint$ControllerStage$$anon$12$$anon$13.onPush(HttpServerBluePrint.scala:395)
at <snip>
From a client perspective:
curl -v --header "User-Agent: Typesafe ConductR Docker Entrypoint Script" --form "bundleConf=@conductr-grafana/target/bundle/bundle/t37684c33cbd8b075fb837e218bc6eebc2d60671f3ba3cc90f.zip" 127.0.0.1:9005/v2/bundles/0-372d167b74f9ebc3
* Trying 127.0.0.1...
* Connected to 127.0.0.1 (127.0.0.1) port 9005 (#0)
> POST /v2/bundles/ HTTP/1.1
> Host: 127.0.0.1:9005
> Accept: */*
> User-Agent: Typesafe ConductR Docker Entrypoint Script
> Content-Length: 45461050
> Expect: 100-continue
> Content-Type: multipart/form-data; boundary=------------------------de24fc60856e2fb5
>
< HTTP/1.1 100 Continue
< Server: akka-http/2.4.7
< Date: Tue, 23 Aug 2016 04:26:46 GMT
< HTTP/1.1 400 Bad Request
< Server: akka-http/2.4.7
< Date: Tue, 23 Aug 2016 04:26:51 GMT
< Connection: close
< Content-Type: text/plain; charset=UTF-8
< Content-Length: 59
<
* we are done reading and this is set to close, stop send
* Closing connection 0
In the above response times Akka http will respond with the 100
relatively quickly.
In situations where all is well, 100
is also being replied:
< HTTP/1.1 100 Continue
< Server: akka-http/2.4.7
< Date: Tue, 23 Aug 2016 04:26:44 GMT
< HTTP/1.1 200 OK
< Server: akka-http/2.4.7
< Date: Tue, 23 Aug 2016 04:26:44 GMT
< Content-Type: application/json
< Content-Length: 98
Possibly related, although I am not suggesting that we should avoid sending a 100 Continue
response. As stated, some clients, such as curl
expect the 100
status.
As a work-around I can disable curl
's expect behaviour with `-H "Expect: ".
Full stack trace: stack.txt
If a route is using an unmarshaller with framing (or an actual streaming unmarshaller), e.g.
// unmarshal an incoming JSON array into individual string elements
Unmarshaller<HttpEntity, Source<String,NotUsed>> unmarshaller;
Route route = entity(unmarshaller, source ->
onSuccess(() -> source.runWith(Sink.fold("", (s1, s2) -> s1 + s2), materializer), s ->
complete(s)
)
);
then currently, if there's a JSON parse (or JSON API) failure in the stream, runWith
will fail the future with that exception, hitting the default exception handler. However, MarshallingDirectives
has some nice default handling of IllegalArgumentException
that is very similar to this failure situation.
Suggested way forward:
IllegalArgumentException
(ValidationException
, ProtocolException
, ?) for unmarshalling/protocol failuresValidationRejection
in the default exception handlerOf course, this only applies to routes where the source is completely consumed before invoking complete
(the status code can't be changed if the user has already begun writing a response).
Issue by stanch
Monday Apr 25, 2016 at 23:10 GMT
Originally opened as akka/akka#20393
I think Test Kit is great and it’s really a shame that we have to dispense with the nice syntax it offers when testing external web services or doing system tests. In those cases there is no Route
, but an HTTP client can be used to provide a function HttpRequest => Future[HttpResponse]
. The problem is that RouteTestResult
, which is needed for the check
DSL to work, can only be initialized within the akka.http.scaladsl.testkit
package. Other than that, it seems straight-forward to initialize RouteTestResult
from an HttpResponse
. Here is what it takes:
implicit def injectIntoPipeline(
implicit timeout: RouteTestTimeout,
defaultHostInfo: DefaultHostInfo,
executionContext: ExecutionContext
): TildeArrow[HttpRequest, Future[HttpResponse]] { type Out = RouteTestResult } =
new TildeArrow[HttpRequest, Future[HttpResponse]] {
type Out = RouteTestResult
def apply(request: HttpRequest, pipeline: HttpRequest => Future[HttpResponse]): Out = {
val routeTestResult = new RouteTestResult(timeout.duration)
pipeline(request) foreach { response =>
routeTestResult.handleResult(RouteResult.Complete(response))
}
routeTestResult
}
}
With this added to akka.http.scaladsl.testkit.RouteTest
, the following can be done:
// define your HTTP pipeline (I think this is what it was called in Spray)
def pipeline(
implicit system: ActorSystem, materializer: Materializer, ec: ExecutionContext
): HttpRequest => Future[HttpResponse] = { request =>
Http().singleRequest {
request
.withUri(???) // add your custom host/port
.withHeaders(???) // add your custom auth
}
}
// test away
val request = Json.obj("foo" -> "bar")
Post("/v1.1/api", request) ~> pipeline ~> check {
// your custom checks
}
Issue by patriknw
Monday Sep 05, 2016 at 07:00 GMT
Originally opened as akka/akka#21364
https://jenkins.akka.io:8498/job/akka-multi-node-nightly/4155/consoleFull
[info] DontLeakActorsOnFailingConnectionSpecs:
[info] Http.superPool
4: FAIL: akka.http.impl.engine.client.PoolSlot$UnexpectedDisconnectException: Unexpected (early) disconnect
activeShells (actor: Actor[akka://DontLeakActorsOnFailingConnectionSpecs/user/StreamSupervisor-101/flow-0-0-unknown-operation#-930512737]):
GraphInterpreterShell
GraphAssembly
[ GraphStage(StatefulMapConcat) [070d76df] [Name(iterableSource), Name(statefulMapConcat)]
GraphStage(Map(<function1>)) [115f1049] [Name(map)]
GraphStage(SingleSource(Range(1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23, 24, 25, 26, 27, 28, 29, 30, 31, 32, 33, 34, 35, 36, 37, 38, 39, 40, 41, 42, 43, 44, 45, 46, 47, 48, 49, 50, 51, 52, 53, 54, 55, 56, 57, 58, 59, 60, 61, 62, 63, 64, 65, 66, 67, 68, 69, 70, 71, 72, 73, 74, 75, 76, 77, 78, 79, 80, 81, 82, 83, 84, 85, 86, 87, 88, 89, 90, 91, 92, 93, 94, 95, 96, 97, 98, 99, 100))) [09ca984e] [Name(iterableSource), Name(singleSource)]
GraphStage(MapAsyncUnordered(4,<function1>)) [0df51daf] [Name(mapAsyncUnordered)]
GraphStage(Map(<function1>)) [48f79275] [Name(foreachSink), Name(foreachSink), Name(map)]
]
[StatefulMapConcat.in,Map.in,MapAsyncUnordered.in,Map.in,null]
[0,1,3,4,-1]
[single.out,StatefulMapConcat.out,Map.out,MapAsyncUnordered.out,Map.out]
[2,0,1,3,4]digraph waits {
N0 [label="StatefulMapConcat"]
N1 [label="Map(<function1>)"]
N2 [label="SingleSource(Range(1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23, 24, 25, 26, 27, 28, 29, 30, 31, 32, 33, 34, 35, 36, 37, 38, 39, 40, 41, 42, 43, 44, 45, 46, 47, 48, 49, 50, 51, 52, 53, 54, 55, 56, 57, 58, 59, 60, 61, 62, 63, 64, 65, 66, 67, 68, 69, 70, 71, 72, 73, 74, 75, 76, 77, 78, 79, 80, 81, 82, 83, 84, 85, 86, 87, 88, 89, 90, 91, 92, 93, 94, 95, 96, 97, 98, 99, 100))"]
N3 [label="MapAsyncUnordered(4,<function1>)"]
N4 [label="Map(<function1>)"]
N1 -> N0 [label=shouldPull; color=blue]
N3 -> N1 [label=shouldPull; color=blue]
N3 -> N4 [label=shouldPush; color=red];
N4 -> Out4 [label=shouldPush; color=red];
}
// (8, 29, 29)() (running=4, shutdown=1,2,0,2,2)newShells:
[info] - should not leak connection Actors when hitting non-existing endpoint *** FAILED *** (19 seconds, 973 milliseconds)
[info] java.lang.AssertionError: assertion failed: expected no StreamSupervisor children, but got [Actor[akka://DontLeakActorsOnFailingConnectionSpecs/user/StreamSupervisor-101/flow-0-0-unknown-operation#-930512737]]
[info] at scala.Predef$.assert(Predef.scala:170)
[info] at akka.stream.testkit.Utils$$anonfun$assertAllStagesStopped$1$$anonfun$apply$mcV$sp$1.apply$mcV$sp(Utils.scala:31)
[info] at akka.stream.testkit.Utils$$anonfun$assertAllStagesStopped$1$$anonfun$apply$mcV$sp$1.apply(Utils.scala:28)
[info] at akka.stream.testkit.Utils$$anonfun$assertAllStagesStopped$1$$anonfun$apply$mcV$sp$1.apply(Utils.scala:28)
[info] at akka.testkit.TestKitBase$class.poll$2(TestKit.scala:303)
[info] at akka.testkit.TestKitBase$class.awaitAssert(TestKit.scala:314)
[info] at akka.testkit.TestKit.awaitAssert(TestKit.scala:814)
[info] at akka.stream.testkit.Utils$$anonfun$assertAllStagesStopped$1.apply$mcV$sp(Utils.scala:28)
[info] at akka.stream.testkit.Utils$$anonfun$assertAllStagesStopped$1.apply(Utils.scala:26)
[info] at akka.stream.testkit.Utils$$anonfun$assertAllStagesStopped$1.apply(Utils.scala:26)
[info] at akka.testkit.TestKitBase$class.within(TestKit.scala:345)
[info] at akka.testkit.TestKit.within(TestKit.scala:814)
[info] at akka.testkit.TestKitBase$class.within(TestKit.scala:359)
[info] at akka.testkit.TestKit.within(TestKit.scala:814)
[info] at akka.stream.testkit.Utils$.assertAllStagesStopped(Utils.scala:26)
[info] at akka.http.scaladsl.server.DontLeakActorsOnFailingConnectionSpecs$$anonfun$1$$anonfun$apply$mcV$sp$1.apply$mcV$sp(DontLeakActorsOnFailingConnectionSpecs.scala:40)
[info] at akka.http.scaladsl.server.DontLeakActorsOnFailingConnectionSpecs$$anonfun$1$$anonfun$apply$mcV$sp$1.apply(DontLeakActorsOnFailingConnectionSpecs.scala:40)
[info] at akka.http.scaladsl.server.DontLeakActorsOnFailingConnectionSpecs$$anonfun$1$$anonfun$apply$mcV$sp$1.apply(DontLeakActorsOnFailingConnectionSpecs.scala:40)
Issue by jypma
Thursday Sep 01, 2016 at 13:37 GMT
Originally opened as akka/akka#21340
The Java Marshaller
and Unmarshaller
classes are missing a few of the scala variants that make it impossible to write streaming marshallers and unmarshallers in Java.
I've already written the needed additions for Marshaller and Unmarshaller.
Shall I create a PR?
Issue by jrudolph
Monday Sep 05, 2016 at 12:45 GMT
Originally opened as akka/akka#21370
PoolFlow, PoolConductor, and PoolSlot do a sophisticated dance to send state information around. This could probably be much simplified by having just two GraphStages:
Flow[HttpRequest, HttpResponse]
and creates the slots internally, or could have the shape of a flow plus additional holes for all of the slots to be filled from the outside.A declarative, efficient, and flexible JavaScript library for building user interfaces.
🖖 Vue.js is a progressive, incrementally-adoptable JavaScript framework for building UI on the web.
TypeScript is a superset of JavaScript that compiles to clean JavaScript output.
An Open Source Machine Learning Framework for Everyone
The Web framework for perfectionists with deadlines.
A PHP framework for web artisans
Bring data to life with SVG, Canvas and HTML. 📊📈🎉
JavaScript (JS) is a lightweight interpreted programming language with first-class functions.
Some thing interesting about web. New door for the world.
A server is a program made to process requests and deliver data to clients.
Machine learning is a way of modeling and interpreting data that allows a piece of software to respond intelligently.
Some thing interesting about visualization, use data art
Some thing interesting about game, make everyone happy.
We are working to build community through open source technology. NB: members must have two-factor auth.
Open source projects and samples from Microsoft.
Google ❤️ Open Source for everyone.
Alibaba Open Source for everyone
Data-Driven Documents codes.
China tencent open source team.