ringmaster_client 0.1.1

//! Client library for a ringmaster_client.
//! THe ring master advertises the RING_MASTER service
//! on the port manager.  It accepts several
//! textual requests: 
//! 
//! *  CONNECT - connects us, the client to a ring,
//! either as a producer or a consumer.
//! *  DISCONNECT - disconnects an existing client from a ring.
//! *  LIST  - returns a list of rings and their statistics.
//! this is in the form of a Tcl list and we include a simple
//! list parsing package.
//! * REGISTER - registers the creation of a new ringbuffer.
//! * UNREGISTER - informs the ring  master of a ring destruction.
//! * REMOTE - asks the ringmaster to set up a ring2stdout
//! that will then ship data across the connection.  This supports
//! obtainng data from remote ringbuffers.
//! 
//! It is intended that this crate be used in conjunction with the
//! nscldaq_ringbuffer package to correctly access rings.
//! 
//! Note:  Tests for this 6crate must be run with --test-threads 1 or they
//! will fail.  Alternatives will essentially serialize anyway by 
//! e.g. entering a mutex when a test starts and then exiting it...so the
//! only point of that would be to make it not necessary for the user
//! to supply -- --test-threads 1 to Cargo test.
//! 

use std::net;
use std::{net::ToSocketAddrs, process, io::Write, io::Read, ops,
     thread, time::Duration, clone, fs, sync, collections::HashSet, 
     env, path, process::Command, process::Stdio};
use portman_client;
use nscldaq_ringbuffer::ringbuffer;
use url::Url;
use local_ip_address;
use std::os::fd::IntoRawFd;
use std::os::fd::FromRawFd;

const PORTMAN_PORT: u16 =30000;
const RINGMASTER_SERVICE : &str = "RingMaster";
const RINGBUFFER_DIRECTORY : &str = "/dev/shm";
const DEFAULT_RING_SIZE : u32 =8*1024*1024;     // Default ring size is 8Mbytes.

/// parse a Tcl list.  Note that to process sublists, you will
/// need to know that an element is a sublist and then call
/// parsee_tcl_list on that subelement.
/// 
/// Example usage:
/// let tcl_string = r#"element1 {a b {nested list}} "quoted element with spaces" element4"#;
/// let parsed_list = parse_tcl_list(tcl_string);
/// parsed_list would contain ["element1", "a b {nested list}", "quoted element with spaces", "element4"]
/// 
/// Note we got this initially from goodle's artificial Idiot so I guess changes will be needed.
fn parse_tcl_list(input: &str) -> Vec<String> {
    let mut result = Vec::new();
    let mut chars = input.chars().peekable();

    while let Some(c) = chars.peek() {
        match c {
            '{' => {
                // Handle braced strings (recursive parsing for nested lists)
                let mut brace_count = 0;
                let mut current_item = String::new();
                chars.next(); // Consume '{'
                while let Some(inner_c) = chars.next() {
                    if inner_c == '{' {
                        brace_count += 1;
                    } else if inner_c == '}' {
                        if brace_count == 0 {
                            break; // End of current braced item
                        } else {
                            brace_count -= 1;
                        }
                    }
                    current_item.push(inner_c);
                }
                result.push(current_item.trim().to_string()); // Trim potential whitespace within braces
            }
            '"' => {
                // Handle quoted strings
                let mut current_item = String::new();
                chars.next(); // Consume '"'
                while let Some(inner_c) = chars.next() {
                    if inner_c == '"' {
                        break; // End of quoted item
                    }
                    current_item.push(inner_c);
                }
                result.push(current_item);
            }
            _ if c.is_whitespace() => {
                // Consume whitespace
                chars.next();
            }
            _ => {
                // Handle unquoted words
                let mut current_item = String::new();
                while let Some(inner_c) = chars.peek() {
                    if inner_c.is_whitespace() || *inner_c == '{' || *inner_c == '"' {
                        break;
                    }
                    current_item.push(chars.next().unwrap());
                }
                if !current_item.is_empty() {
                    result.push(current_item);
                }
            }
        }
    }
    result
}
// Private method to construct the path to the 
// stdintoring program which is the local side
// of the remote to local hoisting pipeline.
// Assumptions:
//    The environment variable DAQBIN is defined and points
//  to the directory that holds that program.
//
//  We ensure that this program exists.
fn stdin_to_ring_path() -> Result<String, String> {
    let program_file = "stdintoring";
    let bindir_env = "DAQBIN";
    match env::var(bindir_env) {
        Ok(bindir) => {
            let full_path = format!("{}/{}", bindir, program_file);
            let path = path::Path::new(&full_path);
            match path.try_exists() {
                Ok(yesno)  => {
                    if yesno {
                        return Ok(String::from(path.to_str().unwrap()));
                    } else {
                        return Err(format!("{} can't be found", path.to_str().unwrap()));
                    }
                },
                Err(reason) => {
                    return Err(format!("Could not look up {} in the filesystem: {}", path.to_str().unwrap(), reason));
                }
            }
        },
        Err(reason) =>
            return Err(format!("DAQBIN must be defined: {} ", reason))
    }
    
}
// proxy ring name given a host and remote ring name:

fn proxy_ring_name(host: &str, remote_ring: &str) -> String {
    format!("{}.{}", host, remote_ring)
}
// Get the proxy ring data size... It's the # megabytes in 
// The env var NSCLDAQ_DEFAULT_PROXYMB if it's an integer.

fn proxy_ring_size() -> u32 {
    let mut result = DEFAULT_RING_SIZE;

    match env::var("NSCLDAQ_DEFAULT_PROXYMB") {
        Ok(strval) => {
            // Only use it if the integer parse is ok:

            let intval  = strval.parse::<u32>();
            if let Ok(mb) = intval {
                result = mb*1024*1024;
            }
        },
        Err(_) => {
        
        }
    };

    return result;
}

/// Some methods require the eactual path to the ring.
/// this returns it.
/// 
pub fn ring_path(ring: &str) -> String {
    // Create the full ring path given a ring name:

    format!("{}/{}", RINGBUFFER_DIRECTORY, ring)
}
///
/// The Client struct and implementation provide the
/// mechanism by means requests are sent to the server 
/// and responses an analyzed. With a few exceptions
/// that will be described in the implementation,
/// Each request:
///  * gets the port number of the  ringmaster.
///  * connects to the ringmaster.
///  * sends the request.
///  * processes the response
///  * closes the connection.
/// 
/// See, however REMOTE and CONNECT
/// 

#[derive(Debug)]
pub struct Client {
    host : String,                // Which ringmaster.
    socket : Option<net::TcpStream>, // When we need a persistent connection. This is Some.
}


impl ops::Drop for Client {
    #[doc = r"Shutdown the socket if it exists -- don't care if the shutdown fails."]
    fn drop(&mut self) {
        if let Some(s) = &self.socket {
            let _ = s.shutdown(net::Shutdown::Both);   // If the socket is open shut it down.
        }
    }
}

