Skip to content

Commit 33d4bf0

Browse files
rossabakerrossabaker
authored
Merge pull request #806 from http4s/http4s-0.23.19
Upgrade to http4s-0.23.19
2 parents c308b23 + 796d959 commit 33d4bf0

File tree

11 files changed

+152
-122
lines changed

11 files changed

+152
-122
lines changed

‎blaze-client/src/main/scala/org/http4s/blaze/client/Http1Connection.scala‎

Lines changed: 12 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -214,7 +214,10 @@ private final class Http1Connection[F[_]](
214214
}
215215

216216
val idleTimeoutF: F[TimeoutException] = idleTimeoutStage match {
217-
case Some(stage) => F.async_[TimeoutException](stage.setTimeout)
217+
case Some(stage) =>
218+
F.async[TimeoutException] { cb =>
219+
F.delay(stage.setTimeout(cb)).as(Some(F.delay(stage.cancelTimeout())))
220+
}
218221
case None => F.never[TimeoutException]
219222
}
220223

@@ -254,15 +257,20 @@ private final class Http1Connection[F[_]](
254257
}
255258
}
256259

260+
private[this] val shutdownCancelToken = Some(F.delay(stageShutdown()))
261+
257262
private def receiveResponse(
258263
closeOnFinish: Boolean,
259264
doesntHaveBody: Boolean,
260265
idleTimeoutS: F[Either[Throwable, Unit]],
261266
idleRead: Option[Future[ByteBuffer]],
262267
): F[Response[F]] =
263-
F.async_[Response[F]] { cb =>
264-
val read = idleRead.getOrElse(channelRead())
265-
handleRead(read, cb, closeOnFinish, doesntHaveBody, "Initial Read", idleTimeoutS)
268+
F.async[Response[F]] { cb =>
269+
F.delay {
270+
val read = idleRead.getOrElse(channelRead())
271+
handleRead(read, cb, closeOnFinish, doesntHaveBody, "Initial Read", idleTimeoutS)
272+
shutdownCancelToken
273+
}
266274
}
267275

268276
// this method will get some data, and try to continue parsing using the implicit ec

‎blaze-client/src/main/scala/org/http4s/blaze/client/PoolManager.scala‎

Lines changed: 64 additions & 60 deletions
Original file line numberDiff line numberDiff line change
@@ -182,6 +182,8 @@ private final class PoolManager[F[_], A <: Connection[F]](
182182
idleQueues.update(key, q)
183183
}
184184

185+
private[this] val noopCancelToken = Some(F.unit)
186+
185187
/** This generates a effect of Next Connection. The following calls are executed asynchronously
186188
* with respect to whenever the execution of this task can occur.
187189
*
@@ -203,75 +205,77 @@ private final class PoolManager[F[_], A <: Connection[F]](
203205
*/
204206
def borrow(key: RequestKey): F[NextConnection] =
205207
F.async { callback =>
206-
semaphore.permit.use { _ =>
207-
if (!isClosed) {
208-
def go(): F[Unit] =
209-
getConnectionFromQueue(key).flatMap {
210-
case Some(pooled) if pooled.conn.isClosed =>
211-
F.delay(logger.debug(s"Evicting closed connection for $key: $stats")) *>
212-
decrConnection(key) *>
213-
go()
214-
215-
case Some(pooled) if pooled.borrowDeadline.exists(_.isOverdue()) =>
216-
F.delay(
217-
logger.debug(s"Shutting down and evicting expired connection for $key: $stats")
218-
) *>
219-
decrConnection(key) *>
220-
F.delay(pooled.conn.shutdown()) *>
221-
go()
222-
223-
case Some(pooled) =>
224-
F.delay(logger.debug(s"Recycling connection for $key: $stats")) *>
225-
F.delay(callback(Right(NextConnection(pooled.conn, fresh = false))))
226-
227-
case None if numConnectionsCheckHolds(key) =>
228-
F.delay(
229-
logger.debug(s"Active connection not found for $key. Creating new one. $stats")
230-
) *>
231-
createConnection(key, callback)
232-
233-
case None if maxConnectionsPerRequestKey(key) <= 0 =>
234-
F.delay(callback(Left(NoConnectionAllowedException(key))))
235-
236-
case None if curTotal == maxTotal =>
237-
val keys = idleQueues.keys
238-
if (keys.nonEmpty)
208+
semaphore.permit
209+
.surround {
210+
if (!isClosed) {
211+
def go(): F[Unit] =
212+
getConnectionFromQueue(key).flatMap {
213+
case Some(pooled) if pooled.conn.isClosed =>
214+
F.delay(logger.debug(s"Evicting closed connection for $key: $stats")) *>
215+
decrConnection(key) *>
216+
go()
217+
218+
case Some(pooled) if pooled.borrowDeadline.exists(_.isOverdue()) =>
239219
F.delay(
240-
logger.debug(
241-
s"No connections available for the desired key, $key. Evicting random and creating a new connection: $stats"
242-
)
220+
logger.debug(s"Shutting down and evicting expired connection for $key: $stats")
221+
) *>
222+
decrConnection(key) *>
223+
F.delay(pooled.conn.shutdown()) *>
224+
go()
225+
226+
case Some(pooled) =>
227+
F.delay(logger.debug(s"Recycling connection for $key: $stats")) *>
228+
F.delay(callback(Right(NextConnection(pooled.conn, fresh = false))))
229+
230+
case None if numConnectionsCheckHolds(key) =>
231+
F.delay(
232+
logger.debug(s"Active connection not found for $key. Creating new one. $stats")
243233
) *>
244-
F.delay(keys.iterator.drop(Random.nextInt(keys.size)).next()).flatMap {
245-
randKey =>
246-
getConnectionFromQueue(randKey).map(
247-
_.fold(
248-
logger.warn(s"No connection to evict from the idleQueue for $randKey")
249-
)(_.conn.shutdown())
250-
) *>
251-
decrConnection(randKey)
252-
} *>
253234
createConnection(key, callback)
254-
else
235+
236+
case None if maxConnectionsPerRequestKey(key) <= 0 =>
237+
F.delay(callback(Left(NoConnectionAllowedException(key))))
238+
239+
case None if curTotal == maxTotal =>
240+
val keys = idleQueues.keys
241+
if (keys.nonEmpty)
242+
F.delay(
243+
logger.debug(
244+
s"No connections available for the desired key, $key. Evicting random and creating a new connection: $stats"
245+
)
246+
) *>
247+
F.delay(keys.iterator.drop(Random.nextInt(keys.size)).next()).flatMap {
248+
randKey =>
249+
getConnectionFromQueue(randKey).map(
250+
_.fold(
251+
logger.warn(s"No connection to evict from the idleQueue for $randKey")
252+
)(_.conn.shutdown())
253+
) *>
254+
decrConnection(randKey)
255+
} *>
256+
createConnection(key, callback)
257+
else
258+
F.delay(
259+
logger.debug(
260+
s"No connections available for the desired key, $key. Adding to waitQueue: $stats"
261+
)
262+
) *>
263+
addToWaitQueue(key, callback)
264+
265+
case None => // we're full up. Add to waiting queue.
255266
F.delay(
256267
logger.debug(
257-
s"No connections available for the desired key, $key. Adding to waitQueue: $stats"
268+
s"No connections available for $key. Waiting on new connection: $stats"
258269
)
259270
) *>
260271
addToWaitQueue(key, callback)
272+
}
261273

262-
case None => // we're full up. Add to waiting queue.
263-
F.delay(
264-
logger.debug(
265-
s"No connections available for $key. Waiting on new connection: $stats"
266-
)
267-
) *>
268-
addToWaitQueue(key, callback)
269-
}
270-
271-
F.delay(logger.debug(s"Requesting connection for $key: $stats")).productR(go()).as(None)
272-
} else
273-
F.delay(callback(Left(new IllegalStateException("Connection pool is closed")))).as(None)
274-
}
274+
F.delay(logger.debug(s"Requesting connection for $key: $stats")).productR(go())
275+
} else
276+
F.delay(callback(Left(new IllegalStateException("Connection pool is closed"))))
277+
}
278+
.as(noopCancelToken)
275279
}
276280

277281
private def releaseRecyclable(key: RequestKey, connection: A): F[Unit] =

‎blaze-core/src/main/scala/org/http4s/blazecore/Http1Stage.scala‎

Lines changed: 42 additions & 38 deletions
Original file line numberDiff line numberDiff line change
@@ -197,6 +197,8 @@ private[http4s] trait Http1Stage[F[_]] { self: TailStage[ByteBuffer] =>
197197
// we are not finished and need more data.
198198
else streamingBody(buffer, eofCondition)
199199

200+
private[this] val shutdownCancelToken = Some(F.delay(stageShutdown()))
201+
200202
// Streams the body off the wire
201203
private def streamingBody(
202204
buffer: ByteBuffer,
@@ -205,46 +207,48 @@ private[http4s] trait Http1Stage[F[_]] { self: TailStage[ByteBuffer] =>
205207
@volatile var currentBuffer = buffer
206208

207209
// TODO: we need to work trailers into here somehow
208-
val t = F.async_[Option[Chunk[Byte]]] { cb =>
209-
if (!contentComplete()) {
210-
def go(): Unit =
211-
try {
212-
val parseResult = doParseContent(currentBuffer)
213-
logger.debug(s"Parse result: $parseResult, content complete: ${contentComplete()}")
214-
parseResult match {
215-
case Some(result) =>
216-
cb(Either.right(Chunk.byteBuffer(result).some))
217-
218-
case None if contentComplete() =>
219-
cb(End)
220-
221-
case None =>
222-
channelRead().onComplete {
223-
case Success(b) =>
224-
currentBuffer = BufferTools.concatBuffers(currentBuffer, b)
225-
go()
226-
227-
case Failure(Command.EOF) =>
228-
cb(eofCondition())
229-
230-
case Failure(t) =>
231-
logger.error(t)("Unexpected error reading body.")
232-
cb(Either.left(t))
233-
}
210+
val t = F.async[Option[Chunk[Byte]]] { cb =>
211+
F.delay {
212+
if (!contentComplete()) {
213+
def go(): Unit =
214+
try {
215+
val parseResult = doParseContent(currentBuffer)
216+
logger.debug(s"Parse result: $parseResult, content complete: ${contentComplete()}")
217+
parseResult match {
218+
case Some(result) =>
219+
cb(Either.right(Chunk.byteBuffer(result).some))
220+
221+
case None if contentComplete() =>
222+
cb(End)
223+
224+
case None =>
225+
channelRead().onComplete {
226+
case Success(b) =>
227+
currentBuffer = BufferTools.concatBuffers(currentBuffer, b)
228+
go()
229+
230+
case Failure(Command.EOF) =>
231+
cb(eofCondition())
232+
233+
case Failure(t) =>
234+
logger.error(t)("Unexpected error reading body.")
235+
cb(Either.left(t))
236+
}
237+
}
238+
} catch {
239+
case t: ParserException =>
240+
fatalError(t, "Error parsing request body")
241+
cb(Either.left(InvalidBodyException(t.getMessage())))
242+
243+
case t: Throwable =>
244+
fatalError(t, "Error collecting body")
245+
cb(Either.left(t))
234246
}
235-
} catch {
236-
case t: ParserException =>
237-
fatalError(t, "Error parsing request body")
238-
cb(Either.left(InvalidBodyException(t.getMessage())))
239-
240-
case t: Throwable =>
241-
fatalError(t, "Error collecting body")
242-
cb(Either.left(t))
243-
}
244-
go()
245-
} else cb(End)
247+
go()
248+
} else cb(End)
249+
shutdownCancelToken
250+
}
246251
}
247-
248252
(repeatEval(t).unNoneTerminate.flatMap(chunk(_)), () => drainBody(currentBuffer))
249253
}
250254

‎blaze-core/src/main/scala/org/http4s/blazecore/util/package.scala‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -42,6 +42,8 @@ package object util extends ParasiticExecutionContextCompat {
4242
case Some(value) =>
4343
F.fromTry(value)
4444
case None =>
45+
// Scala futures are uncancelable. There's not much we can
46+
// do here other than async_.
4547
F.async_ { cb =>
4648
future.onComplete {
4749
case Success(a) => cb(Right(a))

‎blaze-core/src/main/scala/org/http4s/blazecore/websocket/Http4sWSStage.scala‎

Lines changed: 20 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -66,23 +66,29 @@ private[http4s] class Http4sWSStage[F[_]](
6666

6767
def snkFun(frame: WebSocketFrame): F[Unit] = isClosed.ifM(F.unit, evalFrame(frame))
6868

69+
private[this] val shutdownCancelToken = Some(F.delay(stageShutdown()))
70+
6971
private[this] def writeFrame(frame: WebSocketFrame, ec: ExecutionContext): F[Unit] =
70-
writeSemaphore.permit.use { _ =>
71-
F.async_[Unit] { cb =>
72-
channelWrite(frame).onComplete {
73-
case Success(res) => cb(Right(res))
74-
case Failure(t) => cb(Left(t))
75-
}(ec)
76-
}
77-
}
72+
writeSemaphore.permit.surround(
73+
F.async[Unit](cb =>
74+
F.delay(
75+
channelWrite(frame).onComplete {
76+
case Success(res) => cb(Right(res))
77+
case Failure(t) => cb(Left(t))
78+
}(ec)
79+
).as(shutdownCancelToken)
80+
)
81+
)
7882

7983
private[this] def readFrameTrampoline: F[WebSocketFrame] =
80-
F.async_[WebSocketFrame] { cb =>
81-
channelRead().onComplete {
82-
case Success(ws) => cb(Right(ws))
83-
case Failure(exception) => cb(Left(exception))
84-
}(trampoline)
85-
}
84+
F.async[WebSocketFrame](cb =>
85+
F.delay(
86+
channelRead().onComplete {
87+
case Success(ws) => cb(Right(ws))
88+
case Failure(exception) => cb(Left(exception))
89+
}(trampoline)
90+
).as(shutdownCancelToken)
91+
)
8692

8793
/** Read from our websocket.
8894
*

‎blaze-server/src/main/scala/org/http4s/blaze/server/Http2NodeStage.scala‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -105,6 +105,8 @@ private class Http2NodeStage[F[_]](
105105
closePipeline(Some(e))
106106
}
107107

108+
private[this] val shutdownStageToken = Some(F.delay(stageShutdown()))
109+
108110
/** collect the body: a maxlen < 0 is interpreted as undefined */
109111
private def getBody(maxlen: Long): EntityBody[F] = {
110112
var complete = false
@@ -156,7 +158,7 @@ private class Http2NodeStage[F[_]](
156158
closePipeline(Some(e))
157159
}
158160

159-
None
161+
shutdownStageToken
160162
}
161163
}
162164

‎blaze-server/src/test/scala/org/http4s/blaze/server/Http1ServerStageSpec.scala‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -537,7 +537,7 @@ class Http1ServerStageSpec extends CatsEffectSuite {
537537
}
538538
}
539539

540-
fixture.test("Http1ServerStage: routes should cancels on stage shutdown".flaky) { tw =>
540+
fixture.test("Http1ServerStage: routes should cancel on stage shutdown".flaky) { tw =>
541541
Deferred[IO, Unit]
542542
.flatMap { canceled =>
543543
Deferred[IO, Unit].flatMap { gate =>

‎build.sbt‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,7 @@ import Dependencies._
44
val Scala212 = "2.12.17"
55
val Scala213 = "2.13.10"
66
val Scala3 = "3.2.2"
7-
val http4sVersion = "0.23.18"
7+
val http4sVersion = "0.23.19"
88
val munitCatsEffectVersion = "2.0.0-M3"
99

1010
ThisBuild / resolvers +=

‎examples/src/main/scala/com/example/http4s/blaze/demo/server/Module.scala‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,8 @@ import com.example.http4s.blaze.demo.server.endpoints.auth.BasicAuthHttpEndpoint
2424
import com.example.http4s.blaze.demo.server.endpoints.auth.GitHubHttpEndpoint
2525
import com.example.http4s.blaze.demo.server.service.FileService
2626
import com.example.http4s.blaze.demo.server.service.GitHubService
27+
import fs2.compression.Compression
28+
import fs2.io.file.Files
2729
import org.http4s.HttpRoutes
2830
import org.http4s.client.Client
2931
import org.http4s.server.HttpMiddleware
@@ -34,7 +36,7 @@ import org.http4s.server.middleware.Timeout
3436

3537
import scala.concurrent.duration._
3638

37-
class Module[F[_]: Async](client: Client[F]) {
39+
class Module[F[_]: Async: Compression: Files](client: Client[F]) {
3840
private val fileService = new FileService[F]
3941

4042
private val gitHubService = new GitHubService[F](client)

0 commit comments

Comments
 (0)