使用Netty和Scala actor的异步http请求

嘿,希望有人可以帮我这个忙。

我正在尝试使用Scala Actors和Netty.io库来获取异步http请求。 (是的,我知道Scala Actor 已被弃用,但这对我来说是一个学习练习)

我已经编写了一个 Actor HttpRequestActor,它接受案例类RequestPage(uri:URI)形式的消息。

当它收到消息时,它创建了发出http请求所需的必要Netty对象,我基于[HttpSnoopClient](http://static.netty.io/3.5/xref/org/jboss/netty/example/http/snoop/HttpSnoopClient.html)示例。

我创建了一个客户端,并将当前的actor实例传递给ChannelPipelineFactory的实现,该实例也将actor传递给了我的SimpleChannelUpstreamHandler的实现,在此我重写了messageReceived函数。

actor实例作为监听器传递,我使用DefaultHttpRequest类创建一个请求,并将其写入 channel 以进行请求。

使用从写入 channel 返回的ChannelFuture对象对actor对象进行了阻塞调用。调用我的处理程序类的messageRecieved函数时,我将netty http请求的响应解析为字符串,将响应的内容发送回actor并关闭 channel 。

将来完成后,我的代码尝试将收到的http内容响应发送给主叫 Actor 。

该代码有效,我能够得到答复,将其发送到我的actor实例,打印出内容,并向该actor实例发送一条消息,释放正在使用的资源。

问题是当我对其进行测试时,对actor的原始调用没有得到答复,并且该线程只是保持打开状态。

代码示例-HttpRequestActor

我的HttpRequestActor类的代码

    import scala.actors.Actor
import java.net.{InetSocketAddress,URI}
import org.jboss.netty.handler.codec.http._
import org.jboss.netty.bootstrap.ClientBootstrap
import org.jboss.netty.channel.Channel
import org.jboss.netty.channel._
import org.jboss.netty.channel.socket.nio.NioClientSocketChannelFactory
import org.jboss.netty.channel.group.DefaultChannelGroup
import java.util.concurrent.{Executors,CancellationException}
import org.jboss.netty.util.CharsetUtil
import scala.concurrent.{ Promise, Future }
import scala.concurrent.ExecutionContext.Implicits.global

/**
 * @author mebinum
 *
 */
class HttpRequestActor extends Actor {
    //initialize response with default uninitialized value
    private var resp:Response = _
    private val executor = Executors.newCachedThreadPool
    private val executor2 = Executors.newCachedThreadPool
    private val factory = new NioClientSocketChannelFactory(
                          executor,
                          executor2);

    private val allChannels = new DefaultChannelGroup("httpRequester")

    def act = loop {
        react {
            case RequestPage(uri) => requestUri(uri)
            case Reply(msg) => setResponse(Reply(msg))
            case NoReply => println("didnt get a reply");setResponse(NoReply)
            case NotReadable => println("got a reply but its not readable");setResponse(NotReadable)
            case ShutDown => shutDown()
        }
    }

    private def requestUri(uri:URI) = {

      makeChannel(uri) map {
          channel => {
              allChannels.add(channel)
              val request = new DefaultHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.GET, uri.toString)
              request.setHeader(HttpHeaders.Names.HOST, uri.getHost())
              request.setHeader(HttpHeaders.Names.CONNECTION, HttpHeaders.Values.CLOSE)
              request.setHeader(HttpHeaders.Names.ACCEPT_ENCODING, HttpHeaders.Values.GZIP)

              val writeFuture = channel.write(request).awaitUninterruptibly()

              FutureReactor !? writeFuture match {
                  case future : ChannelFuture => {
                      future.addListener(new ChannelFutureListener() {
                          def operationComplete(future:ChannelFuture) {
                              // Perform post-closure operation
                              println("current response is " + resp)
                              sendResponse("look ma I finished")
                          }
                      })
                      future.getChannel().close()
                  }
              }

              this ! ShutDown
          }
      }
      //thread ends only if you send a reply from here
      //println("this is final sender " + sender)
      //reply("I am the true end")
    }

    private def makeChannel(uri:URI) = {
      val scheme = Some(uri.getScheme()).getOrElse("http")
      val host = Some(uri.getHost()).getOrElse("localhost")

      val port = Utils.getPort(uri.getPort, uri.getScheme)

      // Set up the event pipeline factory.
      val client = new ClientBootstrap(factory)
      client.setPipelineFactory(new PipelineFactory(this))

      //get the promised channel
      val channel = NettyFutureBridge(client.connect(new InetSocketAddress(host, port)))
      channel
    }

    private def setResponse(aResponse:Response) = resp = aResponse

    private def sendResponse(msg:String) = {
      println("Sending the response " + msg)
      reply(resp)
    }

    private def shutDown() = {
        println("got a shutdown message")
        val groupFuture = allChannels.close().awaitUninterruptibly()
        factory.releaseExternalResources()
    }

    override def exceptionHandler = {
      case e : CancellationException => println("The request was cancelled"); throw e
      case tr: Throwable => println("An unknown exception happened " + tr.getCause()); throw tr
    }
}



trait Response
case class RequestPage(url:URI)