impl clone::Clone for Client {
    #[doc = r"Provide clone... if can't clone socket it'll be None."]
    fn clone(&self) -> Self {
        Client {
            host: self.host.clone(),
            socket: match &self.socket {
                Some(s) => {
                    match s.try_clone() {
                        Ok(s) => Some(s),
                        Err(_) => None
                    }
                },
                None => None
            }
        }
    }
}
/// RingInformation provides the information about a ring decoded from the ring master
/// LIST request.    Note that we do use nscldaq_ringbuffer::ringbuffer::ConsumerInformation strucs
/// for the consumer data.

#[derive(Debug)]
pub struct RingInformation {
    pub name : String,
    pub size : u32,
    pub free : u32,
    pub maxconsumers : u32, 
    pub producer_pid : i32,       // Negative if none.
    pub max_get : u32,
    pub min_get : u32,

    pub consumers : Vec<ringbuffer::ConsumerUsage>,
}
impl Client {
    // private methods:

    // Get the ring master port.
    fn get_port(&self) -> Result<u16, portman_client::Error> {
        let mut portman = portman_client::Client::new(PORTMAN_PORT);
        let matches = portman.find_by_service(RINGMASTER_SERVICE)?;
        if matches.len() > 0 {
            Ok(matches[0].port)
        } else {
            Err(portman_client::Error::Unimplemented)
        }
    }
    // Open a socket to the ringmaster:

    fn connect(&self) -> std::io::Result<net::TcpStream> {
        let port = self.get_port();
        if let Ok(num) = port {
            net::TcpStream::connect(format!("{}:{}", self.host, num))
        } else {
            Err(std::io::Error::other("Failed to get ringmaster port"))      
        }
    }
    // Opens a connection and svaes it in socket.
    // not.
    fn connect_persistent(&mut self) -> Result<net::TcpStream, std::io::Error> {
        let sock_result = self.connect();
        match sock_result {
            Ok(socket) => {
                self.socket = Some(socket.try_clone().unwrap());
                return Ok(socket);
                
            },
            Err(e) => Err(e)
        }
    }
    // Read a line from the socket but strip the \r\n characters.
    // line is assumed to end with \n

    fn read_line(sock: &mut net::TcpStream) -> Result<String, std::io::Error> {
        let mut reply = String::from("");     // We'll extend this.
        let mut buf : [u8; 1] = [0];                  // have to read byte by byte _sigh_.
        loop {
            let n = sock.read(&mut buf)?;
            if n == 0 {
                break;                    // Peer closed.              
            }
            let c = buf[0] as char;
            if c == '\n' {
                break;     // end of line.
            }
            if c != '\r' {
                reply.push(buf[0] as char);
            }
        }

        Ok(reply)
    }
    // Perform a transaction and either give the error or the returned string.
    // Send a message to the ring master and get the resopnse:

    fn transaction(&mut self, sock : &mut net::TcpStream, request : &str) -> Result<String, std::io::Error> {
        sock.write_all(request.as_bytes())?;
        sock.flush()?;

        Client::read_line(sock)
        
    }
    //   Get the existing socket or create a new persistent one if needed:

    fn get_socket_or_create(&mut self) -> Result<net::TcpStream, String>  {
        if let None = self.socket {
            if let Err(result) = self.connect_persistent() {
                return Err(format!("Failed to open socket to ringmaster {}", result));
            }
        }
        return Ok(self.socket()
            .expect("Ring master client expected to have socket but did not"));

        
    }
    // Do a transaction and analyze the response.  No we also take care of
    // our side of shutting down the socket on failure.
    fn transact_and_analyze(&mut self, request : &str) -> Result<(), String> {
        let mut socket = self.get_socket_or_create()?;
        
        let response = match self.transaction(&mut socket, &request) {
            Ok(reply) => Ok(reply),
            Err(reason) => Err(format!("Ring master transaction failed {}", reason))
        }?;
        
        if response == "OK" {
            return Ok(());
        } else if response == "OK BINARY FOLLOWS"
        {
            // That's what we get on a REMOTE request:

            return Ok(());
        } else {
            let _ = socket.shutdown(net::Shutdown::Both);
            self.socket = None;
            return Err(response);
        }
    
        
    }
    fn parse_else<T: std::str::FromStr>(string : &str) -> Result<T, String> {
        let parsed = string.parse::<T>();
        match parsed {
            Ok(result) => Ok(result),
            Err(_) => Err(format!("Numeric parse failed for {} ", string))
        }
    }
    // Analyze the information about a consumer to, hopefully, get a ringbuffer:ClientInformation
    // struct:

    fn analyze_consumer(consumer: &str, ring_size: u32) -> Result<ringbuffer::ConsumerUsage, String> {
        let consumer_list = parse_tcl_list(consumer);
        if consumer_list.len() != 2  {
            return Err(format!("Expected consumer info to have {} entries, got {}", 2, consumer_list.len()));
        }
        let consumer_pid : u32 = Self::parse_else(&consumer_list[0])?;
        let consumer_maxget : u32 = Self::parse_else(&consumer_list[1])?;

        let result = ringbuffer::ConsumerUsage {
            pid: consumer_pid,
            free : (ring_size - consumer_maxget) as usize,
            available : consumer_maxget as usize
        };
        Ok(result)
    }
    // Analyze the list for one ring.
    fn analyze_ring(ring : &str) -> Result<RingInformation, String>  {
        // The ring is a list too:, 
        // name, {size, maxput, maxconsumers, producer, maxget, minget, consumer-list}
        let ring_list = parse_tcl_list(ring);
        if ring_list.len() != 2 {
            return Err(format!("Ring info was not correct expected {} got {}", 2, ring_list.len()));
        }
        let name = ring_list[0].clone();                
        let info_list = parse_tcl_list(&ring_list[1]);
        
        // Require the right number of eleme4nts:

        if info_list.len() != 7 {
            return Err(format!("Size/consumer list incorrect size expected {} got {}",7, info_list.len()));
        }
        let ring_size : u32 = Self::parse_else(&info_list[0])?;
        let ring_free : u32 = Self::parse_else(&info_list[1])?;
        let max_consumer : u32 = Self::parse_else(&info_list[2])?;
        let producer_pid :i32 = Self::parse_else(&info_list[3])?;
        let max_get : u32 = Self::parse_else(&info_list[4])?;
        let min_get : u32 = Self::parse_else(&info_list[5])?;

        let consumers = parse_tcl_list(&info_list[6]);  // List of consumer info:

        // Fill in what we can of the RingInfo result and
        // then get the consumer stuff:

        let mut result = RingInformation {
            name : name.clone(),
            size : ring_size,
            free : ring_free,
            maxconsumers : max_consumer,
            producer_pid: producer_pid,
            max_get : max_get, 
            min_get: min_get,
            consumers : vec![]
        };
        for consumer in consumers {
            result.consumers.push(Self::analyze_consumer(&consumer, ring_size)?);
        }
        Ok(result)
        
    }
    // Analyze the Tcl list that was returned from LIST:
    // See e.g. https://docs.frib.msu.edu/daq/newsite/nscldaq-11.3/c8123.html#AEN8136 
    // for a description of the list elements.
    // While parse_tcl_list returns a Vec<String> it's possible, theoretically
    // to fail parsing some elmenets so we return a result

