NonBlockingServletIo request body truncated to one chunk

[dwalend]

NonBlockingServletIo request bodies are truncated to just one chunk. (With Tomcat 8.0.39, JDK 1.8.0_121-b13 , scala 2.11.12 , and http4s 0.20 M4.)

Work-around is to use BlockingServletIo, which does not have this problem.

This could be related to http4s/http4s#1724 or http4s/http4s#541 .

The server is

package net.shrine.testapp

import cats.effect.{ConcurrentEffect, ContextShift, IO}
import javax.servlet.annotation.WebListener
import javax.servlet.{ServletContext, ServletContextEvent, ServletContextListener, ServletRegistration}
import org.http4s.server.DefaultServiceErrorHandler
import org.http4s.servlet.{AsyncHttp4sServlet, BlockingServletIo, NonBlockingServletIo}

import scala.concurrent.duration.Duration
import org.http4s.HttpRoutes
import org.http4s.dsl.impl.{->, /}
import org.http4s.dsl.io.{GET, Ok, PUT, Root, http4sOkSyntax}

import scala.language.higherKinds
import scala.concurrent.ExecutionContext.Implicits.global

/**
  * A simple servlet to bracket problems in Shrine without bringing in all of Shrine's complexity
  */

@WebListener
class TestAppBootstrap extends ServletContextListener {

  override def contextInitialized(sce: ServletContextEvent): Unit = {
    val context = sce.getServletContext

    implicit val contextShift: ContextShift[IO] = IO.contextShift(global)
    implicit val concurrentEffect: ConcurrentEffect[IO] = IO.ioConcurrentEffect(contextShift)

    mountService(
      context = context,
      name = "test http4s",
      service = service,
      mapping = "/*"
    )
  }

  def mountService[F[_]: ConcurrentEffect: ContextShift](
                    context: ServletContext,
                    name: String,
                    service: HttpRoutes[F],
                    mapping: String): ServletRegistration.Dynamic = {

    val servlet = new AsyncHttp4sServlet(
      service = service,
      asyncTimeout = Duration("30 seconds"),
      servletIo = NonBlockingServletIo(4096), //limited to 4096 bytes per request
//      servletIo = BlockingServletIo(4096,global), //works
      serviceErrorHandler = DefaultServiceErrorHandler[F]
    )
    val reg = context.addServlet(name, servlet)
    reg.setLoadOnStartup(1)
    reg.setAsyncSupported(true)
    reg.addMapping(mapping)
    reg
  }

  val service: HttpRoutes[IO] = HttpRoutes.of[IO] {

    case _@GET -> Root / "ping" => Ok("pong")

    case sizeReq@PUT -> Root / "size" =>
      val messageTextIO: IO[String] = sizeReq.as[String]
      messageTextIO.flatMap{text => Ok(s"Received ${text.length}")}
  }

  override def contextDestroyed(sce: ServletContextEvent): Unit = {}
}

The client is

import cats.effect.{ContextShift, IO, Resource}
import io.netty.handler.ssl.{SslContext, SslContextBuilder}
import io.netty.handler.ssl.util.InsecureTrustManagerFactory
import org.asynchttpclient.{DefaultAsyncHttpClientConfig, Dsl, Realm}
import org.asynchttpclient.proxy.ProxyServer
import org.http4s.client.Client
import org.http4s.client.asynchttpclient.AsyncHttpClient
import org.http4s.{EntityDecoder, Method, Request, Response, Status, Uri}

import scala.concurrent.ExecutionContext.Implicits.global

val serverUri = Uri.unsafeFromString("https://shrine-dev1.catalyst:6443/test/size")

val httpClientBuilder: DefaultAsyncHttpClientConfig.Builder = {

  //todo use a more particular trust manager
  val sslContext: SslContext = SslContextBuilder.forClient().trustManager(InsecureTrustManagerFactory.INSTANCE).build()
  val sslConfigBuilder: DefaultAsyncHttpClientConfig.Builder = Dsl.config().setSslContext(sslContext)

  sslConfigBuilder
}

val httpClientResource: Resource[IO, Client[IO]] = {
  implicit val concurrentEffect: ContextShift[IO] = IO.contextShift(global)
  AsyncHttpClient.resource[IO](httpClientBuilder.build)
}

def webFetchAndDecodeIO[A](request:Request[IO])(toA: (Status,String) => IO[A]): IO[A] = {
  httpClientResource.use { client: Client[IO] =>
    client.fetch(request) { response: Response[IO] =>
      EntityDecoder.decodeString(response).flatMap(toA(response.status, _))
    }
  }
}

def createRequest(contents:String): Request[IO] = Request(
  method = Method.PUT,
  uri = serverUri
).withEntity(contents)

val iters = 14
val contents = (1 to iters).map{ i =>
  val number = scala.math.pow(2,i).toInt -1 //-1 for the end-of-line to make awk play nice
  s"$i $number ${"*".*(number)}\n"
}

contents.foreach{ content =>
  val request = createRequest(content)
  webFetchAndDecodeIO(request){(status,string) =>
    println(s"Sent ${content.length}; $string")
    IO{s"$status $string"}
  }.unsafeRunSync()
}

Let me know if you need more odd parts like the web.xml to reproduce the problem .

[dwalend]

The other bit of note is that calling the server via curl didn't have this problem. I believe that's because curl works with 16k chunks by default, but stopped exploring when I found the work-around.

[EnoughTea]

This is still actual for v0.21.2 and servlets deployed on both Tomcat 8 and 9.

[rossabaker]

http4s/http4s#4762 would seem like it would expose this. If anyone is still seeing this in the wild, please speak up, and we'll reopen.

[dwalend]

@rossabaker - I'm seeing this with http4s-servlet 0.23.15 . (I finally got to upgrade to the latest cats!)

The work-around involved bumbling through a deprecated constructor insetad of using the builder:

      val servlet = new AsyncHttp4sServlet(
        httpApp = service,
        asyncTimeout = ConfigSource.config.get("shrine.api.asyncTimeout", Duration(_)),
        servletIo = BlockingServletIo[IO](4096),
        serviceErrorHandler = DefaultServiceErrorHandler[IO],
        dispatcher = dispatcher,
      )
      val registration = context.addServlet(name, servlet)
      registration.setLoadOnStartup(1)
      registration.setAsyncSupported(true)
      registration.addMapping(mapping)

      registration

[danicheg]

This project has been migrated into a dedicated repository — https://github.com/http4s/http4s-servlet.
So, I'm going to transfer this issue there.

UPD: uh oh, I'm lacking access to that repo, so someone with needed rights, please proceed further.