//! A Riak client for Rust.
//!
//! This client can be used to communicate with Riak clusters to send and receive objects
//! and other information. Operations are done through the `Client` struct and there are
//! several other structs designed to build data structures for sending and receiving
//! data from a Riak cluster.
//!
//! Examples:
//!
//! Storing an object is the most fundamental operation of Riak, it can be done like the following:
//!
//! ```
//! use riak::Client;
//! use riak::object::{ObjectContent, StoreObjectReq};
//!
//! // connect to Riak and ping the server
//! let mut riak = Client::new("10.0.0.2:8087").unwrap();
//! riak.ping().unwrap();
//!
//! // prepare an object
//! let contents = ObjectContent::new("I am the night, I am Batman!".as_bytes());
//!
//! // build a request to store the object
//! let bucket_name = "superheroes".to_string();
//! let mut req = StoreObjectReq::new(&bucket_name, &contents);
//! req.set_key("batman");
//!
//! // store the object
//! riak.store_object(&req).unwrap();
//! ```

#[macro_use]
extern crate log;
extern crate protobuf;

pub mod bucket;
pub mod errors;
pub mod object;

mod connection;
mod rpb;

use bucket::BucketProps;
use errors::RiakError;
use connection::RiakConn;
use object::{FetchObjectReq, StoreObjectReq, FetchObjectResp};
use protobuf::{Message, parse_from_bytes};
use rpb::codes;
use rpb::riak::{RpbGetBucketReq, RpbGetBucketResp};
use rpb::riak_kv::{RpbGetResp, RpbListBucketsResp};
use rpb::utils::{rpb_get_resp_to_fetch_object_resp, rpb_bucket_props_to_bucket_props,
                 RpbGenerator, RpbGeneratorID};
use std::net::ToSocketAddrs;
use std::string::String;

/// `Client` Represents a connection to a Riak server's Protocol Buffers API.
///
/// For more information: https://docs.basho.com/riak/kv/latest/developing/api/protocol-buffers/
#[derive(Debug)]
pub struct Client {
    conn: RiakConn,
}

impl Client {
    /// Constructs a new `Client`.
    ///
    /// # Examples
    ///
    /// ```
    /// use riak::Client;
    ///
    /// let mut riak = Client::new("10.0.0.2:8087").unwrap();
    /// riak.ping().unwrap();
    /// ```
    ///
    /// # Errors
    ///
    /// TODO
    pub fn new<A: ToSocketAddrs>(addr: A) -> Result<Client, RiakError> {
        debug!("creating a new Riak client");
        let conn = match RiakConn::new(addr) {
            Ok(c) => c,
            Err(err) => return Err(err),
        };

        Ok(Client { conn: conn })
    }

    /// Reconnect to the Riak server originally connected to when this client was initiated.
    ///
    /// # Examples
    ///
    /// ```
    /// use riak::Client;
    ///
    /// let mut riak = Client::new("10.0.0.2:8087").unwrap();
    /// riak.reconnect().unwrap();
    /// ```
    ///
    /// # Errors
    ///
    /// TODO
    pub fn reconnect(&mut self) -> Result<(), RiakError> {
        self.conn.reconnect()
    }

    /// Sends a ping message to Riak and returns a Result.
    ///
    /// # Examples
    ///
    /// ```
    /// use riak::Client;
    ///
    /// let mut riak = Client::new("10.0.0.2:8087").unwrap();
    /// riak.ping().unwrap();
    /// ```
    ///
    /// # Errors
    ///
    /// TODO
    pub fn ping(&mut self) -> Result<(), RiakError> {
        let ping_data: Vec<u8> = vec![];
        match self.conn.send(codes::RpbPingReq, codes::RpbPingResp, &ping_data) {
            Ok(_) => Ok(()),
            Err(err) => Err(err),
        }
    }

    /// Produces a list of bucket names.
    ///
    /// Note: This operation is expensive for the Riak server and should be
    /// used as rarely as possible.
    ///
    /// # Examples
    ///
    /// ```
    /// use riak::Client;
    ///
    /// let mut riak = Client::new("10.0.0.2:8087").unwrap();
    /// let buckets = riak.list_buckets().unwrap();
    ///
    /// for bucket in buckets.iter() {
    ///     println!("found bucket named {}", bucket);
    /// }
    /// ```
    ///
    /// # Errors
    ///
    /// TODO
    pub fn list_buckets(&mut self) -> Result<Vec<String>, RiakError> {
        let response = match self.conn.send(codes::RpbListBucketsReq,
                                            codes::RpbListBucketsResp,
                                            &Vec::new()) {
            Ok(resp) => resp,
            Err(err) => {
                debug!("failed to list buckets error was: {:?}", err);
                return Err(err);
            }
        };

        let buckets_resp = match parse_from_bytes::<RpbListBucketsResp>(&response) {
            Ok(parsed) => parsed,
            Err(err) => {
                debug!("failed to parse RpbListBucketsResp error was: {:?}", err);
                return Err(RiakError::ProtobufError(err));
            }
        };

        let buckets = buckets_resp.get_buckets();
        debug!("found buckets: {:?}", buckets);
        let mut bucket_names: Vec<String> = Vec::new();
        for b in buckets {
            let bucket = b.clone();
            let bucket_name = String::from_utf8(bucket).unwrap(); // FIXME
            debug!("found a bucket named: {}", bucket_name);
            bucket_names.push(bucket_name);
        }

        Ok(bucket_names)
    }