    fn analyze_ring_listing(tcl_list : &str) -> Result<Vec<RingInformation>, String> {
        let mut result = Vec::<RingInformation>::new();

        let ring_list = parse_tcl_list(tcl_list);  // Vector one per ring def.
        for ring in ring_list {
            result.push(Client::analyze_ring(&ring)?);
        }

        Ok(result)
    }

    /// Create a new client that will connect to the specified host.
    /// Requests figure out the port and either make transitory
    /// TcpStreams or alternatively,
    /// create 
    pub fn new(host : &str) -> Client {
        Client {
            host: host.to_string(), 
            socket: None
        }
    }
    /// Get the host we're talking to.
    pub fn host(&self) -> String {
        self.host.clone()
    }
    /// If there's a persistent socket, get it.
    pub fn socket(&self) -> Option<net::TcpStream> {
        match &self.socket {
            None => None,
            Some(s) => Some(s.try_clone().expect("Ringmaster client unable to clone the socket"))
        }
    }


    // Operations on the ringmaster:

    /// register_ring
    ///    Register a new ring with the ring master.
    /// 
    pub fn register_ring(&mut self, ringname : &str) -> Result<(), String> {

        let request = format!("REGISTER {} \n", ringname);
        self.transact_and_analyze(&request)
        
        
    }
    /// connect_as_producer
    ///  Connect ourselves to a ringbuffer as a producer.
    /// This requires that we hold a persistent connection to the
    /// ringmaster until we disconnect.
    ///   THe format of the message we send is:
    ///
    ///  CONNECT {ringname} producer mypid "comment"
    // The response should be OK or FAIL reason for failure
    //
    // On failure we return the string for the failure.
    pub fn connect_as_producer(&mut self, ringname : &str, comment : &str) -> Result<(), String> {
        
        let pid = process::id();
        let request = format!("CONNECT {{{}}} producer {} \"{}\" \n", ringname, pid, comment);
        self.transact_and_analyze(&request)

    }
    ///
    /// disconnect_producer
    ///    Disconnects the current pid as the producer.
    /// 
    pub fn disconnect_producer(&mut self, ringname: &str) -> Result<(), String> {
        let pid = process::id();
        let request = format!("DISCONNECT {{{}}} producer {}\n", ringname, pid);
        self.transact_and_analyze(&request)
    }

    ///
    /// connect_as_consumer
    ///    Register connection to a ringbuffer as a consumer.
    /// 
    pub fn connect_as_consumer(&mut self, ring: &str, slot : u32, comment: &str) -> Result<(), String> {
        let pid = process::id();
        let request = format!("CONNECT {{{}}} consumer.{} {} \"{}\" \n", ring, slot, pid, comment);
        self.transact_and_analyze(&request)
    }
    ///
    /// disconnect_consumer
    ///    Disconnect a specific consumer from the ring as the ringmaster sees it.
    /// 
    pub fn disconnect_consumer(&mut self, ring : &str, slot : u32) -> Result<(), String> {
        let pid = process::id();
        let request = format!("DISCONNECT {{{}}} consumer.{} {} \n", ring, slot, pid);
        self.transact_and_analyze(&request)
    }
    ///
    /// list_rings
    ///    Produces a list of ring 
    ///    Note that after the Ok there's another line to read from the socket.
    /// 
    pub fn list_rings(&mut self) -> Result<Vec<RingInformation>, String> {
        self.transact_and_analyze("LIST\n")?;

        // The next line from the socket is the Tcl formatted listing:
        let mut socket = self.socket()
            .expect("If list_rings got here there shoulid still be a socket!! but there wasn't");
        let tcl_list = Client::read_line(&mut socket);
        if let Err(ioerr) = tcl_list {
            return Err(format!("Error reading listing: {}", ioerr));
        }
        let tcl_list = tcl_list.unwrap();
        Client::analyze_ring_listing(&tcl_list)

    }
    /// 
    /// unregister_ring
    ///   Make the ringmaster aware of the pending destruction of a ringbuffer.
    /// Note that this shoulid be invoked _before_ deleting the ringbuffer file
    /// because the ringmaster first will force the client processes to back off the
    /// ring (by deleting them) and prevent further connection requests by
    /// refusing to recognized CONNECT's for that ring.
    /// 
    pub fn unregister_ring(&mut self, name :&str) -> Result<(), String> {
        let request = format!("UNREGISTER {}\n", name);
        self.transact_and_analyze(&request)
        
    } 
    ///
    /// get_data
    ///     This is usually called to connect to a remote ring.  On success,
    /// the ring master:
    ///     - Spools up  client to the remote ring that will push data out the stream.
    ///     - Returns a clone of the tcp stream connected to that client.
    /// Note that once this is done, no other ring master requests can be made of this
    /// client and it should be dropped.
    /// 
    /// This is usually used to set up a proxy ring in the local host for a remote ring.
    /// 
    pub fn get_data(&mut self, ring: &str) -> Result<net::TcpStream, String> {
        let request = format!("REMOTE {} \n", ring);
        if let Err(s) = self.transact_and_analyze(&request) {
            Err(s)
        } else {
            // Let the consumer/hoister get established:

            thread::sleep(Duration::from_secs(2));
            let result = self.socket().unwrap().try_clone().unwrap();
            self.socket = None;
            Ok(result)
        }
   
    }
}


/// A ring buffer producer, needs a continuous connection to the ring master or it will be 
/// forced off the ring. This struct packages that with the producer.  Creating one of these
/// will make the producer connection if it can and register it with the ring master.
/// 
pub struct RingBufferProducer {
    ringmaster : Client,
    name       : String,
    pub ring   : ringbuffer::producer::Producer      // User can just do what it does.
}

/// When we drop we should disconnect just to be sure.
/// though theoretically, closing the client wouuld take care of that...
/// there's always the possibility of cloned clients.
impl ops::Drop for RingBufferProducer {
    fn drop(&mut self) {
        let _ = self.ringmaster.disconnect_producer(&self.name);
    }
}
impl RingBufferProducer {
    
    // The ring buffer is known to exist, produce into it:

    fn produce(name : &str, path: &str) -> Result<RingBufferProducer, String> {
        let map = ringbuffer::RingBufferMap::new(path)?;
        let ring = ringbuffer::ThreadSafeRingBuffer::new(sync::Mutex::new(map));
        if let Ok(producer) = ringbuffer::producer::Producer::attach(&ring) {
            let mut c = Client::new("localhost");   // Must produce locally.
            c.connect_as_producer(name, "Rust Ring producer")?;
            return Ok(RingBufferProducer {
                ringmaster: c,
                name: String::from(name),
                ring: producer
            });

        } else {
            return Err(String::from("Could not create a producer object for the ring"));
        }

    }
    /// Create and register a ringbuffer:

