allow to use Connection for sending client request

src/client/connector.rs

@@ -55,11 +55,11 @@ pub enum ClientConnectorError {
     /// SSL error
     #[cfg(feature="alpn")]
     #[fail(display="{}", _0)]
-    SslError(OpensslError),
+    SslError(#[cause] OpensslError),
 
     /// Connection error
     #[fail(display = "{}", _0)]
-    Connector(ConnectorError),
+    Connector(#[cause] ConnectorError),
 
     /// Connecting took too long
     #[fail(display = "Timeout out while establishing connection")]
@@ -71,7 +71,7 @@ pub enum ClientConnectorError {
 
     /// Connection io error
     #[fail(display = "{}", _0)]
-    IoError(io::Error),
+    IoError(#[cause] io::Error),
 }
 
 impl From<ConnectorError> for ClientConnectorError {
@@ -98,7 +98,6 @@ impl Default for ClientConnector {
         #[cfg(feature="alpn")]
         {
             let mut builder = SslConnector::builder(SslMethod::tls()).unwrap();
-            builder.set_verify(SslVerifyMode::NONE);
             ClientConnector {
                 connector: builder.build()
             }
@@ -269,6 +268,10 @@ impl Connection {
     pub fn stream(&mut self) -> &mut IoStream {
         &mut *self.stream
     }
+
+    pub fn from_stream<T: IoStream>(io: T) -> Connection {
+        Connection{stream: Box::new(io)}
+    }
 }
 
 impl IoStream for Connection {

src/client/pipeline.rs

@@ -39,6 +39,7 @@ impl From<io::Error> for SendRequestError {
 enum State {
     New,
     Connect(actix::dev::Request<Unsync, ClientConnector, Connect>),
+    Connection(Connection),
     Send(Box<Pipeline>),
     None,
 }
@@ -64,6 +65,14 @@ impl SendRequest {
             state: State::New,
             conn: conn}
     }
+
+    pub(crate) fn with_connection(req: ClientRequest, conn: Connection) -> SendRequest
+    {
+        SendRequest{
+            req: req,
+            state: State::Connection(conn),
+            conn: ClientConnector::from_registry()}
+    }
 }
 
 impl Future for SendRequest {
@@ -84,31 +93,34 @@ impl Future for SendRequest {
                     },
                     Ok(Async::Ready(result)) => match result {
                         Ok(stream) => {
-                            let mut writer = HttpClientWriter::new(SharedBytes::default());
-                            writer.start(&mut self.req)?;
-
-                            let body = match self.req.replace_body(Body::Empty) {
-                                Body::Streaming(stream) => IoBody::Payload(stream),
-                                Body::Actor(ctx) => IoBody::Actor(ctx),
-                                _ => IoBody::Done,
-                            };
-
-                            let mut pl = Box::new(Pipeline {
-                                body: body,
-                                conn: stream,
-                                writer: writer,
-                                parser: HttpResponseParser::default(),
-                                parser_buf: BytesMut::new(),
-                                disconnected: false,
-                                running: RunningState::Running,
-                                drain: None,
-                            });
-                            self.state = State::Send(pl);
+                            self.state = State::Connection(stream)
                         },
                         Err(err) => return Err(SendRequestError::Connector(err)),
                     },
-                    Err(_) =>
-                        return Err(SendRequestError::Connector(ClientConnectorError::Disconnected))
+                    Err(_) => return Err(SendRequestError::Connector(
+                        ClientConnectorError::Disconnected))
+                },
+                State::Connection(stream) => {
+                    let mut writer = HttpClientWriter::new(SharedBytes::default());
+                    writer.start(&mut self.req)?;
+
+                    let body = match self.req.replace_body(Body::Empty) {
+                        Body::Streaming(stream) => IoBody::Payload(stream),
+                        Body::Actor(ctx) => IoBody::Actor(ctx),
+                        _ => IoBody::Done,
+                    };
+
+                    let mut pl = Box::new(Pipeline {
+                        body: body,
+                        conn: stream,
+                        writer: writer,
+                        parser: HttpResponseParser::default(),
+                        parser_buf: BytesMut::new(),
+                        disconnected: false,
+                        running: RunningState::Running,
+                        drain: None,
+                    });
+                    self.state = State::Send(pl);
                 },
                 State::Send(mut pl) => {
                     pl.poll_write()

src/client/request.rs

@@ -13,7 +13,7 @@ use body::Body;
 use error::Error;
 use headers::ContentEncoding;
 use super::pipeline::SendRequest;
-use super::connector::ClientConnector;
+use super::connector::{Connection, ClientConnector};
 
 /// An HTTP Client Request
 pub struct ClientRequest {
@@ -179,9 +179,14 @@ impl ClientRequest {
         SendRequest::new(self)
     }
 
+    /// Send request using custom connector
     pub fn with_connector(self, conn: Addr<Unsync, ClientConnector>) -> SendRequest {
         SendRequest::with_connector(self, conn)
+    }
 
+    /// Send request using existing Connection
+    pub fn with_connection(self, conn: Connection) -> SendRequest {
+        SendRequest::with_connection(self, conn)
     }
 }
 

tests/test_client.rs

@@ -0,0 +1,59 @@
+extern crate actix;
+extern crate actix_web;
+extern crate bytes;
+extern crate futures;
+
+use bytes::Bytes;
+
+use actix_web::*;
+
+
+const STR: &str =
+    "Hello World Hello World Hello World Hello World Hello World \
+     Hello World Hello World Hello World Hello World Hello World \
+     Hello World Hello World Hello World Hello World Hello World \
+     Hello World Hello World Hello World Hello World Hello World \
+     Hello World Hello World Hello World Hello World Hello World \
+     Hello World Hello World Hello World Hello World Hello World \
+     Hello World Hello World Hello World Hello World Hello World \
+     Hello World Hello World Hello World Hello World Hello World \
+     Hello World Hello World Hello World Hello World Hello World \
+     Hello World Hello World Hello World Hello World Hello World \
+     Hello World Hello World Hello World Hello World Hello World \
+     Hello World Hello World Hello World Hello World Hello World \
+     Hello World Hello World Hello World Hello World Hello World \
+     Hello World Hello World Hello World Hello World Hello World \
+     Hello World Hello World Hello World Hello World Hello World \
+     Hello World Hello World Hello World Hello World Hello World \
+     Hello World Hello World Hello World Hello World Hello World \
+     Hello World Hello World Hello World Hello World Hello World \
+     Hello World Hello World Hello World Hello World Hello World \
+     Hello World Hello World Hello World Hello World Hello World \
+     Hello World Hello World Hello World Hello World Hello World";
+
+
+#[test]
+fn test_simple() {
+    let mut srv = test::TestServer::new(
+        |app| app.handler(|_| httpcodes::HTTPOk.build().body(STR)));
+
+    let request = srv.get().header("x-test", "111").finish().unwrap();
+    let repr = format!("{:?}", request);
+    assert!(repr.contains("ClientRequest"));
+    assert!(repr.contains("x-test"));
+
+    let response = srv.execute(request.send()).unwrap();
+    assert!(response.status().is_success());
+
+    // read response
+    let bytes = srv.execute(response.body()).unwrap();
+    assert_eq!(bytes, Bytes::from_static(STR.as_ref()));
+
+    let request = srv.post().finish().unwrap();
+    let response = srv.execute(request.send()).unwrap();
+    assert!(response.status().is_success());
+
+    // read response
+    let bytes = srv.execute(response.body()).unwrap();
+    assert_eq!(bytes, Bytes::from_static(STR.as_ref()));
+}