Я пытаюсь реализовать обратный HTTP-прокси с помощью Spray/Akka, но возникает проблема. Я обнаружил, что в некоторых случаях мой прокси-сервер будет продолжать получать данные с восходящего сервера даже после того, как клиент отключится.Сброс обратного прокси: сохранить передачу данных после того, как клиент отключился
Вот как я реализую мой 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()
Что в принципе ничего не делает, но открыть Chunked ответ (для долгосрочного бега, так что мы можем экзамен вопросы). И если я приковать два слоя прокси следующим образом:
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 для подключения к прокси-серверу, и вы увидите, что система Akka продолжает передавать данные даже после того, как ротор был убит (вы можете включить журналы уровня DEBUG, чтобы увидеть подробности).
Как я могу справиться с этой проблемой? Благодарю.