    pub fn make_and_register(path: &str, name: &str) -> Result<(), String> {
        let mut c = Client::new("localhost");
        let ring_size : u32 = DEFAULT_RING_SIZE;
        ringbuffer::RingBufferMap::create(path, ring_size)?;
        c.register_ring(name)?;
        Ok(())
    }
    ///
    /// Attach to a pre-existing ring as a producer.  This will fail
    /// if there is no such ringbuffer.
    /// 
    pub fn attach(name : &str) -> Result<RingBufferProducer, String> {
        // check ring for existence (if it exists we assume it's registered)
        let path = ring_path(name);
        match fs::exists(&path) {
            Ok(exists) => { if !exists {
                return Err(format!("Ring buffer file for {} does not exist", name));
            } else {
                return Self::produce(name, &path);
            }},
            Err(_)  => {
                return Err(String::from("Could not check existence of ring file"));
            }
        };
    }
    ///
    /// Attach as a producer creating the ring if it does not already exists.
    /// 
    pub fn create_and_attach(name: &str) -> Result<RingBufferProducer, String> {
        let path = ring_path(name);
        if let Ok(exists) = fs::exists(&path) {
            if !exists {
                // Need to make/register it:

                Self::make_and_register(&path, name)?;
            }
            // ow the ring exists:

            return Self::produce(name, &path);
        } else {
            return Err(String::from("Unable to check ringbuffer existence"));
        }
    }
}

pub fn kill_ring(name : &str) -> Result<(), String> {
    let path = ring_path(name);
    let mut c = Client::new("localhost");
    c.unregister_ring(name)?;
    ringbuffer::RingBufferMap::delete(&path)?;


    Ok(())
}

///  Provides a consumer that auto registers with the ring. Note that
/// consumers require that a ring already exist to connect.. there's no
/// create_and_attach method unlike Producers.
///  There is, however, an exception.  consumer rings are specified via URLs
/// of the form:  tcp://hostname/ringname
/// 
/// If hostname is not a locl host, the remote ringmaster is contacted and a hoister to
/// a local proxy ring is created...it is this proxy ring that is actually what is
/// consumed from.
/// 
pub struct RingBufferConsumer {
    ringmaster : Client,
    name       : String,
    pub consumer : ringbuffer::consumer::Consumer
}
/// Drop allows us to unregister the consumer.:
/// Note that in the end, the ringbuffer will be a local one
/// 
impl Drop for RingBufferConsumer {
    fn drop(&mut self) {
        let idx = self.consumer.get_index();
        let _ignore = self.ringmaster
            .disconnect_consumer(&self.name, idx);
    }
}
impl RingBufferConsumer {
    // Determine if a host is local;
    // We need to :
    //  - Turn 'host' into an IP.
    //  - Get our IP addresses.
    //  - See if any of the IP addresses 'host' has
    //    are in our local set.
    fn is_local(host : &str) -> Result<bool, String> {
        // Get the requested ips:
        // We need a port on the host _I think_:
        let host_and_port = String::from(host) + ":30000";
        let ips = host_and_port.to_socket_addrs();
        if let Err(reason) = ips {
            return Err(format!("Could not resolve URL host to an ip {}", reason));
        }
        let ips = ips.unwrap();
        // get our host's ips:
        
        if let Ok(my_addresses) = local_ip_address::list_afinet_netifas() {
            let mut my_address_list = HashSet::<net::IpAddr>::new();
            for (_, ip) in my_addresses.iter() {
                my_address_list.insert(ip.clone());
            }
            // Now determine if any of ips is in my_addresses:

            for host_ip in ips {
                if my_address_list.contains(&host_ip.ip()) {
                    return Ok(true);
                }
            }
            Ok(false)
        } else {
            Err(String::from("Could not get our local addresses to check if the URI is local"))
        }
    }
    // Check for ring existence in two ways:
    // there emust be a ringbuffer file and
    // it must be registered in the ringmaster:
    //
    fn no_such(local_ring: &str) -> bool {
        let path =ring_path(local_ring);
        if let Ok(e) = fs::exists(&path) {
            if !e {
                return true;
            }
            // Check ringmaster:

            let mut c  = Client::new("localhost");
            let listing = c.list_rings();
            if let Err(_e) = listing {
                false                       // Probably not even a ringmaster.
            } else {
                let list = listing.unwrap();
                let mut ring_list = HashSet::new();
                for ring in list {
                    ring_list.insert(ring.name);
                }
                !ring_list.contains(local_ring)
            }
            
        } else {             // If we can't even ask, then it's not there.
            false
        }
    }
    // Starts the client side of the hoister (stdintoring) as a subprocess.
    // We start it under nohup so it outlives us and 
    // with stdout/stderr pointed to /dev/null so that 
    // it will not make some large messy log file.

    fn start_stdintoring(path : &str, ring: &str, stdin : &mut net::TcpStream) -> Result<String, String> {
        
        // For now we only care about the linux-like case:
        
        let command = Command::new("sh")
            .stdin(unsafe {Stdio::from_raw_fd(stdin.try_clone().unwrap().into_raw_fd())})
            .arg("nohup")
            .arg(path)
            .arg(ring)
            .arg(">")
            .arg("/dev/null")
            .arg("2>&1").spawn();
            

        if let Err(reason) = command {
             Err(format!("Could not spawn stdintoring: {}", reason))
        } else {
            Ok(String::from(ring))                 // The local ring.
        }

    }
    fn start_hoister(host : &str, remote_ring: &str) -> Result<String, String> {
        // Note if the proxy ring already exists we're fine just returning ok for it.
        // At this point we don't attempt to restart any dead hoister.
        
        // To start the hoister we need to 
        // 1. contact the remote ringmaster.
        // 2. Do a REMOTE request to start the remote side of the hoist.
        // 3. Create/register a proxy ring
        // 4. Start an stdintoring method with stdin the socket we got back
        // from the REMOTE request and destination the ringbuffer.

        // figure out the full path to the stdntoring program.

        let stdintoring =  stdin_to_ring_path();
        if let Err(reason) = stdintoring {
            return Err(format!("Could not find stdintoring: {}", reason));
        }
        let stdintoring = stdintoring.unwrap();

        // do a REMOTE on the remote ringmaster to get the hoist socket.

        let mut c = Client::new(host);             // Remote ringmaster.
        let hoist_socket = c.get_data(remote_ring);
        if let Err(reason) = hoist_socket {
            return Err(format!("Failed to set up remote host hoist: {}", reason));
        }
        let mut hoist_socket = hoist_socket.unwrap();
        // Create a proxy ring:

        let ring_name = proxy_ring_name(host, remote_ring);
        let ring_path = ring_path(&ring_name);
        let ring_size = proxy_ring_size();

        if let Err(reason) = ringbuffer::RingBufferMap::create(&ring_path, ring_size) {
            return Err(format!("Could not create proxy ring: {}", reason));
        }
        // Register the ring locally:

        let mut local_master = Client::new("localhost");
        if let Err(reason) = local_master.register_ring(&ring_name) {
            return Err(format!("Unable to register proxy ring: {}", reason));
        }

        // Start the local side of the hoister.

        Self::start_stdintoring(&stdintoring, &ring_name, &mut hoist_socket)

    }
    // the api:


