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.TcpListener;
13 
14 import hunt.io.TcpStream;
15 import hunt.io.TcpStreamOptions;
16 import hunt.io.IoError;
17 
18 import hunt.event;
19 import hunt.Exceptions;
20 import hunt.Functions;
21 import hunt.logging.ConsoleLogger;
22 import hunt.util.Common;
23 
24 import std.socket;
25 import std.exception;
26 import core.thread;
27 import core.time;
28 
29 
30 alias AcceptEventHandler = void delegate(TcpListener sender, TcpStream stream);
31 alias PeerCreateHandler = TcpStream delegate(TcpListener sender, Socket socket, size_t bufferSize);
32 alias EventErrorHandler = void delegate(IoError error);
33 
34 /**
35 */
36 class TcpListener : AbstractListener {
37     private bool _isSslEnabled = false;
38     private bool _isBlocking = false;
39     private bool _isBinded = false;
40     private TcpStreamOptions _tcpStreamoption;
41     protected EventHandler _shutdownHandler;
42 
43     /// event handlers
44     AcceptEventHandler acceptHandler;
45     SimpleEventHandler closeHandler;
46     PeerCreateHandler peerCreateHandler;
47     EventErrorHandler errorHandler;
48 
49     this(EventLoop loop, AddressFamily family = AddressFamily.INET, size_t bufferSize = 4 * 1024) {
50         _tcpStreamoption = TcpStreamOptions.create();
51         _tcpStreamoption.bufferSize = bufferSize;
52         version (HAVE_IOCP)
53             super(loop, family, bufferSize);
54         else
55             super(loop, family);
56     }
57 
58     TcpListener accepted(AcceptEventHandler handler) {
59         acceptHandler = handler;
60         return this;
61     }
62 
63     TcpListener error(EventErrorHandler handler)
64     {
65         errorHandler = handler;
66         return this;
67     }
68 
69     TcpListener onPeerCreating(PeerCreateHandler handler) {
70         peerCreateHandler = handler;
71         return this;
72     }
73 
74     TcpListener onShutdown(EventHandler handler) {
75         _shutdownHandler = handler;
76         return this;
77     }
78 
79     TcpListener bind(string ip, ushort port) {
80         return bind(parseAddress(ip, port));
81     }
82 
83     TcpListener bind(ushort port) {
84         return bind(createAddress(this.socket.addressFamily, port));
85     }
86 
87     TcpListener bind(Address addr) {
88         try {
89         this.socket.setOption(SocketOptionLevel.SOCKET, SocketOption.REUSEADDR, true);
90         this.socket.bind(addr);
91         this.socket.blocking = _isBlocking;
92         _localAddress = _socket.localAddress();
93         _isBinded = true;
94         } catch (SocketOSException e)
95         {
96             if (errorHandler !is null)
97             {
98                 this.errorHandler(new IoError(ErrorCode.ADDRINUSE , e.msg));
99             }
100         }
101         return this;
102     }
103 
104     Address bindingAddress() {
105         return _localAddress;
106     }
107 
108     void blocking(bool flag) {
109         _isBlocking = flag;
110         // if(_isBinded)
111         this.socket.blocking = flag;
112     }
113 
114     bool blocking() {
115         return _isBlocking;
116     }
117 
118     /**
119      * https://stackoverflow.com/questions/14388706/socket-options-so-reuseaddr-and-so-reuseport-how-do-they-differ-do-they-mean-t
120      * https://www.cnblogs.com/xybaby/p/7341579.html
121      * https://rextester.com/BUAFK86204
122      */
123     TcpListener reusePort(bool flag) {
124         if(_isBinded) {
125             throw new IOException("Must be set before binding.");
126         }
127 
128         version (Posix) {
129             import core.sys.posix.sys.socket;
130 
131             this.socket.setOption(SocketOptionLevel.SOCKET, SocketOption.REUSEADDR, flag);
132             this.socket.setOption(SocketOptionLevel.SOCKET, cast(SocketOption) SO_REUSEPORT, flag);
133         } else version (Windows) {
134             // https://docs.microsoft.com/en-us/windows/win32/winsock/using-so-reuseaddr-and-so-exclusiveaddruse
135             // https://docs.microsoft.com/zh-cn/windows/win32/winsock/so-exclusiveaddruse
136             import core.sys.windows.winsock2;
137             this.socket.setOption(SocketOptionLevel.SOCKET, cast(SocketOption) SO_EXCLUSIVEADDRUSE, !flag);
138         }
139 
140         return this;
141     }
142 
143     TcpListener listen(int backlog) {
144         this.socket.listen(backlog);
145         return this;
146     }
147 
148     override void start() {
149         _inLoop.register(this);
150         _isRegistered = true;
151         version (HAVE_IOCP)
152             this.doAccept();
153     }
154 
155     override void close() {
156         if (closeHandler !is null)
157             closeHandler();
158         else if (_shutdownHandler !is null)
159             _shutdownHandler(this, null);
160         this.onClose();
161     }
162 
163     protected override void onRead() {
164         bool canRead = true;
165         version (HUNT_DEBUG)
166             trace("start to listen");
167         // while(canRead && this.isRegistered) // why??
168         {
169             version (HUNT_DEBUG)
170                 trace("listening...");
171 
172             try
173             {
174               canRead = onAccept((Socket socket) {
175 
176                 version (HUNT_DEBUG) {
177                   infof("new connection from %s, fd=%d",
178                   socket.remoteAddress.toString(), socket.handle());
179                 }
180 
181                 if (acceptHandler !is null) {
182                   TcpStream stream;
183                   if (peerCreateHandler is null) {
184                     stream = new TcpStream(_inLoop, socket, _tcpStreamoption);
185                   }
186                   else
187                     stream = peerCreateHandler(this, socket, _tcpStreamoption.bufferSize);
188 
189                   acceptHandler(this, stream);
190                   stream.start();
191                 }
192               });
193 
194               if (this.isError) {
195                 canRead = false;
196                 hunt.logging.ConsoleLogger.error("listener error: ", this.errorMessage);
197                 this.close();
198               }
199             } catch (SocketOSException e)
200             {
201                 if (errorHandler !is null)
202                 {
203                     errorHandler(new IoError(ErrorCode.OTHER , e.msg));
204                 }
205             }
206         }
207     }
208 }
209 
210 
211 
212 // dfmt off
213 version(linux):
214 // dfmt on
215 static if (CompilerHelper.isLessThan(2078)) {
216     version (X86) {
217         enum SO_REUSEPORT = 15;
218     } else version (X86_64) {
219         enum SO_REUSEPORT = 15;
220     } else version (MIPS32) {
221         enum SO_REUSEPORT = 0x0200;
222     } else version (MIPS64) {
223         enum SO_REUSEPORT = 0x0200;
224     } else version (PPC) {
225         enum SO_REUSEPORT = 15;
226     } else version (PPC64) {
227         enum SO_REUSEPORT = 15;
228     } else version (ARM) {
229         enum SO_REUSEPORT = 15;
230     }
231 
232 }
233 
234 version (AArch64) {
235     enum SO_REUSEPORT = 15;
236 }
237 
238 version(CRuntime_Musl) {
239     enum SO_REUSEPORT = 15;
240 }