scala - 使用 Netty 和 Scala Actor 的异步 http 请求

Question

使用 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 以发出请求。

使用 ChannelFuture 阻塞调用 actor 对象。写入 channel 返回的对象。当messageRecieved我的处理程序类的函数被称为我将 netty http 请求的响应解析为字符串，将带有响应内容的消息发送回 actor 并关闭 channel 。

future 完成后，我的代码尝试使用收到的 http 内容响应向调用actor 发送回复。

代码有效，我能够得到回复，将其发送到我的 Actor 实例，打印出内容并向正在使用的 Actor 实例发布资源发送消息。

问题是当我测试它时，对 Actor 的原始调用没有得到回复，线程只是保持打开状态。

代码示例 - HttpRequestActor

我的代码 HttpRequestActor类(class)

    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);}

    }
  }

使用的库:
- 斯卡拉 2.9.2
- Netty.io 3.6.1.Final
- Junit 4.7
- 升级版 1.8
- 我也在用 @viktorklang NettyFutureBridge 对象 gist为返回的 Channel 对象创建一个 scala future

如何使用来自 Netty 的响应内容将回复发送回 actor 对象并结束线程？

任何帮助都感激不尽

Answer 1

我不知道 Scala，但我有类似的问题。尝试指定响应的内容长度 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 中有一个库可以这样做)。