tcpclient 0.2.1

Asynchronous tcpclient based on aqueue actor.
Documentation 
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
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105

#![feature(async_closure)]
use tokio::net::{TcpStream, ToSocketAddrs};
use std::io;
use tokio::net::tcp::{OwnedReadHalf, OwnedWriteHalf};
use tokio::io::{AsyncWriteExt, ErrorKind};
use aqueue::Actor;
use std::ops::Deref;
use std::sync::Arc;
use log::*;
use std::error::Error;
use aqueue::AError::{Other, StrErr};
use std::future::Future;

pub struct TcpClient {
    disconnect:bool,
    sender:OwnedWriteHalf
}


impl TcpClient {
    pub async fn connect<T:ToSocketAddrs,F:Future<Output=Result<bool,Box<dyn Error>>>+Send+'static,A:Send+'static>(addr:T, f:impl FnOnce(A,Arc<Actor<TcpClient>>,OwnedReadHalf)->F+Send+'static,token:A) ->io::Result<Arc<Actor<TcpClient>>>{

        let stream= TcpStream::connect(addr).await?;
        let target= stream.peer_addr()?;
        let(reader,sender)= stream.into_split();
        let client=Arc::new(Actor::new(TcpClient {
            disconnect:false,
            sender
        }));

        let read_client=client.clone();
        tokio::spawn(async move{
            let disconnect_client=read_client.clone();
            let need_disconnect=
                match f(token,read_client,reader).await{
                    Ok(disconnect)=>{
                        disconnect
                    },
                    Err(err)=>{
                        error!("reader error:{}",err);
                        true
                    }
                };

            if need_disconnect {
                if let Err(er)= disconnect_client.disconnect().await{
                    error!("disconnect to{} err:{}",target,er);
                }
                else{
                    debug!("disconnect to {}",target)
                }
            }
            else{
                debug!("{} reader is close",target);
            }
        });
        Ok(client)
    }

    pub async fn disconnect(&mut self)->io::Result<()>{
        if !self.disconnect {
            self.sender.shutdown().await?;
            self.disconnect = true;
        }
        Ok(())
    }

    pub async fn send(&mut self, buff:&[u8])->io::Result<usize>{
        if !self.disconnect {
            self.sender.write(buff).await
        }else{
            Err(io::Error::new(ErrorKind::Other,StrErr("Send Error,Disconnect".into())))
        }
    }


}

#[aqueue::aqueue_trait]
pub trait SocketClientTrait{
    async fn send<T:Deref<Target=[u8]>+Send+Sync+'static>(&self,buff:T)->Result<usize,Box<dyn Error>>;
    async fn disconnect(&self)->Result<(),Box<dyn Error>>;
}

#[aqueue::aqueue_trait]
impl SocketClientTrait for Actor<TcpClient>{
    async fn send<T:Deref<Target=[u8]>+Send+Sync+'static>(&self, buff:T)->Result<usize,Box<dyn Error>>{
        Ok(self.inner_call(async move |inner|{
            match inner.get_mut().send(&buff).await {
                Ok(size)=>Ok(size),
                Err(er)=>Err(Other(er.into()))
            }
        }).await?)
    }

    async fn disconnect(&self) ->Result<(),Box<dyn Error>> {
        self.inner_call(async move |inner| {
            match inner.get_mut().disconnect().await {
                Ok(_) => Ok(()),
                Err(er) => Err(Other(er.into()))
            }
        }).await?;
        Ok(())
    }
}