actix/actix-web
1
0
Fork 0
mirror of https://github.com/fafhrd91/actix-web synced 2024-11-27 09:42:57 +01:00
Code Issues Releases Wiki Activity

make remote addr available to http request

Browse Source
This commit is contained in:
Nikolay Kim 2017-11-10 13:08:15 -08:00
parent 265628750c
commit f369d9af0e
5 changed files with 79 additions and 46 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

30
src/channel.rs
View File

@ -1,4 +1,5 @@
use std::rc::Rc;
use std::net::SocketAddr;
use actix::dev::*;
use bytes::Bytes;
@ -19,23 +20,24 @@ pub trait HttpHandler: 'static {
fn handle(&self, req: &mut HttpRequest, payload: Payload) -> Task;
}
enum HttpProtocol<T, A, H>
where T: AsyncRead + AsyncWrite + 'static, A: 'static, H: 'static
enum HttpProtocol<T, H>
where T: AsyncRead + AsyncWrite + 'static, H: 'static
{
H1(h1::Http1<T, A, H>),
H2(h2::Http2<T, A, H>),
H1(h1::Http1<T, H>),
H2(h2::Http2<T, H>),
}
pub struct HttpChannel<T, A, H>
where T: AsyncRead + AsyncWrite + 'static, A: 'static, H: 'static
pub struct HttpChannel<T, H>
where T: AsyncRead + AsyncWrite + 'static, H: 'static
{
proto: Option<HttpProtocol<T, A, H>>,
proto: Option<HttpProtocol<T, H>>,
}
impl<T, A, H> HttpChannel<T, A, H>
where T: AsyncRead + AsyncWrite + 'static, A: 'static, H: HttpHandler + 'static
impl<T, H> HttpChannel<T, H>
where T: AsyncRead + AsyncWrite + 'static, H: HttpHandler + 'static
{
pub fn new(stream: T, addr: A, router: Rc<Vec<H>>, http2: bool) -> HttpChannel<T, A, H> {
pub fn new(stream: T, addr: Option<SocketAddr>, router: Rc<Vec<H>>, http2: bool)
-> HttpChannel<T, H> {
if http2 {
HttpChannel {
proto: Some(HttpProtocol::H2(
@ -54,14 +56,14 @@ impl<T, A, H> HttpChannel<T, A, H>
}
}*/
impl<T, A, H> Actor for HttpChannel<T, A, H>
where T: AsyncRead + AsyncWrite + 'static, A: 'static, H: HttpHandler + 'static
impl<T, H> Actor for HttpChannel<T, H>
where T: AsyncRead + AsyncWrite + 'static, H: HttpHandler + 'static
{
type Context = Context<Self>;
}
impl<T, A, H> Future for HttpChannel<T, A, H>
where T: AsyncRead + AsyncWrite + 'static, A: 'static, H: HttpHandler + 'static
impl<T, H> Future for HttpChannel<T, H>
where T: AsyncRead + AsyncWrite + 'static, H: HttpHandler + 'static
{
type Item = ();
type Error = ();

15
src/h1.rs
View File

@ -1,6 +1,7 @@
use std::{self, io, ptr};
use std::rc::Rc;
use std::cell::UnsafeCell;
use std::net::SocketAddr;
use std::time::Duration;
use std::collections::VecDeque;
@ -35,10 +36,10 @@ pub(crate) enum Http1Result {
Switch,
}
pub(crate) struct Http1<T: AsyncWrite + 'static, A: 'static, H: 'static> {
pub(crate) struct Http1<T: AsyncWrite + 'static, H: 'static> {
router: Rc<Vec<H>>,
#[allow(dead_code)]
addr: A,
addr: Option<SocketAddr>,
stream: H1Writer<T>,
reader: Reader,
read_buf: BytesMut,
@ -57,12 +58,11 @@ struct Entry {
finished: bool,
}
impl<T, A, H> Http1<T, A, H>
impl<T, H> Http1<T, H>
where T: AsyncRead + AsyncWrite + 'static,
A: 'static,
H: HttpHandler + 'static
{
pub fn new(stream: T, addr: A, router: Rc<Vec<H>>) -> Self {
pub fn new(stream: T, addr: Option<SocketAddr>, router: Rc<Vec<H>>) -> Self {
Http1{ router: router,
addr: addr,
stream: H1Writer::new(stream),
@ -75,7 +75,7 @@ impl<T, A, H> Http1<T, A, H>
h2: false }
}
pub fn into_inner(mut self) -> (T, A, Rc<Vec<H>>, Bytes) {
pub fn into_inner(mut self) -> (T, Option<SocketAddr>, Rc<Vec<H>>, Bytes) {
(self.stream.unwrap(), self.addr, self.router, self.read_buf.freeze())
}
@ -172,6 +172,9 @@ impl<T, A, H> Http1<T, A, H>
Ok(Async::Ready(Item::Http1(mut req, payload))) => {
not_ready = false;
// set remote addr
req.set_remove_addr(self.addr.clone());
// stop keepalive timer
self.keepalive_timer.take();

25
src/h2.rs
View File

@ -3,6 +3,7 @@ use std::rc::Rc;
use std::io::{Read, Write};
use std::cell::UnsafeCell;
use std::time::Duration;
use std::net::SocketAddr;
use std::collections::VecDeque;
use actix::Arbiter;
@ -24,12 +25,12 @@ use payload::{Payload, PayloadError, PayloadWriter};
const KEEPALIVE_PERIOD: u64 = 15; // seconds
pub(crate) struct Http2<T, A, H>
where T: AsyncRead + AsyncWrite + 'static, A: 'static, H: 'static
pub(crate) struct Http2<T, H>
where T: AsyncRead + AsyncWrite + 'static, H: 'static
{
router: Rc<Vec<H>>,
#[allow(dead_code)]
addr: A,
addr: Option<SocketAddr>,
state: State<IoWrapper<T>>,
disconnected: bool,
tasks: VecDeque<Entry>,
@ -42,12 +43,11 @@ enum State<T: AsyncRead + AsyncWrite> {
Empty,
}
impl<T, A, H> Http2<T, A, H>
impl<T, H> Http2<T, H>
where T: AsyncRead + AsyncWrite + 'static,
A: 'static,
H: HttpHandler + 'static
{
pub fn new(stream: T, addr: A, router: Rc<Vec<H>>, buf: Bytes) -> Self {
pub fn new(stream: T, addr: Option<SocketAddr>, router: Rc<Vec<H>>, buf: Bytes) -> Self {
Http2{ router: router,
addr: addr,
disconnected: false,
@ -132,12 +132,15 @@ impl<T, A, H> Http2<T, A, H>
entry.task.disconnected()
}
},
Ok(Async::Ready(Some((req, resp)))) => {
Ok(Async::Ready(Some((mut req, resp)))) => {
not_ready = false;
let (parts, body) = req.into_parts();
self.tasks.push_back(
Entry::new(parts, body, resp, &self.router));
// stop keepalive timer
self.keepalive_timer.take();
self.tasks.push_back(
Entry::new(parts, body, resp, self.addr.clone(), &self.router));
}
Ok(Async::NotReady) => {
// start keep-alive timer
@ -210,6 +213,7 @@ impl Entry {
fn new<H>(parts: Parts,
recv: RecvStream,
resp: Respond<Bytes>,
addr: Option<SocketAddr>,
router: &Rc<Vec<H>>) -> Entry
where H: HttpHandler + 'static
{
@ -219,6 +223,9 @@ impl Entry {
let mut req = HttpRequest::new(
parts.method, path, parts.version, parts.headers, query);
// set remote addr
req.set_remove_addr(addr);
// Payload and Content-Encoding
let (psender, payload) = Payload::new(false);

19
src/httprequest.rs
View File

@ -1,5 +1,6 @@
//! HTTP Request message related code.
use std::{str, fmt};
use std::net::SocketAddr;
use std::collections::HashMap;
use bytes::BytesMut;
use futures::{Async, Future, Stream, Poll};
@ -25,6 +26,7 @@ pub struct HttpRequest {
cookies: Vec<Cookie<'static>>,
cookies_loaded: bool,
extensions: Extensions,
addr: Option<SocketAddr>,
}
impl HttpRequest {
@ -43,6 +45,7 @@ impl HttpRequest {
cookies: Vec::new(),
cookies_loaded: false,
extensions: Extensions::new(),
addr: None,
}
}
@ -57,6 +60,7 @@ impl HttpRequest {
cookies: Vec::new(),
cookies_loaded: false,
extensions: Extensions::new(),
addr: None,
}
}
@ -86,6 +90,21 @@ impl HttpRequest {
&self.path
}
/// Remote IP of client initiated HTTP request.
///
/// The IP is resolved through the following headers, in this order:
///
/// - Forwarded
/// - X-Forwarded-For
/// - peername of opened socket
pub fn remote(&self) -> Option<&SocketAddr> {
self.addr.as_ref()
}
pub(crate) fn set_remove_addr(&mut self, addr: Option<SocketAddr>) {
self.addr = addr
}
/// Return a new iterator that yields pairs of `Cow<str>` for query parameters
#[inline]
pub fn query(&self) -> HashMap<String, String> {

36
src/server.rs
View File

@ -1,5 +1,6 @@
use std::{io, net};
use std::rc::Rc;
use std::net::SocketAddr;
use std::marker::PhantomData;
use actix::dev::*;
@ -7,6 +8,8 @@ use futures::Stream;
use tokio_io::{AsyncRead, AsyncWrite};
use tokio_core::net::{TcpListener, TcpStream};
#[cfg(feature="tls")]
use futures::Future;
#[cfg(feature="tls")]
use native_tls::TlsAcceptor;
#[cfg(feature="tls")]
@ -64,7 +67,7 @@ impl<T, A, H> HttpServer<T, A, H>
S: Stream<Item=(T, A), Error=io::Error> + 'static
{
Ok(HttpServer::create(move |ctx| {
ctx.add_stream(stream.map(|(t, a)| IoStream(t, a, false)));
ctx.add_stream(stream.map(|(t, _)| IoStream(t, None, false)));
self
}))
}
@ -109,7 +112,7 @@ impl<H: HttpHandler> HttpServer<TcpStream, net::SocketAddr, H> {
Ok(HttpServer::create(move |ctx| {
for (addr, tcp) in addrs {
info!("Starting http server on {}", addr);
ctx.add_stream(tcp.incoming().map(|(t, a)| IoStream(t, a, false)));
ctx.add_stream(tcp.incoming().map(|(t, a)| IoStream(t, Some(a), false)));
}
self
}))
@ -146,7 +149,7 @@ impl<H: HttpHandler> HttpServer<TlsStream<TcpStream>, net::SocketAddr, H> {
ctx.add_stream(tcp.incoming().and_then(move |(stream, addr)| {
TlsAcceptorExt::accept_async(acc.as_ref(), stream)
.map(move |t| {
IoStream(t, addr)
IoStream(t, Some(addr), false)
})
.map_err(|err| {
trace!("Error during handling tls connection: {}", err);
@ -200,7 +203,7 @@ impl<H: HttpHandler> HttpServer<SslStream<TcpStream>, net::SocketAddr, H> {
} else {
false
};
IoStream(stream, addr, http2)
IoStream(stream, Some(addr), http2)
})
.map_err(|err| {
trace!("Error during handling tls connection: {}", err);
@ -213,32 +216,31 @@ impl<H: HttpHandler> HttpServer<SslStream<TcpStream>, net::SocketAddr, H> {
}
}
struct IoStream<T, A>(T, A, bool);
struct IoStream<T>(T, Option<SocketAddr>, bool);
impl<T, A> ResponseType for IoStream<T, A>
where T: AsyncRead + AsyncWrite + 'static,
A: 'static
impl<T> ResponseType for IoStream<T>
where T: AsyncRead + AsyncWrite + 'static
{
type Item = ();
type Error = ();
}
impl<T, A, H> StreamHandler<IoStream<T, A>, io::Error> for HttpServer<T, A, H>
impl<T, A, H> StreamHandler<IoStream<T>, io::Error> for HttpServer<T, A, H>
where T: AsyncRead + AsyncWrite + 'static,
A: 'static,
H: HttpHandler + 'static {}
impl<T, A, H> Handler<IoStream<T, A>, io::Error> for HttpServer<T, A, H>
where T: AsyncRead + AsyncWrite + 'static,
A: 'static,
H: HttpHandler + 'static,
A: 'static {}
impl<T, A, H> Handler<IoStream<T>, io::Error> for HttpServer<T, A, H>
where T: AsyncRead + AsyncWrite + 'static,
H: HttpHandler + 'static,
A: 'static,
{
fn error(&mut self, err: io::Error, _: &mut Context<Self>) {
debug!("Error handling request: {}", err)
}
fn handle(&mut self, msg: IoStream<T, A>, _: &mut Context<Self>)
-> Response<Self, IoStream<T, A>>
fn handle(&mut self, msg: IoStream<T>, _: &mut Context<Self>)
-> Response<Self, IoStream<T>>
{
Arbiter::handle().spawn(
HttpChannel::new(msg.0, msg.1, Rc::clone(&self.h), msg.2));