Merge pull request #149 from fbascheper/feature/upgrade-AHC-2.0.32

Upgrade to AsyncHttpClient 2.0.32

Author: farmdawgnation

README.markdown

@@ -11,6 +11,10 @@ as its underlying transport. For more info, see the
 [docs]: http://dispatch.databinder.net/Dispatch.html
 [async]: https://github.com/AsyncHttpClient/async-http-client
 
+### Dependencies
+* [Async HTTP client](https://github.com/AsyncHttpClient/async-http-client)
+* JDK 8, as required by the Async HTTP client library
+
 ### Mailing List
 
 There's a [mailing list for Dispatch][mail]. Please mail the list **before opening

build.sbt

@@ -48,9 +48,7 @@ lazy val tagsoup = module("tagsoup")
 lazy val xmlDependency = libraryDependencies ++= Seq("org.scala-lang.modules" %% "scala-xml" % "1.0.6")
 
 /** Util module for using unfiltered with scalacheck */
-lazy val ufcheck = Project(
-  "ufcheck", file("ufcheck")
-).settings(
+lazy val ufcheck = project.in(file("ufcheck")).settings(
   scalaVersion := Common.defaultScalaVersion
 )
 

core/build.sbt

@@ -1,10 +1,10 @@
 name := "dispatch-core"
 
 description :=
-  "Core Dispatch module wrapping sonatype/async-http-client"
+  "Core Dispatch module wrapping async-http-client"
 
 libraryDependencies +=
-  "com.ning" % "async-http-client" % "1.9.11"
+  "org.asynchttpclient" % "async-http-client" % "2.0.32"
 
 Seq(lsSettings :_*)
 

core/src/main/scala/as/core.scala

@@ -2,36 +2,36 @@ package dispatch.as
 
 import dispatch._
 
-import com.ning.http.client
 import java.nio.charset.Charset
+import org.asynchttpclient
 
 object Response {
-  def apply[T](f: client.Response => T) = f
+  def apply[T](f: asynchttpclient.Response => T) = f
 }
 
-object String extends (client.Response => String) {
+object String extends (asynchttpclient.Response => String) {
   /** @return response body as a string decoded as either the charset provided by
    *  Content-Type header of the response or ISO-8859-1 */
-  def apply(r: client.Response) = r.getResponseBody
+  def apply(r: asynchttpclient.Response) = r.getResponseBody
 
   /** @return a function that will return response body decoded in the provided charset */
-  case class charset(set: Charset) extends (client.Response => String) {
-    def apply(r: client.Response) = r.getResponseBody(set.name)
+  case class charset(set: Charset) extends (asynchttpclient.Response => String) {
+    def apply(r: asynchttpclient.Response) = r.getResponseBody(set)
   }
 
   /** @return a function that will return response body as a utf8 decoded string */
   object utf8 extends charset(Charset.forName("utf8"))
 }
 
-object Bytes extends (client.Response => Array[Byte]) {
-  def apply(r: client.Response) = r.getResponseBodyAsBytes
+object Bytes extends (asynchttpclient.Response => Array[Byte]) {
+  def apply(r: asynchttpclient.Response) = r.getResponseBodyAsBytes
 }
 
 object File extends {
   def apply(file: java.io.File) =
-    (new client.resumable.ResumableAsyncHandler with OkHandler[Nothing])
+    (new asynchttpclient.handler.resumable.ResumableAsyncHandler with OkHandler[asynchttpclient.Response])
       .setResumableListener(
-        new client.extra.ResumableRandomAccessFileListener(
+        new asynchttpclient.handler.resumable.ResumableRandomAccessFileListener(
           new java.io.RandomAccessFile(file, "rw")
         )
       )

core/src/main/scala/as/oauth/token.scala

@@ -1,8 +1,7 @@
 package dispatch.as.oauth
 
-import dispatch.oauth._
-import com.ning.http.client.Response
-import com.ning.http.client.oauth._
+import org.asynchttpclient.Response
+import org.asynchttpclient.oauth.RequestToken
 
 object Token extends (Response => Either[String, RequestToken]) {
   def apply(res: Response) = tokenDecode(dispatch.as.String(res))

core/src/main/scala/as/stream/lines.scala

@@ -2,8 +2,6 @@ package dispatch.as.stream
 
 import dispatch._
 
-import com.ning.http.client._
-
 object Lines {
   def apply[U](f: String => U) =
     new stream.StringsByLine[Unit] {

core/src/main/scala/defaults.scala

@@ -1,108 +1,35 @@
 package dispatch
 
-import org.jboss.netty.util.{Timer, HashedWheelTimer}
-import org.jboss.netty.channel.socket.nio.{
-  NioClientSocketChannelFactory, NioWorkerPool}
-import java.util.{concurrent => juc}
-import com.ning.http.client.{
-  AsyncHttpClient, AsyncHttpClientConfig
-}
-import com.ning.http.client.providers.netty.NettyAsyncHttpProviderConfig
+import io.netty.util.{HashedWheelTimer, Timer}
+import org.asynchttpclient.DefaultAsyncHttpClientConfig.Builder
+import org.asynchttpclient._
 
 object Defaults {
   implicit def executor = scala.concurrent.ExecutionContext.Implicits.global
+
   implicit lazy val timer: Timer = InternalDefaults.timer
 }
 
-private [dispatch] object InternalDefaults {
-  /** true if we think we're runing un-forked in an sbt-interactive session */
-  val inSbt = (
-    for (group <- Option(Thread.currentThread.getThreadGroup))
-    yield (
-      group.getName == "trap.exit" // sbt version <= 0.13.0
-      || group.getName.startsWith("run-main-group") // sbt 0.13.1+
-    )
-  ).getOrElse(false)
-
-  private lazy val underlying = 
-    if (inSbt) SbtProcessDefaults
-    else BasicDefaults
+private[dispatch] object InternalDefaults {
+  private lazy val underlying = BasicDefaults
 
-  def client = new AsyncHttpClient(underlying.builder.build())
-  lazy val timer = underlying.timer
+  lazy val clientBuilder: Builder = underlying.builder
+  lazy val timer: Timer = underlying.timer
 
   private trait Defaults {
-    def builder: AsyncHttpClientConfig.Builder
+    def builder: DefaultAsyncHttpClientConfig.Builder
+
     def timer: Timer
   }
 
   /** Sets a user agent, no timeout for requests  */
   private object BasicDefaults extends Defaults {
     lazy val timer = new HashedWheelTimer()
-    def builder = new AsyncHttpClientConfig.Builder()
+
+    def builder: Builder = new DefaultAsyncHttpClientConfig.Builder()
       .setUserAgent("Dispatch/%s" format BuildInfo.version)
       .setRequestTimeout(-1) // don't timeout streaming connections
       .setUseProxyProperties(true)
   }
 
-  /** Uses daemon threads and tries to exit cleanly when running in sbt  */
-  private object SbtProcessDefaults extends Defaults {
-    def builder = {
-      val shuttingDown = new juc.atomic.AtomicBoolean(false)
-
-      def shutdown(): Unit = {
-        if (shuttingDown.compareAndSet(false, true)) {
-          nioClientSocketChannelFactory.releaseExternalResources()
-          timer.stop()
-        }
-        ()
-      }
-      /** daemon threads that also shut down everything when interrupted! */
-      lazy val interruptThreadFactory = new juc.ThreadFactory {
-        def newThread(runnable: Runnable) = {
-          new Thread(runnable) {
-            setDaemon(true)
-            /** only reliably called on any thread if all spawned threads are daemon */
-            override def interrupt() = {
-              shutdown()
-              super.interrupt()
-            }
-          }
-        }
-      }
-      lazy val nioClientSocketChannelFactory = {
-        val workerCount = 2 * Runtime.getRuntime().availableProcessors()
-        new NioClientSocketChannelFactory(
-          juc.Executors.newCachedThreadPool(interruptThreadFactory),
-          1,
-          new NioWorkerPool(
-            juc.Executors.newCachedThreadPool(interruptThreadFactory),
-            workerCount
-          ),
-          timer
-        )
-      }
-
-      val config = new NettyAsyncHttpProviderConfig().addProperty(
-        "socketChannelFactory",
-        nioClientSocketChannelFactory
-      )
-      config.setNettyTimer(timer)
-      BasicDefaults.builder.setAsyncHttpClientProviderConfig(config)
-    }
-    lazy val timer = new HashedWheelTimer(DaemonThreads.factory)
-  }
-}
-
-object DaemonThreads {
-  /** produces daemon threads that won't block JVM shutdown */
-  val factory = new juc.ThreadFactory {
-    def newThread(runnable: Runnable): Thread ={
-      val thread = new Thread(runnable)
-      thread.setDaemon(true)
-      thread
-    }
-  }
-  def apply(threadPoolSize: Int) =
-    juc.Executors.newFixedThreadPool(threadPoolSize, factory)
 }

core/src/main/scala/enrich.scala

@@ -1,11 +1,7 @@
 package dispatch
 
-import com.ning.http.client.ListenableFuture
-import java.util.{concurrent => juc}
-import juc.TimeUnit
 import scala.concurrent.{ExecutionContext,Await,ExecutionException}
 import scala.concurrent.duration.Duration
-import scala.util.control.Exception.{allCatch,catching}
 
 class EnrichedFuture[A](underlying: Future[A]) {
 

core/src/main/scala/execution.scala

@@ -1,36 +1,39 @@
 package dispatch
 
-import com.ning.http.client.{
-  AsyncHttpClient, Request, Response, AsyncHandler,
-  AsyncHttpClientConfig
-}
-
 import java.util.{concurrent => juc}
-import scala.concurrent.{ExecutionContext}
+
+import org.asynchttpclient._
+
+import scala.concurrent.ExecutionContext
+import scala.util.Try
 
 /** Http executor with defaults */
 case class Http(
-  client: AsyncHttpClient = InternalDefaults.client
-) extends HttpExecutor {
-  import AsyncHttpClientConfig.Builder
+                 clientBuilder: DefaultAsyncHttpClientConfig.Builder = InternalDefaults.clientBuilder
+               ) extends HttpExecutor {
 
-  /** Replaces `client` with a new instance configured using the withBuilder
-      function. The current client config is the builder's prototype.  */
-  def configure(withBuilder: Builder => Builder) =
-    copy(client =
-      new AsyncHttpClient(withBuilder(
-        new AsyncHttpClientConfig.Builder(client.getConfig)
-      ).build)
-    )
+  import DefaultAsyncHttpClientConfig.Builder
+
+  lazy val client = new DefaultAsyncHttpClient(clientBuilder.build)
+
+  /*§*
+    * Replaces `clientBuilder` with a new config built using the withBuilder function.
+    * The current client config is the builder's prototype.
+    */
+  def configure(withBuilder: Builder => Builder): Http = {
+    val newBuilder = new Builder(this.clientBuilder.build)
+    copy(clientBuilder = withBuilder(newBuilder))
+  }
 }
 
 /** Singleton default Http executor, can be used directly or altered
- *  with its case-class `copy` */
+  * with its case-class `copy` */
 object Http extends Http(
-  InternalDefaults.client
+  InternalDefaults.clientBuilder
 )
 
-trait HttpExecutor { self =>
+trait HttpExecutor {
+  self =>
   def client: AsyncHttpClient
 
   def apply(req: Req)
@@ -42,12 +45,12 @@ trait HttpExecutor { self =>
     apply(pair._1, pair._2)
 
   def apply[T]
-    (request: Request, handler: AsyncHandler[T])
-    (implicit executor: ExecutionContext): Future[T] = {
+  (request: Request, handler: AsyncHandler[T])
+  (implicit executor: ExecutionContext): Future[T] = {
     val lfut = client.executeRequest(request, handler)
     val promise = scala.concurrent.Promise[T]()
     lfut.addListener(
-      () => promise.complete(util.Try(lfut.get())),
+      () => promise.complete(Try(lfut.get())),
       new juc.Executor {
         def execute(runnable: Runnable) = {
           executor.execute(runnable)

core/src/main/scala/handlers.scala

@@ -1,7 +1,6 @@
 package dispatch
 
-import com.ning.http.client
-import client.{
+import org.asynchttpclient.{
   Response, AsyncCompletionHandler, AsyncHandler,
   HttpResponseStatus
 }