    ///
    /// Attaches to the ring specified by the URI ```uri```.
    /// The form of the URI is tcp://host-name/ringname
    /// * If the URI specifies a local ring, the ring must exist
    /// * If the URI is remote, the remote ring master is contacted and a
    /// hoisting pipe is set up to transfer data fromt he remote ring to
    /// a local proxy ring buffer.  The local proxy ring buffer is thn
    /// attached to. 
    /// 
    /// The name, and ringmasterk in the struct will always be for a local
    /// ring and the local ringmaster.  
    /// 
    pub fn attach(uri : &str) -> Result<RingBufferConsumer, String> {
        // Parse the URI:

        let parsed_uri = Url::parse(uri);
        if let Err(e) = parsed_uri {
            return Err(format!("Failed to parse the ring uri '{}' : {}", uri, e));
        }
        let parsed_uri = parsed_uri.unwrap();
        let scheme          = parsed_uri.scheme();   // tcp:.
        let host                  = parsed_uri.host_str();
        let mut ring                  = String::from(parsed_uri.path());

        // THere's a leading / we need to remove.
        let _ = ring.remove(0);
        
        if scheme != "tcp" {
            return Err(String::from("Invalid scheme/protocol for ringbuffer must be tcp:"));
        }
        if host.is_none() {
            return Err(String::from("Ringbuffer URIs' must supply a host."));
        }
        let host = String::from(host.unwrap());                // The host string.

        // If the ring is not local we need to start a hoist pipeline:

        if ! Self::is_local(&host)? {
            let hoist = Self::start_hoister(&host, &ring);
            if let Err(s) = hoist {
                return Err(format!("Failed to start hoister for {} : {}", uri, s));
            }
            ring = hoist.unwrap();
        }
        // Ring is now a local ringbffer so we can attach as a consumer and 
        // off we go.. THe ring file might still not exist if it was local:

        if Self::no_such(&ring) {
            return Err(format!("There is no such local ring: {}", ring));
        }
        // Create the consumer, register it and make the Ok struct.

        let path = ring_path(&ring);             // FUll path to shm
        let map = ringbuffer::RingBufferMap::new(&path)?;
        let ringbuffer = ringbuffer::ThreadSafeRingBuffer::new(sync::Mutex::new(map));
        let consumer = ringbuffer::consumer::Consumer::attach(&ringbuffer);
        if let Err(e) = consumer {
            return Err(format!("Failed to attach as consumer to {} : {:?}", uri, e));
        }

        let consumer = consumer.unwrap();

        // Register:

        let mut c = Client::new("localhost");
        c.connect_as_consumer(&ring, consumer.get_index(), "")?;

        Ok(RingBufferConsumer {
            ringmaster: c,
            name : ring.clone(),
            consumer,
        })

    }
}
//////////////////////////////// Tests /////////////////////////////
///  Note it is issential, to prevent test failures inthe client tests
/// that these run with it 
/// 
/// cargo test -- --test-threads=1

#[cfg(test)]
mod list_parse_tests {
    use super::*;
    // Tests for parse_tcl_list

    #[test] 
        fn simple_list() {
        let list = "a b cd";
        let parse = parse_tcl_list(list);
        assert_eq!(3, parse.len());
        assert_eq!(vec!["a", "b", "cd"], parse);
    }
    #[test]
    fn empty_list() {
        let list="";
        let parse = parse_tcl_list(list);
        assert_eq!(0, parse.len());
    }
    #[test] 
    fn nested_list() {
        let list="a b {cd ef}";
        let parse = parse_tcl_list(list);
        assert_eq!(3, parse.len());
        assert_eq!(vec!["a", "b", "cd ef"], parse);
        let sublist = parse_tcl_list(&parse[2]);
        assert_eq!(2, sublist.len());
        assert_eq!(vec!["cd", "ef"], sublist);

    }
    #[test]
    fn nested_nested() {
        let list = "a b {c d {e f}}";
        let outer = parse_tcl_list(list);
        assert_eq!(3, outer.len());
        assert_eq!(vec!["a", "b", "c d {e f}"], outer);
        let inner=parse_tcl_list(&outer[2]);
        assert_eq!(3, inner.len());
        assert_eq!(vec!["c", "d", "e f"], inner);
        let innermost = parse_tcl_list(&inner[2]);
        assert_eq!(2, innermost.len());
        assert_eq!(vec!["e", "f"], innermost);
    }
}

#[cfg(test)]
mod client_tests {
    // Prerequisites for these tests to work
    // is a running ringmaster and no existing ringbuffers.
    use super::*;
    use nscldaq_ringbuffer::ringbuffer::RingBufferMap;
    use std::collections::HashSet;
    use std::sync::Mutex;
    use std::time::Duration;
    
    fn ring_name(base_name : &str) -> String {
        format!("{}/{}", RINGBUFFER_DIRECTORY, base_name)
    }

    fn create_and_register(path : &str, name: &str) {
        let mut c = Client::new("localhost");
        let ring_size : u32 = 1024*1024;
        RingBufferMap::create(path, ring_size).expect("Failed to create ring");
        c.register_ring(name).expect("Failed to register the ring");
    }

    fn kill_ring_pre_unregister(path : &str, name: &str) {
        let mut c = Client::new("localhost");
        let unreg_req = format!("UNREGISTER {} \n", name);

        c.transact_and_analyze(&unreg_req).expect("Unable to unregister a ring manually");
        RingBufferMap::delete(path).expect("Unable to delete ring");
    }
    fn kill_ring(path : &str, name: &str) {
        let mut c = Client::new("localhost");
        c.unregister_ring(name).expect("Unable to unregister ring");
        RingBufferMap::delete(path).expect("Unable to delete ring buffer file");
    }
    #[test]
    fn new() {
        let c = Client::new("localhost");
        assert_eq!("localhost".to_string(), c.host);
        assert!(c.socket.is_none());
    }

    #[test]
    fn host() {
        let c = Client::new("localhost");
        assert_eq!("localhost".to_string(), c.host());
    }
    #[test]
    fn socket_none() {
        let c = Client::new("localhost");
        assert!(c.socket().is_none());
    }
    // Note we can't predict the port of the ring master but we can
    // predict that if it's running we wont' get Err:
    #[test]
    fn get_port() {
        let c = Client::new("localhost");
        let port = c.get_port();
        assert!(port.is_ok());
    }
    #[test]
    fn register_1() {
        // Registering a nonexistent ring fails.

        let mut c  = Client::new("localhost");
        let result = c.register_ring("no-such-ring");
        assert!(result.is_err());
    }
    // Note due to threading we need to keep the ring names unique:
    // or turn off threading in the tests which we may well need anyway.

