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 }