case class Reply(content:String) extends Response
case object NoReply extends Response
case object NotReadable extends Response
case object ShutDown

object FutureReactor extends Actor{
  def act = //loop {
      react {
        case future: ChannelFuture => {
            if (future.isCancelled) {
                throw new CancellationException()
            }
            if (!future.isSuccess()) {
                future.getCause().printStackTrace()
                throw future.getCause()
            }
            if(future.isSuccess() && future.isDone()){
                future.getChannel().getCloseFuture().awaitUninterruptibly()
                reply(future)
            }
        }
      }
    //}
  this.start
}


class ClientHandler(listener:Actor) extends SimpleChannelUpstreamHandler {

  override def exceptionCaught( ctx:ChannelHandlerContext, e:ExceptionEvent){
    e.getCause().printStackTrace()
    e.getChannel().close();
    throw e.getCause()
  }

  override def messageReceived(ctx:ChannelHandlerContext,  e:MessageEvent) = {
        var contentString = ""
        var httpResponse:Response =  null.asInstanceOf[Response]

        e.getMessage match {
          case (response: HttpResponse) if !response.isChunked => {
              println("STATUS: " + response.getStatus);
              println("VERSION: " + response.getProtocolVersion);
              println

              val content = response.getContent();
              if (content.readable()) {
                  contentString = content.toString(CharsetUtil.UTF_8)
                  httpResponse = Reply(contentString)
                  //notify actor

              }else{
                 httpResponse = NotReadable
              }
          }
          case chunk: HttpChunk if !chunk.isLast => {
            //get chunked content
            contentString = chunk.getContent().toString(CharsetUtil.UTF_8)
            httpResponse = Reply(contentString)
          }
          case _ => httpResponse = NoReply
        }
         println("sending actor my response")
         listener ! httpResponse
         println("closing the channel")
         e.getChannel().close()
         //send the close event

    }


}


class PipelineFactory(listener:Actor) extends ChannelPipelineFactory {

    def  getPipeline(): ChannelPipeline = {
            // Create a default pipeline implementation.
            val pipeline = org.jboss.netty.channel.Channels.pipeline()

            pipeline.addLast("codec", new HttpClientCodec())

            // Remove the following line if you don't want automatic content decompression.
            pipeline.addLast("inflater", new HttpContentDecompressor())

            // Uncomment the following line if you don't want to handle HttpChunks.
            //pipeline.addLast("aggregator", new HttpChunkAggregator(1048576))

            pipeline.addLast("decoder", new HttpRequestDecoder())
            //assign the handler
            pipeline.addLast("handler", new ClientHandler(listener))

            pipeline;
    }
}


object NettyFutureBridge {
  import scala.concurrent.{ Promise, Future }
  import scala.util.Try
  import java.util.concurrent.CancellationException
  import org.jboss.netty.channel.{ Channel, ChannelFuture, ChannelFutureListener }

  def apply(nettyFuture: ChannelFuture): Future[Channel] = {
    val p = Promise[Channel]()
    nettyFuture.addListener(new ChannelFutureListener {
      def operationComplete(future: ChannelFuture): Unit = p complete Try(
        if (future.isSuccess) {
          println("Success")
          future.getChannel
        }
        else if (future.isCancelled) {
          println("Was cancelled")
          throw new CancellationException
        }

        else {
          future.getCause.printStackTrace()
          throw future.getCause
        })
    })
    p.future
  }
}

代码进行测试
val url = "http://hiverides.com"

test("Http Request Actor can recieve and react to message"){
    val actor = new HttpRequestActor()
    actor.start

    val response = actor !? new RequestPage(new URI(url))
    match {
      case Reply(msg) => {
          println("this is the reply response in test")
          assert(msg != "")
          println(msg)
        }
      case NoReply => println("Got No Reply")
      case NotReadable => println("Got a not Reachable")
      case None => println("Got a timeout")
      case s:Response => println("response string \n" + s)
      case x => {println("Got a value not sure what it is"); println(x);}

    }
  }

使用的库:
-Scala 2.9.2
-Netty.io 3.6.1。最终版
-Junit 4.7
-最严格的1.8
-我还使用@viktorklang NettyFutureBridge对象gist为返回的Channel对象创建一个scala Future

如何将包含Netty响应内容的答复发送回actor对象并结束线程?

任何帮助都感激不尽

最佳答案

我不了解Scala,但我遇到了类似的问题。尝试指定响应的content-length header 。

在纯Java中:

HttpRequest r = new DefaultHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.POST, uri);
            ChannelBuffer buffer = ChannelBuffers.copiedBuffer(input);
            r.setHeader(HttpHeaders.Names.HOST, "host");
            r.setHeader(HttpHeaders.Names.CONTENT_TYPE, "application/octet-stream");
            r.setHeader(HttpHeaders.Names.CONTENT_LENGTH, buffer.readableBytes());
            r.setHeader(HttpHeaders.Names.CONNECTION, HttpHeaders.Values.CLOSE);
            r.setContent(buffer);

否则,除非客户端关闭连接,否则服务器不知道客户端何时完成内容。

您也可以使用分块编码,但是您必须自己实现分块编码(至少我不知道Netty中的库会这样做)。

09-16 04:06