yu.asyncsocket.server.connection source code

1 module yu.asyncsocket.server.connection;
2 
3 import yu.timer.timingwheeltimer;
4 import yu.asyncsocket.tcpsocket;
5 import yu.eventloop;
6 import yu.memory.allocator;
7 import yu.task;
8 import yu.exception;
9 
10 @trusted abstract class ServerConnection : IWheelTimer!YuAlloctor {
11     this(TCPSocket socket) {
12         restSocket(socket);
13     }
14 
15     ~this() {
16         if (_socket)
17             yDel(_socket);
18     }
19 
20     final TCPSocket restSocket(TCPSocket socket) {
21         TCPSocket tmp = _socket;
22         if (_socket !is null) {
23             _socket.setCloseCallBack(null);
24             _socket.setReadCallBack(null);
25             _socket = null;
26             _loop = null;
27         }
28         if (socket !is null) {
29             _socket = socket;
30             _loop = socket.eventLoop;
31             _socket.setCloseCallBack(&doClose);
32             _socket.setReadCallBack(&onRead);
33         }
34         return tmp;
35     }
36 
37     final bool isAlive() @trusted {
38         return _socket && _socket.isAlive;
39     }
40 
41     final bool active() @trusted nothrow {
42         if (_socket is null)
43             return false;
44         bool active = false;
45         yuCathException(_socket.start(), active);
46         if (active)
47             onActive();
48         return active;
49     }
50 
51     final void write(const(ubyte)[] data, TCPWriteCallBack cback = null) @trusted {
52         if (_loop.isInLoopThread()) {
53             _postWrite(data, cback);
54         } else {
55             auto task = makeTask(yuAlloctor, &_postWrite, data, cback);
56             task.finishedCall = &_loop.finishDoFreeYuTask;
57             _loop.post(task);
58         }
59     }
60 
61     final void write(TCPWriteBuffer buffer) @trusted
62     {
63         if (_loop.isInLoopThread()) {
64             _postWriteBuffer(buffer);
65         } else {
66             auto task = makeTask(yuAlloctor, &_postWriteBuffer, buffer);
67             task.finishedCall = &_loop.finishDoFreeYuTask;
68             _loop.post(task);
69         }
70     }
71 
72 
73     final void restTimeout() @trusted {
74         if (_loop.isInLoopThread()) {
75             rest();
76         } else {
77             auto task = makeTask(yuAlloctor, &rest, 0);
78             task.finishedCall = &_loop.finishDoFreeYuTask;
79             _loop.post(task);
80         }
81     }
82 
83     pragma(inline) final void close() @trusted {
84         _loop.post(&_postClose);
85     }
86 
87     final @property tcpSocket() @safe {
88         return _socket;
89     }
90 
91     final @property eventLoop() @safe {
92         return _loop;
93     }
94 
95 protected:
96     void onActive() nothrow;
97     void onClose() nothrow;
98     void onRead(in ubyte[] data) nothrow;
99 
100 private:
101     final void _postClose() {
102         if (_socket)
103             _socket.close();
104     }
105 
106     final void _postWriteBuffer(TCPWriteBuffer buffer)
107     {
108         if (_socket) {
109             rest();
110             _socket.write(buffer);
111         } else
112             buffer.doFinish();
113     }
114 
115     final void _postWrite(const(ubyte)[] data, TCPWriteCallBack cback) {
116         if (_socket) {
117             rest();
118             _socket.write(data, cback);
119         } else if (cback)
120             cback(data, 0);
121     }
122 
123     final void doClose() nothrow {
124         stop();
125         onClose();
126     }
127 
128 private:
129     TCPSocket _socket;
130     EventLoop _loop;
131 }