0
我試圖用Spray/Akka實現反向HTTP代理，但遇到了麻煩。我發現在某些情況下，即使在客戶端斷開連接之後，我的代理服務器仍會繼續從上游服務器接收數據。Spray反向代理：在客戶端斷開連接後繼續傳輸數據
以下是我實現我的噴霧代理指令（只是一點點修改bthuillier's implementation）：
trait ProxyDirectives {
private def sending(f: RequestContext ⇒ HttpRequest)(implicit system: ActorSystem): Route = {
val transport = IO(Http)(system)
ctx ⇒ transport.tell(f(ctx), ctx.responder)
}
/**
* Re-shape the original request, to match the destination server.
*/
private def reShapeRequest(req: HttpRequest, uri: Uri): HttpRequest = {
req.copy(
uri = uri,
headers = req.headers.map {
case x: HttpHeaders.Host => HttpHeaders.Host(uri.authority.host.address, uri.authority.port)
case x => x
}
)
}
/**
* proxy the request to the specified uri
*
*/
def proxyTo(uri: Uri)(implicit system: ActorSystem): Route = {
sending(ctx => reShapeRequest(ctx.request, uri))
}
}
這個反向代理將工作做好，如果我把客戶端和服務器之間的一個代理層（即，客戶端< - > proxyTo < - >服務器），但是如果我在客戶端和服務器之間放置兩層，它將會遇到麻煩。舉例來說，如果我有以下簡單的Python HTTP服務器：
import socket
from threading import Thread, Semaphore
import time
from BaseHTTPServer import BaseHTTPRequestHandler, HTTPServer
from SocketServer import ThreadingMixIn
class MyHTTPHandler(BaseHTTPRequestHandler):
protocol_version = 'HTTP/1.1'
def do_GET(self):
self.send_response(200)
self.send_header('Transfer-Encoding', 'chunked')
self.end_headers()
for i in range(100):
data = ('%s\n' % i).encode('utf-8')
self.wfile.write(hex(len(data))[2:].encode('utf-8'))
self.wfile.write(b'\r\n')
self.wfile.write(data)
self.wfile.write(b'\r\n')
time.sleep(1)
self.wfile.write(b'0\r\n\r\n')
class MyServer(ThreadingMixIn, HTTPServer):
def server_bind(self):
HTTPServer.server_bind(self)
self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
def server_close(self):
HTTPServer.server_close(self)
if __name__ == '__main__':
server = MyServer(('127.0.0.1', 8080), MyHTTPHandler)
server.serve_forever()
這基本上什麼也不做，但打開一個分塊響應（長期運行，這樣我們就可以考試的問題）。如果我連鎖代理的兩層以下方式：
class TestActor(val target: String)(implicit val system: ActorSystem) extends Actor
with HttpService
with ProxyDirectives
{
// we use the enclosing ActorContext's or ActorSystem's dispatcher for our Futures and Scheduler
implicit private def executionContext = actorRefFactory.dispatcher
// the HttpService trait defines only one abstract member, which
// connects the services environment to the enclosing actor or test
def actorRefFactory = context
val serviceRoute: Route = {
get {
proxyTo(target)
}
}
// runs the service routes.
def receive = runRoute(serviceRoute) orElse handleTimeouts
private def handleTimeouts: Receive = {
case Timedout(x: HttpRequest) =>
sender ! HttpResponse(StatusCodes.InternalServerError, "Request timed out.")
}
}
object DebugMain extends App {
val actorName = "TestActor"
implicit val system = ActorSystem(actorName)
// create and start our service actor
val service = system.actorOf(
Props { new TestActor("http://127.0.0.1:8080") },
s"${actorName}Service"
)
val service2 = system.actorOf(
Props { new TestActor("http://127.0.0.1:8081") },
s"${actorName}2Service"
)
IO(Http) ! Http.Bind(service, "::0", port = 8081)
IO(Http) ! Http.Bind(service2, "::0", port = 8082)
}
使用curl http://localhost:8082連接到代理服務器，你會看到阿卡系統保持數據傳輸捲曲已被殺害即使（你可以打開詳細的DEBUG級別的日誌）。
我該如何處理這個問題？謝謝。