    #[test]
    fn register_2() {
        // Create and register a ring buffer where the ringmaster looks for it.

        let name = "register2_ring";
        let path = ring_name(name);

        RingBufferMap::create(&path, 1024*1024).expect("register_1 could not make ring");   // small but functional.

        let mut c = Client::new("localhost");
        let result = c.register_ring(name);
        assert!(result.is_ok());
        
        // we actually have to UNREGISTER the ring here....
        kill_ring_pre_unregister(&path, name);
        
    }
    #[test]
    fn register_3() {
        // Registering a registered ring is evidently also ok.

        let name = "register3_ring";
        let path = ring_name(name);

        RingBufferMap::create(&path, 1024*1024).expect("register_1 could not make ring");   // small but functional.

        let mut c = Client::new("localhost");
        c.register_ring(name).expect("Failed initial ring registration.");
        assert!(c.register_ring(name).is_ok());                   // Double registration.

        kill_ring_pre_unregister(&path, name);
        
    }
    #[test]
    fn list_1() {
        // There are no rings list_rings shouild give us an empty vector:

        let mut c = Client::new("localhost");
        let info = c.list_rings();
        assert!(info.is_ok());
        let listing = info.unwrap();
        assert_eq!(0, listing.len());
    }
    #[test]
    fn list_2() {
        // If a make and register a ring I should be able to  list it and it will have no consumers.
        // create makes rings with 100 consumers.

        let ring_size : u32 = 1024*1024;
        let name = "list_2";
        let ring_path = ring_name(name);
        println!("Formatted : '{}'",  ring_path);
        RingBufferMap::create(&ring_path, ring_size).expect("list_2 failed to make ringbuffer");

        let mut c = Client::new("localhost");
        c.register_ring(name).expect("list_2 failed to register ring");

        let list_result = c.list_rings();
        assert!(list_result.is_ok());

        // Unregister and kill the ringfile:

        

        let listing = list_result.unwrap();
        assert_eq!(1, listing.len());

        let ring_info = &listing[0];
        assert_eq!(String::from(name), ring_info.name);
        assert_eq!(ring_size, ring_info.size);
        assert_eq!(ring_size, ring_info.free);   // It's all free.
        assert_eq!(100, ring_info.maxconsumers);
        assert_eq!(-1, ring_info.producer_pid);
        assert_eq!(0, ring_info.max_get);
        assert_eq!(0, ring_info.min_get);

        assert_eq!(0, ring_info.consumers.len());   // No consumers registered.

        kill_ring_pre_unregister(&ring_path, name);
        

    }
    #[test]
    fn list_3() {
        // All of the rings are listed.. in some order.:

        let ring1_name ="list3_1";
        let ring2_name = "list3_2";

        let ring1 = ring_name(ring1_name);
        let ring2 = ring_name(ring2_name);

        // Create and register the rings:

        create_and_register(&ring1, &ring1_name);
        create_and_register(&ring2, &ring2_name);

        // List them:

        let mut c = Client::new("localhost");
        let listing_reply = c.list_rings();

        assert!(listing_reply.is_ok());
        let listing = listing_reply.unwrap();
        assert_eq!(2, listing.len());    // There are two rings.

        // We're just going to check that the two names are there.

        let mut set = HashSet::new();     // Lazy way to check since order of listing is not defined
        set.insert(listing[0].name.clone());
        set.insert(listing[1].name.clone());  

        assert!(set.contains(&String::from(ring1_name)));
        assert!(set.contains(&String::from(ring2_name)));

        // Kill the rings:

        kill_ring_pre_unregister(&ring1, ring1_name);
        kill_ring_pre_unregister(&ring2, ring2_name);

    }
    // any more listing tests require that we be able to connect consumers etc.
    #[test]
    fn unregister_1() {
        // Actualy unregistering a nonexistent ring is not an error
        // I think that's supposed to support two processes doing that nearly simultaneously.

        
        let mut c = Client::new("localhost");
        let response = c.unregister_ring("does_not_exist");
        assert!(response.is_ok());
    }
    #[test]
    fn unregister_2() {
        // Unregistration of a ring removes it from the list:
        let name ="unregister_1";
        let path = ring_name(name);


        create_and_register(&path, name);

        let mut c = Client::new("localhost");
        let response = c.unregister_ring(name);
        assert!(response.is_ok());

        // The list should now be empty:

        assert_eq!(0, c.list_rings().expect("Unable to list rings").len());

        // Delete the ring file:

        RingBufferMap::delete(&path).expect("Unable to delete ring buffer file");
    }
    #[test]
    fn pconnect_1() {
        // Connect as producer but no such ring fails.

        let mut c = Client::new("localhost");
        let response = c.connect_as_producer("aring", "Why not");
        assert!(response.is_err());
    }
    #[test]
    fn pconnect_2() {
        // Cnnect as a producer. .. if the ring exists I can connnect.
        // Note we're assumed to have set the pid in the ringbuffer ourself.

        let name = "pconnect_2";
        let path = ring_name(name);

        create_and_register(&path,name);

        // Get the ring buffer and set our pid as the producer - in the ring.

        let mut ring = RingBufferMap::new(&path)
            .expect("Could not map the ring we made");

        let pid = process::id();
        let setp = ring.set_producer(pid);
        assert!(setp.is_ok());      // There's no prior producer.

        // Connect as consumer.... and test. This holds until we drop:

        {
            let mut c = Client::new("localhost");
            let connect_response = c.connect_as_producer(&name, "It-is-me");
            assert!(connect_response.is_ok());

            // list should show that:

            let list = c.list_rings().expect("Could not list the rings");
            assert_eq!(1, list.len());
            assert_eq!(pid as i32, list[0].producer_pid);
        }                               // Ring master should have disconnected us now.

        let mut c = Client::new("localhost");
        let list = c.list_rings().expect("Unable to get ring list");
        assert_eq!(1, list.len());
        assert_eq!(-1, list[0].producer_pid);

        // Did it do that to the ring too:

        assert_eq!(ringbuffer::UNUSED_ENTRY, ring.producer().get_pid());

        // Kill the ring:

        kill_ring(&path, name);
    }
    #[test]
    fn pconnect_3() {
        // Sneaky test for connecting a non-duplicate pid as producer..

        let name = "pconnect_3";
        let path = ring_name(name);

        create_and_register(&path,name);

        // Get the ring buffer and set our pid as the producer - in the ring.

        let mut ring = RingBufferMap::new(&path)
            .expect("Could not map the ring we made");

        let pid = process::id();
        let setp = ring.set_producer(pid);
        assert!(setp.is_ok());      // There's no prior producer.

        // Connect as consumer.... and test. This holds until we drop:

        {
            let mut c = Client::new("localhost");
            let connect_response = c.connect_as_producer(&name, "It-is-me");
            assert!(connect_response.is_ok());

            let request = format!("CONNECT {{{}}} producer {} \"Testing\" \n", name, pid+1);
            let response = c.transact_and_analyze(&request);
            assert!(response.is_err());
        }                               // Ring master should have disconnected us now.

        let mut c = Client::new("localhost");
        let list = c.list_rings().expect("Unable to get ring list");
        assert_eq!(1, list.len());
        assert_eq!(-1, list[0].producer_pid);

        // Did it do that to the ring too:

        assert_eq!(ringbuffer::UNUSED_ENTRY, ring.producer().get_pid());

        // Kill the ring:

        kill_ring(&path, name);

    }
    #[test]
    fn pdisconnect_1() {
        // Can't disconnect from no such ring:

        let name = "pdisconnect_1";

        let mut c = Client::new("localhost");
        assert!(c.disconnect_producer(name).is_err());


    }
    #[test]
    fn pdisconnect_2() {
        // Can't disconnect if we are not the producer (e.g. no producer in this case).

        let name = "pdisconnect_2";
        let path = ring_name(name);

        create_and_register(&path, name);

        let mut c = Client::new("localhost");
        assert!(c.disconnect_producer(name).is_err());

        kill_ring(&path, name);
    }
    #[test]
    fn pdisconnect_3() {
        // We can disconnect ourself as a producer from the ring if we are connected as a producer.

        let name = "pdisconnect_3";
        let path = ring_name(name);

        create_and_register(&path, name);
        let mut map = 
            RingBufferMap::new(&path).expect("failed to map ring");
        
        map.set_producer(process::id()).expect("failed to set ring producer pid");

        // register and unregister:

        let mut c = Client::new("localhost");
        c.connect_as_producer(name, "A comment").expect("Failed to connect as producer");

        let status = c.disconnect_producer(name);
        assert!(status.is_ok());
        
        // Ok the ringbuffer and the listing should have been upated:

        assert!(map.free_producer(process::id()).is_ok());      // I guess we need to do this...

        let list = c.list_rings().expect("Failed to list ringgs");
        assert_eq!(1, list.len());
        assert_eq!(-1, list[0].producer_pid);

        kill_ring(&path, name);

    }
    #[test]
    fn cconnect_1() {
        // Can't connect to a ring that does not exist:

        let mut c = Client::new("localhost");
        assert!(c.connect_as_consumer("aring", 1, "Comment").is_err());

    }
    #[test]
    fn cconnect_2() {
        // Can connect to a ring that does exist:

        let ring = "cconnect_2";
        let path = ring_name(ring);

        create_and_register(&path, ring);
        
        let map = RingBufferMap::new(&path).expect("failed to map ring");
        let ringbuffer = ringbuffer::ThreadSafeRingBuffer::new(Mutex::new(map));
        let consumer_stat = ringbuffer::consumer::Consumer::attach(&ringbuffer);
        assert!(consumer_stat.is_ok());
        let consumer = consumer_stat.unwrap();
        let idx = consumer.get_index();

        let mut c = Client::new("localhost");
        let status = c.connect_as_consumer(ring, idx, "WAWA");
        assert!(status.is_ok());

        // We should be in the ring's consumer list:

        let list = c.list_rings().expect("Failed to list rings");
        assert_eq!(1, list.len());
        let consumers = &(list[0].consumers);
        assert_eq!(1, consumers.len());                   // There is a consmer.
        assert_eq!(process::id(), consumers[0].pid as u32);

        drop(consumer);                  // Just remove us from consumers.

        kill_ring(&path, ring);    
    }
    #[test]
    fn cconnect_3() {
        // Dropping the connection will also drop us from the LIST.

        let ring = "cconnect_3";
        let path = ring_name(ring);

        create_and_register(&path, ring);

        let map = RingBufferMap::new(&path).expect("failed to map ring");
        let ringbuffer = ringbuffer::ThreadSafeRingBuffer::new(Mutex::new(map));
        let consumer = ringbuffer::consumer::Consumer::attach(&ringbuffer)
            .expect("Failed to get a consumer for the ring");
        


        // do this in a block so it drops the connection:

        {
            let mut c = Client::new("localhost");
            c
                .connect_as_consumer(ring, consumer.get_index(), "Some comment")
                .expect("Failed to register consumer");
        }                // client drops so we should be released.

        let mut c = Client::new("localhost");
        let listing = c.list_rings().expect("Could not list rings");
        assert_eq!(1, listing.len());
        let clients = &listing[0].consumers;
        assert_eq!(0, clients.len());

        drop(consumer);                     // Release the consumer in the ring itself.
        kill_ring(&path, ring);
    }

