kcpserver 1.1.5

fast kcp server frame.
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
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142

use super::KCPPeer;
use crate::kcp::kcp_module::prelude::Kcp;
use crate::prelude::kcp_module::KcpResult;
use async_lock::MutexGuard;
use futures::future::BoxFuture;
use futures::FutureExt;
use std::pin::Pin;
use std::task::{ready, Context, Poll};

/// kcp 写入器
/// 同时实现了 tokio AsyncWrite 和 futures AsyncWrite
pub struct KcpWriter<'a> {
    /// kcp peer
    peer: &'a KCPPeer,
    /// kcp 状态机
    state: KcpWriterState<'a>,
}

/// 写入器状态机
enum KcpWriterState<'a> {
    /// 开始
    Begin,
    /// 写入状态
    Write(BoxFuture<'a, MutexGuard<'a, Kcp>>),
    /// flush状态
    FlushAsync(BoxFuture<'a, KcpResult<()>>),
    /// 关闭状态
    Close(BoxFuture<'a, ()>),
}

impl<'a> From<&'a KCPPeer> for KcpWriter<'a> {
    fn from(peer: &'a KCPPeer) -> Self {
        Self {
            peer,
            state: KcpWriterState::Begin,
        }
    }
}

impl KcpWriter<'_> {
    /// 实现发送
    #[inline]
    fn poll_send(&mut self, cx: &mut Context<'_>, buf: &[u8]) -> Poll<std::io::Result<usize>> {
        loop {
            match self.state {
                KcpWriterState::Begin => {
                    self.state = KcpWriterState::Write(self.peer.kcp.lock().boxed());
                }
                KcpWriterState::Write(ref mut write_lock) => {
                    let mut kcp = ready!(write_lock.as_mut().poll(cx));
                    self.state = KcpWriterState::Begin;
                    return match kcp.send(buf) {
                        Ok(size) => Ok(size).into(),
                        Err(err) => Err(err.into()).into(),
                    };
                }
                _ => panic!("poll_send KcpWriterState error state"),
            }
        }
    }

    /// 实现 flush
    #[inline]
    fn poll_flush_async(&mut self, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
        loop {
            match self.state {
                KcpWriterState::Begin => {
                    self.state = KcpWriterState::FlushAsync(self.peer.flush().boxed());
                }
                KcpWriterState::FlushAsync(ref mut flush_future) => {
                    let res = ready!(flush_future.as_mut().poll(cx));
                    self.state = KcpWriterState::Begin;
                    return match res {
                        Ok(()) => Ok(()).into(),
                        Err(err) => Err(err.into()).into(),
                    };
                }
                _ => panic!("poll_flush_async KcpWriterState error state"),
            }
        }
    }

    /// 实现close
    #[inline]
    fn poll_close_async(&mut self, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
        loop {
            match self.state {
                KcpWriterState::Begin => {
                    self.state = KcpWriterState::Close(self.peer.close().boxed());
                }
                KcpWriterState::Close(ref mut close_future) => {
                    ready!(close_future.as_mut().poll(cx));
                    self.state = KcpWriterState::Begin;
                    return Ok(()).into();
                }
                _ => panic!("poll_close KcpWriterState error state"),
            }
        }
    }
}

impl<'a> futures::AsyncWrite for KcpWriter<'a> {
    #[inline]
    fn poll_write(
        mut self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        buf: &[u8],
    ) -> Poll<std::io::Result<usize>> {
        self.poll_send(cx, buf)
    }

    #[inline]
    fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
        self.poll_flush_async(cx)
    }

    #[inline]
    fn poll_close(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
        self.poll_close_async(cx)
    }
}

impl<'a> tokio::io::AsyncWrite for KcpWriter<'a> {
    #[inline]
    fn poll_write(
        mut self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        buf: &[u8],
    ) -> Poll<std::io::Result<usize>> {
        self.poll_send(cx, buf)
    }

    #[inline]
    fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
        self.poll_flush_async(cx)
    }

    #[inline]
    fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
        self.poll_close_async(cx)
    }
}