1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51

/*
 * Copyright 2017 Ben Ashford
 *
 * Licensed under the Apache License, Version 2.0 <LICENSE-APACHE or
 * http://www.apache.org/licenses/LICENSE-2.0> or the MIT license
 * <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your
 * option. This file may not be copied, modified, or distributed
 * except according to those terms.
 */

use std::io;
use std::net::SocketAddr;

use futures::{Future, Sink, Stream};

use tokio_core::net::TcpStream;
use tokio_core::reactor::Handle;

use tokio_io::AsyncRead;

use error;
use resp;

/// TODO: comeback and optimise this number
const DEFAULT_BUFFER_SIZE: usize = 100;

/// Connect to a Redis server and return paired Sink and Stream for reading and writing
/// asynchronously.
pub fn connect(addr: &SocketAddr,
               handle: &Handle)
               -> Box<Future<Item = ClientConnection, Error = io::Error>> {
    let con = TcpStream::connect(addr, handle).map(move |socket| {
        let framed = socket.framed(resp::RespCodec);
        let (write_f, read_f) = framed.split();
        let write_b = write_f.buffer(DEFAULT_BUFFER_SIZE);
        ClientConnection {
            sender: Box::new(write_b),
            receiver: Box::new(read_f),
        }
    });
    Box::new(con)
}

// TODO - is the boxing necessary?  It makes the type signature much simpler
/// A low-level client connection representing a sender and a receiver.
///
/// The two halves operate independently from one another
pub struct ClientConnection {
    pub sender: Box<Sink<SinkItem = resp::RespValue, SinkError = io::Error>>,
    pub receiver: Box<Stream<Item = resp::RespValue, Error = error::Error>>,
}