    #[test]
    fn cdisconnect_1() {
        // Cannot disconnect from a nonexistent ring:
        let mut c = Client::new("localhost");
        assert!(c.disconnect_consumer("Nosuch", 1).is_err());

    }
    #[test]
    fn cdisconnect_2() {
        // cannot disconnect if we are not a consumer:

        let ring = "cdisconnect_2";
        let path = ring_name(ring);

        create_and_register(&path, ring);

        let mut c = Client::new("localhost");
        assert!(c.disconnect_consumer(ring, 1).is_err());

        kill_ring(&path, ring);
    }
    #[test]
    fn cdisconnect_3() {
        // If I am a connected consumer I can disconnect and I disappear from
        // the consumer list:

        let ring = "cdisconnect_3";
        let path = ring_name(ring);

        create_and_register(&path, ring);

        let map = RingBufferMap::new(&path).expect("failed to map ring");
        let ringbuffer = ringbuffer::ThreadSafeRingBuffer::new(Mutex::new(map));
        let consumer = ringbuffer::consumer::Consumer::attach(&ringbuffer)
            .expect("Failed to get a consumer for the ring");
        

        let mut c = Client::new("localhost");
        c.connect_as_consumer(ring, consumer.get_index(), "Junk")
            .expect("Failed to register as conumser");

        // Now disconnect and list:

        let result = c.disconnect_consumer(ring, consumer.get_index());
        assert!(result.is_ok());                           // No error reported.

        let listing = c.list_rings().expect("could not list rings");
        assert_eq!(1, listing.len());
        let clist = &listing[0].consumers;

        assert_eq!(0, clist.len());

        drop(consumer);                    // That's what sets the consumer index back to -1.

        kill_ring(&path, ring);

    }
    #[test]
    fn cdisconnect_4() {
        // If I'm connected as a consumer twice and drop one, I'm still connected
        // on my remainig consumer:

        let ring = "cdisconnect_4";
        let path = ring_name(ring);

        create_and_register(&path, ring);

        // Make two consumers:

        let map = RingBufferMap::new(&path).expect("failed to map ring");
        let ringbuffer = ringbuffer::ThreadSafeRingBuffer::new(Mutex::new(map));
        let consumer1 = ringbuffer::consumer::Consumer::attach(&ringbuffer)
            .expect("Failed to make consumer1 for the ring");
        let consumer2 = ringbuffer::consumer::Consumer::attach(&ringbuffer)
            .expect("Failed to make consumer 2 for the ring");


        // Register both:

        let mut c = Client::new("localhost");
        c.connect_as_consumer(ring, consumer1.get_index(), "Consumer1")
            .expect("Failed to connect consumer 1 in ringmaster");
        c.connect_as_consumer(ring, consumer2.get_index(), "Consumer 2")
            .expect("Failed to register consumer2");

        // Unregister consummer 1:

        c.disconnect_consumer(ring, consumer1.get_index()).expect("Could not unregister consumer1");
        drop(consumer1);                          // In the ring too.

        let list = c.list_rings().expect("Failed to list rings");
        assert_eq!(1, list.len());
        let consumers = &list[0].consumers;
        assert_eq!(1, consumers.len());
        assert_eq!(process::id(), consumers[0].pid);

        // get rid of consumer2:

        c.disconnect_consumer(ring, consumer2.get_index()).expect("Failed to unregister consumer 2");
        drop(consumer2);

        // Should be no consumers:

        let list = c.list_rings().expect("Failed to list rings");
        assert_eq!(1, list.len());
        let consumers = &list[0].consumers;
        assert_eq!(0, consumers.len());

        kill_ring(&path, ring);     // Cleanup.

    }

