hunt.io.TcpStream source code

1 /*
2  * Hunt - A refined core library for D programming language.
3  *
4  * Copyright (C) 2018-2019 HuntLabs
5  *
6  * Website: https://www.huntlabs.net/
7  *
8  * Licensed under the Apache-2.0 License.
9  *
10  */
11 
12 module hunt.io.TcpStream;
13 
14 import hunt.io.channel.Common;
15 import hunt.io.TcpStreamOptions;
16 import hunt.io.IoError;
17 
18 import hunt.collection.ByteBuffer;
19 import hunt.collection.BufferUtils;
20 import hunt.event.selector.Selector;
21 import hunt.concurrency.SimpleQueue;
22 import hunt.event;
23 import hunt.logging.ConsoleLogger;
24 import hunt.Functions;
25 
26 import std.exception;
27 import std.format;
28 import std.socket;
29 import std.string;
30 
31 import core.atomic;
32 import core.stdc.errno;
33 import core.thread;
34 import core.time;
35 
36 version (HAVE_EPOLL) {
37     import core.sys.linux.netinet.tcp : TCP_KEEPCNT;
38 }
39 
40 
41 
42 /**
43  *
44  */
45 class TcpStream : AbstractStream {
46     SimpleEventHandler closeHandler;
47     protected shared bool _isConnected; // It's always true for server.
48 
49     private TcpStreamOptions _tcpOption;
50     private int retryCount = 0;
51     private Selector _loop;
52 
53     // for client
54     this(Selector loop, TcpStreamOptions option = null, AddressFamily family = AddressFamily.INET) {
55         if (option is null)
56             _tcpOption = TcpStreamOptions.create();
57         else
58             _tcpOption = option;
59         super(loop, family, _tcpOption.bufferSize);
60         version(HUNT_IO_DEBUG) tracef("buffer size: %d bytes", _tcpOption.bufferSize);
61         this.socket = new Socket(family, SocketType.STREAM, ProtocolType.TCP);
62         _isClient = true;
63         _isConnected = false;
64         _loop = loop;
65     }
66 
67     // for server
68     this(Selector loop, Socket socket, TcpStreamOptions option = null) {
69         if (option is null)
70             _tcpOption = TcpStreamOptions.create();
71         else
72             _tcpOption = option;
73         super(loop, socket.addressFamily, _tcpOption.bufferSize);
74         this.socket = socket;
75         _remoteAddress = socket.remoteAddress();
76         _localAddress = socket.localAddress();
77 
78         _isClient = false;
79         _isConnected = true;
80         _loop = loop;
81         setKeepalive();
82     }
83 
84     void options(TcpStreamOptions option) @property {
85         assert(option !is null);
86         this._tcpOption = option;
87     }
88 
89     TcpStreamOptions options() @property {
90         return this._tcpOption;
91     }
92 
93     override bool isBusy() {
94         return _isWritting;
95     }
96 
97     void connect(string hostname, ushort port) {
98         Address[] addresses = getAddress(hostname, port);
99         if(addresses is null) {
100             throw new SocketException("Can't resolve hostname: " ~ hostname);
101         }
102         Address selectedAddress;
103         foreach(Address addr; addresses) {
104             string ip = addr.toAddrString();
105             if(ip.startsWith("::")) // skip IPV6
106                 continue;
107             if(ip.length <= 16) {
108                 selectedAddress = addr;
109                 break;
110             }
111         }
112 
113         if(selectedAddress is null) {
114             warning("No IPV4 avaliable");
115             selectedAddress = addresses[0];
116         }
117         version(HUNT_IO_DEBUG) {
118             infof("connecting with: hostname=%s, ip=%s, port=%d ", hostname, selectedAddress.toAddrString(), port);
119         }
120         connect(selectedAddress); // always select the first one.
121     }
122 
123     void connect(Address addr) {
124         if (_isConnected)
125             return;
126 
127         _remoteAddress = addr;
128         import std.parallelism;
129 
130         auto connectionTask = task(&doConnect, addr);
131         taskPool.put(connectionTask);
132         // doConnect(addr);
133     }
134 
135     void reconnect() {
136         if (!_isClient) {
137             throw new Exception("Only client can call this method.");
138         }
139 
140         if (_isConnected || retryCount >= _tcpOption.retryTimes)
141             return;
142 
143         retryCount++;
144         _isConnected = false;
145         this.socket = new Socket(this._family, SocketType.STREAM, ProtocolType.TCP);
146 
147         version (HUNT_DEBUG)
148             tracef("reconnecting %d...", retryCount);
149         connect(_remoteAddress);
150     }
151 
152     protected override bool doConnect(Address addr)  {
153         try {
154             version (HUNT_DEBUG)
155                 tracef("connecting to %s...", addr);
156             // Address binded = createAddress(this.socket.addressFamily);
157             // this.socket.bind(binded);
158             this.socket.blocking = true;
159             if(super.doConnect(addr))
160             {
161               this.socket.blocking = false;
162               setKeepalive();
163               _localAddress = this.socket.localAddress();
164               start();
165               _isConnected = true;
166             } else
167             {
168                 errorOccurred(ErrorCode.CONNECTIONEFUSED,"Connection refused");
169                 _isConnected = false;
170             }
171 
172         } catch (SocketOSException ex) {
173             // Must try the best to catch all the exceptions. It's because it will executed in another thread.
174             debug warning(ex.msg);
175             version(HUNT_IO_DEBUG) warning(ex);
176             _isConnected = false;
177         }
178 
179         if (_connectionHandler !is null) {
180             try {
181                 _connectionHandler(_isConnected);
182             } catch(Exception ex) {
183                 debug warning(ex.msg);
184                 version(HUNT_IO_DEBUG) warning(ex);
185             }
186         }
187         return true;
188     }
189 
190     // www.tldp.org/HOWTO/html_single/TCP-Keepalive-HOWTO/
191     // http://www.importnew.com/27624.html
192     private void setKeepalive() {
193         version (HAVE_EPOLL) {
194             if (_tcpOption.isKeepalive) {
195                 this.socket.setKeepAlive(_tcpOption.keepaliveTime, _tcpOption.keepaliveInterval);
196                 this.setOption(SocketOptionLevel.TCP,
197                         cast(SocketOption) TCP_KEEPCNT, _tcpOption.keepaliveProbes);
198                 // version (HUNT_DEBUG) checkKeepAlive();
199             }
200         } else version (HAVE_IOCP) {
201             if (_tcpOption.isKeepalive) {
202                 this.socket.setKeepAlive(_tcpOption.keepaliveTime, _tcpOption.keepaliveInterval);
203                 // this.setOption(SocketOptionLevel.TCP, cast(SocketOption) TCP_KEEPCNT,
204                 //     _tcpOption.keepaliveProbes);
205                 // version (HUNT_DEBUG) checkKeepAlive();
206             }
207         }
208     }
209 
210     version (HUNT_DEBUG) private void checkKeepAlive() {
211         version (HAVE_EPOLL) {
212             int time;
213             int ret1 = getOption(SocketOptionLevel.TCP, cast(SocketOption) TCP_KEEPIDLE, time);
214             tracef("ret=%d, time=%d", ret1, time);
215 
216             int interval;
217             int ret2 = getOption(SocketOptionLevel.TCP, cast(SocketOption) TCP_KEEPINTVL, interval);
218             tracef("ret=%d, interval=%d", ret2, interval);
219 
220             int isKeep;
221             int ret3 = getOption(SocketOptionLevel.SOCKET, SocketOption.KEEPALIVE, isKeep);
222             tracef("ret=%d, keepalive=%s", ret3, isKeep == 1);
223 
224             int probe;
225             int ret4 = getOption(SocketOptionLevel.TCP, cast(SocketOption) TCP_KEEPCNT, probe);
226             tracef("ret=%d, interval=%d", ret4, probe);
227         }
228     }
229 
230     TcpStream connected(ConnectionHandler handler) {
231         _connectionHandler = handler;
232         return this;
233     }
234 
235     TcpStream received(DataReceivedHandler handler) {
236         dataReceivedHandler = handler;
237         return this;
238     }
239 
240     TcpStream writed(SimpleActionHandler handler) {
241         dataWriteDoneHandler = handler;
242         return this;
243     }
244     alias onWritten = writed;
245 
246     TcpStream closed(SimpleEventHandler handler) {
247         closeHandler = handler;
248         return this;
249     }
250 
251     TcpStream disconnected(SimpleEventHandler handler) {
252         disconnectionHandler = handler;
253         return this;
254     }
255 
256     TcpStream error(ErrorEventHandler handler) {
257         errorHandler = handler;
258         return this;
259     }
260 
261     override bool isConnected() nothrow {
262         return _isConnected;
263     }
264 
265     override void start() {
266         if (_isRegistered)
267             return;
268         _inLoop.register(this);
269         _isRegistered = true;
270         version (HAVE_IOCP)
271         {
272             //auto iopc = cast(AbstractSelector)_loop;
273             //iopc.putTast(this);
274            // this.beginRead();
275         }
276     }
277 
278     void write(ByteBuffer buffer) {
279         assert(buffer !is null);
280 
281         if (!_isConnected) {
282             throw new Exception(format("The connection is down! remote: %s",
283                 this.remoteAddress.toString()));
284         }
285 
286         version (HUNT_IO_DEBUG)
287             infof("data buffered (%d bytes): fd=%d", buffer.limit(), this.handle);
288         _isWritting = true;
289         // initializeWriteQueue();
290         _writeQueue.enqueue(buffer);
291         onWrite();
292     }
293 
294     /**
295     */
296     void write(const(ubyte)[] data) {
297         if (data.length == 0 || !_isConnected)
298             return;
299 
300         if (!_isConnected) {
301             throw new Exception("The connection is down!");
302         }
303         version (HAVE_IOCP)
304         {
305             return write(BufferUtils.toBuffer(cast(byte[])data));
306         } else
307         {
308             version (HUNT_IO_DEBUG_MORE) {
309                 infof("%d bytes(fd=%d): %(%02X %)", data.length, this.handle, data[0 .. $]);
310             } else  version (HUNT_IO_DEBUG) {
311                 if (data.length <= 32)
312                     infof("%d bytes(fd=%d): %(%02X %)", data.length, this.handle, data[0 .. $]);
313                 else
314                     infof("%d bytes(fd=%d): %(%02X %)", data.length, this.handle, data[0 .. 32]);
315             }
316 
317             if ((_writeQueue.isEmpty()) && !_isWritting) {
318                 _isWritting = true;
319                 const(ubyte)[] d = data;
320 
321                 while (!isClosing() && !isWriteCancelling && d.length > 0) {
322                     version (HUNT_IO_DEBUG)
323                         infof("to write directly %d bytes, fd=%d", d.length, this.handle);
324                     size_t nBytes = tryWrite(d);
325 
326                     if (nBytes == d.length) {
327                         version (HUNT_IO_DEBUG)
328                             tracef("write all out at once: %d / %d bytes, fd=%d", nBytes, d.length, this.handle);
329                         checkAllWriteDone();
330                         break;
331                     } else if (nBytes > 0) {
332                         version (HUNT_IO_DEBUG)
333                             tracef("write out partly: %d / %d bytes, fd=%d", nBytes, d.length, this.handle);
334                         d = d[nBytes .. $];
335                     } else {
336                         version (HUNT_IO_DEBUG)
337                             warningf("buffering data: %d bytes, fd=%d", d.length, this.handle);
338                         _writeQueue.enqueue(BufferUtils.toBuffer(cast(byte[]) d));
339                         break;
340                     }
341                 }
342             } else {
343                 write(BufferUtils.toBuffer(cast(byte[]) data));
344             }
345         }
346     }
347 
348     void shutdownInput() {
349         this.socket.shutdown(SocketShutdown.RECEIVE);
350     }
351 
352     void shutdownOutput() {
353         this.socket.shutdown(SocketShutdown.SEND);
354     }
355 
356     override protected void onDisconnected() {
357         version(HUNT_DEBUG) {
358             infof("peer disconnected: fd=%d", this.handle);
359         }
360         if (disconnectionHandler !is null)
361             disconnectionHandler();
362 
363         this.close();
364     }
365 
366 protected:
367     bool _isClient;
368     ConnectionHandler _connectionHandler;
369 
370     override void onRead() {
371         version (HUNT_IO_DEBUG)
372             trace("start to read");
373 
374         version (Posix) {
375             while (!_isClosed && !tryRead()) {
376                 version (HUNT_IO_DEBUG)
377                     trace("continue reading...");
378             }
379         } else {
380             if (!_isClosed)
381             {
382                 doRead();
383             }
384 
385         }
386 
387         //if (this.isError) {
388         //    string msg = format("Socket error on read: fd=%d, code=%d, message: %s",
389         //            this.handle, errno, this.errorMessage);
390         //    debug errorf(msg);
391         //    if (!isClosed())
392         //        errorOccurred(msg);
393         //}
394     }
395 
396     override void onClose() {
397         bool lastConnectStatus = _isConnected;
398         super.onClose();
399         if(lastConnectStatus) {
400             version (HUNT_IO_DEBUG) {
401                 if (!_writeQueue.isEmpty) {
402                     warningf("Some data has not been sent yet: fd=%d", this.handle);
403                 }
404                 infof("Closing a connection with: %s, fd=%d", this.remoteAddress, this.handle);
405             }
406 
407             resetWriteStatus();
408             _isConnected = false;
409             version (HUNT_IO_DEBUG)
410                 infof("notifying TCP stream down: fd=%d", this.handle);
411             if (closeHandler !is null)
412                 closeHandler();
413         }
414 
415     }
416 
417 }