如何在scalaz中关闭AsyncHttpClient任务

Question

我正在尝试将AsyncHttpClient和Scalaz Task结合在一起。通常，如果我使用AsyncHttpClient，我可以调用client.close来停止客户端。

val asyncHttpClient = new AsyncHttpClient() 
println(asyncHttpClient.prepareGet("http://www.google.com")) 

asyncHttpClient.close() 

因此电流将停止。但是，如果我将api调用包装到Task中。我不知道如何阻止它。

def get(s: String) = Task.async[Int](k => { 
    asyncHttpClient.prepareGet(s).execute(toHandler) 
    Thread.sleep(5000) 
    asyncHttpClient.closeAsynchronously() 
    }) 

    def toHandler[A] = new AsyncCompletionHandler[Response] { 
    def onCompleted(r: Response) = { 
     println("get response ", r.getResponseBody) 
     r 
    } 
    def onError(e: Throwable) = { 
     println("some error") 
     e 
    } 
    } 

println(get("http://www.google.com").run) 

当前Process仍在运行。我在想，原因是Task和AsynClient都是异步的。我不知道我应该做些什么来关闭它

很多感谢

Answer 1

的问题是，Task.async需要，可以注册一个回调函数。这是一个有点混乱，和类型没有太大的帮助，因为那里有这么多该死的Unit，但它的意思在这里是你想要的东西更是这样的：

import com.ning.http.client._ 
import scalaz.syntax.either._ 
import scalaz.concurrent.Task 

val asyncHttpClient = new AsyncHttpClient() 

def get(s: String): Task[Response] = Task.async[Response](callback => 
    asyncHttpClient.prepareGet(s).execute(
    new AsyncCompletionHandler[Unit] { 
     def onCompleted(r: Response): Unit = callback(r.right) 
     def onError(e: Throwable): Unit = callback(e.left) 
    } 
) 
) 

这不处理闭幕客户 - 它只是为了展示大致的想法。你可以关闭客户端的处理程序，但我建议更多的东西是这样的：

import com.ning.http.client._ 
import scalaz.syntax.either._ 
import scalaz.concurrent.Task 

def get(client: AsyncHttpClient)(s: String): Task[Response] = 
    Task.async[Response](callback => 
    client.prepareGet(s).execute(
     new AsyncCompletionHandler[Unit] { 
     def onCompleted(r: Response): Unit = callback(r.right) 
     def onError(e: Throwable): Unit = callback(e.left) 
     } 
    ) 
) 

def initClient: Task[AsyncHttpClient] = Task(new AsyncHttpClient()) 
def closeClient(client: AsyncHttpClient): Task[Unit] = Task(client.close()) 

然后：

val res = for { 
    c <- initClient 
    r <- get(c)("http://www.google.com") 
    _ <- closeClient(c) 
} yield r 

res.unsafePerformAsync(
    _.fold(
    _ => println("some error"), 
    r => println("get response " + r.getResponseBody) 
) 
) 

这避免closeAsynchronously（这似乎是going away，反正）。