    #[test]
    fn ring_path_1() {
        // Test ring-path private function.

        let ring = "ring_path_1";
        assert_eq!(ring_name(ring), ring_path(ring));
    }
    #[test]
    fn remote_1() {
        // Can't get_data from no suchy ring.

        let mut c = Client::new("localhost");
        assert!(c.get_data("nosuch").is_err());

    }
    #[test]
    fn remote_2() {
        // Can start getting data from existing ring:

        let ring = "remote_2";
        let path = ring_path(ring);

        create_and_register(&path, ring);
        
        let mut c = Client::new("localhost");
        let status = c.get_data(ring);
        assert!(status.is_ok());

        if let Ok(socket) = status {
            let _ = socket.shutdown(net::Shutdown::Both);             // close the connection..
            drop(socket);                                          // fully.
        }

        kill_ring(&path, ring);

    }
    #[test]
    fn remote_3() {
        // we can get data via the remote link...though it might take a bit of time for it 
        // to flush through.

        let ring = "remote_3";
        let path = ring_path(ring);

        create_and_register(&path, ring);

        // Set ourselves up as a producer.

        let map = RingBufferMap::new(&path).expect("failed to map ring");
        let ringbuffer = ringbuffer::ThreadSafeRingBuffer::new(Mutex::new(map));
        let mut producer = ringbuffer::producer::Producer::attach(&ringbuffer)
            .expect("Could not attach as producer");
        
        let mut c = Client::new("localhost");
        c.connect_as_producer(ring, "Producer")
            .expect("Could not connect as a producer");

        //  get_data from the ring to steup the ring2stdout and pipe to it.
        // up to a millisecond.
        let mut c_data = Client::new("localhost");
        let mut sock = c_data.get_data(ring).expect("Filed to set up socket");
       

        // Send a counting pattern to the ring:

        let data : [u8; 20] = [
            'a' as u8 ; 20
        ];

        producer.write_all(&data).expect("Could not put test data in ring");

        // Timeout for now in case the producer barfed up.

        sock.set_read_timeout(Some(Duration::from_secs(5))).expect("could not set read timout for socket");

        let mut received_data : [u8;18] = [0;18];    // So we know if we get anthing.

        

        // Read the data from the hoister.

        let result = sock.read(&mut received_data);
        assert!(result.is_ok());
        assert_eq!(18, result.unwrap());
        
        
        for (i,b) in received_data.into_iter().enumerate() {
            assert_eq!(data[i], b);
        }

        // Shutdown the socket.

        let _ = sock.shutdown(net::Shutdown::Both);


        c.disconnect_producer(ring).expect("Failed to disconnect producer");
        drop(producer);


        kill_ring(&path, ring);


    }
}
#[cfg(test)]
mod producer_tests {
    use super::*;
    use std::collections::HashSet;

    #[test]
    fn attach_1() {
        // Fails if the ring does not already exist:

        assert!(RingBufferProducer::attach("nosuch").is_err());
    }
    #[test]
    fn attach_2() {
        // works ok if the ring  exists:

        let name ="attach_2";
        let path = ring_path(name);
        RingBufferProducer::make_and_register(&path, name).expect("Failed to make/register the ring");

        let producer_stat = RingBufferProducer::attach(name);
        assert!(producer_stat.is_ok());

        // Ring master knows us:

        let mut c = Client::new("localhost");
        let l = c.list_rings().expect("failed to list rings");

        // Make sure we're in the list:
        let mut names =HashSet::new();
        for ring in l {
            names.insert(ring.name);
        }
        assert!(names.contains(name));

        // Should be able to shove some data into the ring:

        let mut producer = producer_stat.unwrap();
        let data : [u8; 10] = ['b' as u8 ;10];
        assert!(producer.ring.blocking_put(&data).is_ok());

        kill_ring(name).expect("could not kill off ringbuffer");
    }
    
    
}
#[cfg(test)]
mod consumer_tests {
    use super::*;
    use nscldaq_ringbuffer::ringbuffer::RingBufferMap;
    fn create_and_register(path : &str, name: &str) {
        let mut c = Client::new("localhost");
        let ring_size : u32 = 1024*1024;
        RingBufferMap::create(path, ring_size).expect("Failed to create ring");
        c.register_ring(name).expect("Failed to register the ring");
    }

    fn kill_ring(path : &str, name: &str) {
        let mut c = Client::new("localhost");
        c.unregister_ring(name).expect("Unable to unregister ring");
        RingBufferMap::delete(path).expect("Unable to delete ring buffer file");
    }
    
    #[test]
    fn islocal_1() {
        // try the verious local hosts:
        // UNwrap because these _shoul_ all work.
        assert!(RingBufferConsumer::is_local("localhost").unwrap());
        
        assert!(RingBufferConsumer::is_local("127.0.0.1").unwrap());
        assert!(RingBufferConsumer::is_local("::1").unwrap());    // IPV6.

        // gethostname does not aways generate a host name in the loca list of IPS.
        // e.g. 127.0.1.1 what I really wanted was the fqdn.
        // doing a lookup will fail on WSL or virtual machines when we're in a 
        // fake network.
        // assert!(RingBufferConsumer::is_local(&gethostname::gethostname().into_string().unwrap()).unwrap());
    }
    #[test]
    fn islocal_2() {
        assert!(!RingBufferConsumer::is_local("www.google.com").unwrap());    //I'm guessing we're not google.
    }
    #[test]
    fn nosuch_1() {
        // There should be no wuch ring below:

        assert!(RingBufferConsumer::no_such("there_is_no_such_ring"));
    }
    #[test]
    fn nosuch_2() {
        // Make and register a ring.. there should be a ring by that name locally:

        let ring_name = "nosuch_2";
        let ring_path = ring_path(ring_name);
        create_and_register(&ring_path, ring_name);
        
        assert!(!RingBufferConsumer::no_such(ring_name));

        kill_ring(&ring_path, ring_name);
    }
    #[test]
    fn attach_1() {
        // local ring but no such:

        let status = RingBufferConsumer::attach("tcp://localhost/no_such_ring");
        assert!(status.is_err());

    }
    #[test]
    fn attach_2() {
        // Can attach a local ring:

        let ring = "attach_2";
        let ring_path = ring_path(ring);
    
        create_and_register(&ring_path, ring);

        let ring_url = format!("tcp://localhost/{}", ring);

        let status = RingBufferConsumer::attach(&ring_url);
        assert!(status.is_ok());

        // Droppting the status also should destroy the consumer
        // which will disconnect us (I hope).


        drop(status);

        // Cleanup.

        kill_ring(&ring_path, ring);
    }
}