|
| 1 | +/* |
| 2 | + * Copyright 2013 http4s.org |
| 3 | + * |
| 4 | + * Licensed under the Apache License, Version 2.0 (the "License"); |
| 5 | + * you may not use this file except in compliance with the License. |
| 6 | + * You may obtain a copy of the License at |
| 7 | + * |
| 8 | + * http://www.apache.org/licenses/LICENSE-2.0 |
| 9 | + * |
| 10 | + * Unless required by applicable law or agreed to in writing, software |
| 11 | + * distributed under the License is distributed on an "AS IS" BASIS, |
| 12 | + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| 13 | + * See the License for the specific language governing permissions and |
| 14 | + * limitations under the License. |
| 15 | + */ |
| 16 | + |
| 17 | +package org.http4s |
| 18 | +package servlet |
| 19 | + |
| 20 | +import cats.Monoid |
| 21 | +import cats.effect.Deferred |
| 22 | +import cats.effect.IO |
| 23 | +import cats.effect.Resource |
| 24 | +import cats.effect.std.Dispatcher |
| 25 | +import cats.syntax.all._ |
| 26 | +import fs2.Chunk |
| 27 | +import fs2.Stream |
| 28 | +import munit.CatsEffectSuite |
| 29 | +import org.eclipse.jetty.client.AsyncRequestContent |
| 30 | +import org.eclipse.jetty.client.BytesRequestContent |
| 31 | +import org.eclipse.jetty.client.HttpClient |
| 32 | +import org.eclipse.jetty.client.{Response => JResponse} |
| 33 | +import org.eclipse.jetty.util.Callback |
| 34 | +import org.http4s.dsl.io._ |
| 35 | +import org.http4s.syntax.all._ |
| 36 | + |
| 37 | +import java.nio.ByteBuffer |
| 38 | +import scala.concurrent.duration._ |
| 39 | + |
| 40 | +class AsyncHttp4sServletSuite extends CatsEffectSuite { |
| 41 | + private val clientR = Resource.make(IO { |
| 42 | + val client = new HttpClient() |
| 43 | + client.start() |
| 44 | + client |
| 45 | + })(client => IO(client.stop())) |
| 46 | + |
| 47 | + private lazy val service = HttpRoutes |
| 48 | + .of[IO] { |
| 49 | + case GET -> Root / "simple" => |
| 50 | + Ok("simple") |
| 51 | + case req @ POST -> Root / "echo" => |
| 52 | + Ok(req.body) |
| 53 | + case GET -> Root / "shifted" => |
| 54 | + // Wait for a bit to make sure we lose the race |
| 55 | + (IO.sleep(50.millis) *> |
| 56 | + Ok("shifted")).evalOn(munitExecutionContext) |
| 57 | + case GET -> Root / "never" => |
| 58 | + IO.never |
| 59 | + } |
| 60 | + .orNotFound |
| 61 | + |
| 62 | + private def servletServer(asyncTimeout: FiniteDuration = 10.seconds) = |
| 63 | + ResourceFunFixture[Int]( |
| 64 | + Dispatcher.parallel[IO].flatMap(d => TestEclipseServer(servlet(d, asyncTimeout))) |
| 65 | + ) |
| 66 | + |
| 67 | + private def get(client: HttpClient, serverPort: Int, path: String): IO[String] = |
| 68 | + IO.blocking( |
| 69 | + client.GET(s"http://127.0.0.1:$serverPort/$path") |
| 70 | + ).map(_.getContentAsString) |
| 71 | + |
| 72 | + servletServer().test("AsyncHttp4sServlet handle GET requests") { server => |
| 73 | + clientR.use(get(_, server, "simple")).assertEquals("simple") |
| 74 | + } |
| 75 | + |
| 76 | + // We should handle an empty body |
| 77 | + servletServer().test("AsyncHttp4sServlet handle empty POST") { server => |
| 78 | + clientR |
| 79 | + .use { client => |
| 80 | + IO.blocking( |
| 81 | + client |
| 82 | + .POST(s"http://127.0.0.1:$server/echo") |
| 83 | + .send() |
| 84 | + ).map(resp => Chunk.array(resp.getContent)) |
| 85 | + } |
| 86 | + .assertEquals(Chunk.empty) |
| 87 | + } |
| 88 | + |
| 89 | + // We should handle a regular, big body |
| 90 | + servletServer().test("AsyncHttp4sServlet handle multiple chunks upfront") { server => |
| 91 | + val bytes = Stream.range(0, DefaultChunkSize * 2).map(_.toByte).to(Array) |
| 92 | + clientR |
| 93 | + .use { client => |
| 94 | + IO.blocking( |
| 95 | + client |
| 96 | + .POST(s"http://127.0.0.1:$server/echo") |
| 97 | + .body(new BytesRequestContent(bytes)) |
| 98 | + .send() |
| 99 | + ).map(resp => Chunk.array(resp.getContent)) |
| 100 | + } |
| 101 | + .assertEquals(Chunk.array(bytes)) |
| 102 | + } |
| 103 | + |
| 104 | + // We should be able to wake up if we're initially blocked |
| 105 | + servletServer().test("AsyncHttp4sServlet handle single-chunk, deferred POST") { server => |
| 106 | + val bytes = Stream.range(0, DefaultChunkSize).map(_.toByte).to(Array) |
| 107 | + clientR |
| 108 | + .use { client => |
| 109 | + for { |
| 110 | + content <- IO(new AsyncRequestContent()) |
| 111 | + bodyFiber <- IO |
| 112 | + .async_[Chunk[Byte]] { cb => |
| 113 | + var body = Chunk.empty[Byte] |
| 114 | + client |
| 115 | + .POST(s"http://127.0.0.1:$server/echo") |
| 116 | + .body(content) |
| 117 | + .send(new JResponse.Listener { |
| 118 | + override def onContent(resp: JResponse, bb: ByteBuffer) = { |
| 119 | + val buf = new Array[Byte](bb.remaining()) |
| 120 | + bb.get(buf) |
| 121 | + body ++= Chunk.array(buf) |
| 122 | + } |
| 123 | + override def onFailure(resp: JResponse, t: Throwable) = |
| 124 | + cb(Left(t)) |
| 125 | + override def onSuccess(resp: JResponse) = |
| 126 | + cb(Right(body)) |
| 127 | + }) |
| 128 | + } |
| 129 | + .start |
| 130 | + _ <- IO(content.write(ByteBuffer.wrap(bytes), Callback.NOOP)) |
| 131 | + _ <- IO(content.close()) |
| 132 | + body <- bodyFiber.joinWithNever |
| 133 | + } yield body |
| 134 | + } |
| 135 | + .assertEquals(Chunk.array(bytes)) |
| 136 | + } |
| 137 | + |
| 138 | + // We should be able to wake up after being blocked |
| 139 | + servletServer().test("AsyncHttp4sServlet handle two-chunk, deferred POST") { server => |
| 140 | + // Show that we can read, be blocked, and read again |
| 141 | + val bytes = Stream.range(0, DefaultChunkSize).map(_.toByte).to(Array) |
| 142 | + Dispatcher |
| 143 | + .parallel[IO] |
| 144 | + .use { dispatcher => |
| 145 | + clientR.use { client => |
| 146 | + for { |
| 147 | + content <- IO(new AsyncRequestContent()) |
| 148 | + firstChunkReceived <- Deferred[IO, Unit] |
| 149 | + bodyFiber <- IO |
| 150 | + .async_[Chunk[Byte]] { cb => |
| 151 | + var body = Chunk.empty[Byte] |
| 152 | + client |
| 153 | + .POST(s"http://127.0.0.1:$server/echo") |
| 154 | + .body(content) |
| 155 | + .send(new JResponse.Listener { |
| 156 | + override def onContent(resp: JResponse, bb: ByteBuffer) = |
| 157 | + dispatcher.unsafeRunSync(for { |
| 158 | + _ <- firstChunkReceived.complete(()).attempt |
| 159 | + buf <- IO(new Array[Byte](bb.remaining())) |
| 160 | + _ <- IO(bb.get(buf)) |
| 161 | + _ <- IO { body = body ++ Chunk.array(buf) } |
| 162 | + } yield ()) |
| 163 | + override def onFailure(resp: JResponse, t: Throwable) = |
| 164 | + cb(Left(t)) |
| 165 | + override def onSuccess(resp: JResponse) = |
| 166 | + cb(Right(body)) |
| 167 | + }) |
| 168 | + } |
| 169 | + .start |
| 170 | + _ <- IO(content.write(ByteBuffer.wrap(bytes), Callback.NOOP)) |
| 171 | + _ <- firstChunkReceived.get |
| 172 | + _ <- IO(content.write(ByteBuffer.wrap(bytes), Callback.NOOP)) |
| 173 | + _ <- IO(content.close()) |
| 174 | + body <- bodyFiber.joinWithNever |
| 175 | + } yield body |
| 176 | + } |
| 177 | + } |
| 178 | + .assertEquals(Monoid[Chunk[Byte]].combineN(Chunk.array(bytes), 2)) |
| 179 | + } |
| 180 | + |
| 181 | + // We shouldn't block when we receive less than a chunk at a time |
| 182 | + servletServer().test("AsyncHttp4sServlet handle two itsy-bitsy deferred chunk POST") { server => |
| 183 | + Dispatcher |
| 184 | + .parallel[IO] |
| 185 | + .use { dispatcher => |
| 186 | + clientR.use { client => |
| 187 | + for { |
| 188 | + content <- IO(new AsyncRequestContent()) |
| 189 | + firstChunkReceived <- Deferred[IO, Unit] |
| 190 | + bodyFiber <- IO |
| 191 | + .async_[Chunk[Byte]] { cb => |
| 192 | + var body = Chunk.empty[Byte] |
| 193 | + client |
| 194 | + .POST(s"http://127.0.0.1:$server/echo") |
| 195 | + .body(content) |
| 196 | + .send(new JResponse.Listener { |
| 197 | + override def onContent(resp: JResponse, bb: ByteBuffer) = |
| 198 | + dispatcher.unsafeRunSync(for { |
| 199 | + _ <- firstChunkReceived.complete(()).attempt |
| 200 | + buf <- IO(new Array[Byte](bb.remaining())) |
| 201 | + _ <- IO(bb.get(buf)) |
| 202 | + _ <- IO { body = body ++ Chunk.array(buf) } |
| 203 | + } yield ()) |
| 204 | + override def onFailure(resp: JResponse, t: Throwable) = |
| 205 | + cb(Left(t)) |
| 206 | + override def onSuccess(resp: JResponse) = |
| 207 | + cb(Right(body)) |
| 208 | + }) |
| 209 | + } |
| 210 | + .start |
| 211 | + _ <- IO(content.write(ByteBuffer.wrap(Array[Byte](0.toByte)), Callback.NOOP)) |
| 212 | + _ <- firstChunkReceived.get |
| 213 | + _ <- IO(content.write(ByteBuffer.wrap(Array[Byte](1.toByte)), Callback.NOOP)) |
| 214 | + _ <- IO(content.close()) |
| 215 | + body <- bodyFiber.joinWithNever |
| 216 | + } yield body |
| 217 | + } |
| 218 | + } |
| 219 | + .assertEquals(Chunk(0.toByte, 1.toByte)) |
| 220 | + } |
| 221 | + |
| 222 | + servletServer().test("AsyncHttp4sServlet should not reorder lots of itsy-bitsy chunks") { |
| 223 | + server => |
| 224 | + val body = (0 until 4096).map(_.toByte).toArray |
| 225 | + Dispatcher |
| 226 | + .parallel[IO] |
| 227 | + .use { dispatcher => |
| 228 | + clientR.use { client => |
| 229 | + for { |
| 230 | + content <- IO(new AsyncRequestContent()) |
| 231 | + bodyFiber <- IO |
| 232 | + .async_[Chunk[Byte]] { cb => |
| 233 | + var body = Chunk.empty[Byte] |
| 234 | + client |
| 235 | + .POST(s"http://127.0.0.1:$server/echo") |
| 236 | + .body(content) |
| 237 | + .send(new JResponse.Listener { |
| 238 | + override def onContent(resp: JResponse, bb: ByteBuffer): Unit = |
| 239 | + dispatcher.unsafeRunSync(for { |
| 240 | + buf <- IO(new Array[Byte](bb.remaining())) |
| 241 | + _ <- IO(bb.get(buf)) |
| 242 | + _ <- IO { body = body ++ Chunk.array(buf) } |
| 243 | + } yield ()) |
| 244 | + override def onFailure(resp: JResponse, t: Throwable): Unit = |
| 245 | + cb(Left(t)) |
| 246 | + override def onSuccess(resp: JResponse): Unit = |
| 247 | + cb(Right(body)) |
| 248 | + }) |
| 249 | + } |
| 250 | + .start |
| 251 | + _ <- body.toList.traverse_(b => |
| 252 | + IO(content.write(ByteBuffer.wrap(Array[Byte](b)), Callback.NOOP)) >> IO( |
| 253 | + content.flush() |
| 254 | + ) |
| 255 | + ) |
| 256 | + _ <- IO(content.close()) |
| 257 | + body <- bodyFiber.joinWithNever |
| 258 | + } yield body |
| 259 | + } |
| 260 | + } |
| 261 | + .assertEquals(Chunk.array(body)) |
| 262 | + } |
| 263 | + |
| 264 | + servletServer().test("AsyncHttp4sServlet work for shifted IO") { server => |
| 265 | + clientR.use(get(_, server, "shifted")).assertEquals("shifted") |
| 266 | + } |
| 267 | + |
| 268 | + servletServer(3.seconds).test("AsyncHttp4sServlet timeout fires") { server => |
| 269 | + clientR.use(get(_, server, "never")).map(_.contains("Error 500 AsyncContext timeout")) |
| 270 | + } |
| 271 | + |
| 272 | + private def servlet(dispatcher: Dispatcher[IO], asyncTimeout: FiniteDuration) = |
| 273 | + AsyncHttp4sServlet |
| 274 | + .builder[IO](service, dispatcher) |
| 275 | + .withChunkSize(DefaultChunkSize) |
| 276 | + .withAsyncTimeout(asyncTimeout) |
| 277 | + .build |
| 278 | +} |
0 commit comments