    /// Sets the properties for a bucket given a bucket name.
    ///
    /// # Examples
    ///
    /// ```
    /// use riak::Client;
    /// use riak::bucket::BucketProps;
    ///
    /// let mut riak = Client::new("10.0.0.2:8087").unwrap();
    /// let mut bucket_props = BucketProps::new();
    /// bucket_props.set_backend("bitcask");
    ///
    /// riak.set_bucket_properties("superheroes", &bucket_props).unwrap();
    /// ```
    ///
    /// # Errors
    ///
    /// TODO
    pub fn set_bucket_properties(&mut self,
                                 bucket_name: &str,
                                 bucket_props: &BucketProps)
                                 -> Result<(), RiakError> {
        let bytes = match bucket_props.generate_rpb(bucket_name) {
            Ok(b) => b,
            Err(err) => return Err(err),
        };
        match self.conn.send(codes::RpbSetBucketReq, codes::RpbSetBucketResp, &bytes) {
            Ok(_) => Ok(()),
            Err(err) => Err(err),
        }
    }

    /// Retrieves bucket properties for a bucket given a bucket name.
    ///
    /// # Examples
    ///
    /// ```
    /// use riak::Client;
    ///
    /// let mut riak = Client::new("10.0.0.2:8087").unwrap();
    /// let bucket_props = riak.get_bucket_properties("superheroes").unwrap();
    /// println!("bucket properties for superheroes: {:?}", bucket_props);
    /// ```
    ///
    /// # Errors
    ///
    /// TODO
    pub fn get_bucket_properties(&mut self, bucket_name: &str) -> Result<BucketProps, RiakError> {
        let mut req = RpbGetBucketReq::new();
        req.set_bucket(bucket_name.to_string().into_bytes());
        // TODO implement field_type
        let serialized = match req.write_to_bytes() {
            Ok(s) => s,
            Err(err) => return Err(RiakError::ProtobufError(err)),
        };
        match self.conn.send(codes::RpbGetBucketReq, codes::RpbGetBucketResp, &serialized) {
            Ok(response) => {
                match parse_from_bytes::<RpbGetBucketResp>(&response) {
                    Ok(mut rpb_get_bucket_resp) => {
                        let rpb_bucket_props = rpb_get_bucket_resp.take_props();
                        let bucket_props = rpb_bucket_props_to_bucket_props(&rpb_bucket_props);
                        Ok(bucket_props)
                    }
                    Err(err) => {
                        debug!("failed to parse RpbGetBucketResp error was: {:?}", err);
                        Err(RiakError::ProtobufError(err))
                    }
                }
            }
            Err(err) => Err(err),
        }
    }

    /// Stores an object on the Riak server.
    ///
    /// # Examples
    ///
    /// ```
    /// use riak::Client;
    /// use riak::object::{ObjectContent, StoreObjectReq};
    ///
    /// let mut riak = Client::new("10.0.0.2:8087").unwrap();
    ///
    /// let contents = ObjectContent::new("I am the night, I am Batman!".as_bytes());
    /// let mut req = StoreObjectReq::new("superheroes", &contents);
    /// req.set_key("batman");
    ///
    /// riak.store_object(&req).unwrap();
    /// ```
    ///
    /// # Errors
    ///
    /// TODO
    pub fn store_object(&mut self, req: &StoreObjectReq) -> Result<(), RiakError> {
        let bytes = match req.generate_rpb() {
            Ok(b) => b,
            Err(err) => return Err(err),
        };
        match self.conn.send(codes::RpbPutReq, codes::RpbPutResp, &bytes) {
            Ok(_) => Ok(()),
            Err(err) => Err(err),
        }
    }

    /// Fetches an object from the Riak server.
    ///
    /// # Examples
    ///
    /// ```
    /// use riak::Client;
    /// use riak::object::FetchObjectReq;
    ///
    /// let mut riak = Client::new("10.0.0.2:8087").unwrap();
    ///
    /// let req = FetchObjectReq::new("superheroes", "batman");
    /// let object = riak.fetch_object(&req).unwrap();
    /// println!("batman object contained: {:?}", object);
    /// ```
    ///
    /// # Errors
    ///
    /// TODO
    pub fn fetch_object(&mut self, req: &FetchObjectReq) -> Result<FetchObjectResp, RiakError> {
        let bytes = match req.generate_rpb() {
            Ok(b) => b,
            Err(err) => return Err(err),
        };
        match self.conn.send(codes::RpbGetReq, codes::RpbGetResp, &bytes) {
            Ok(response) => {
                match parse_from_bytes::<RpbGetResp>(&response) {
                    Ok(mut rpb_get_resp) => {
                        let fetch_object_resp =
                            rpb_get_resp_to_fetch_object_resp(&mut rpb_get_resp);
                        Ok(fetch_object_resp)
                    }
                    Err(err) => {
                        debug!("failed to parse RpbGetResp error was: {:?}", err);
                        Err(RiakError::ProtobufError(err))
                    }
                }
            }
            Err(err) => Err(err),
        }
    }
}