dast.async.tcpserver source code

1 module dast.async.tcpserver;
2 
3 import dast.async.core,
4 dast.async.eventloop,
5 dast.async.selector,
6 dast.async.tcpclient,
7 tame.meta;
8 
9 version (Windows) import dast.async.iocp;
10 
11 @safe:
12 
13 alias AcceptHandler = void delegate(Socket socket);
14 
15 /// TCP Server
16 class TCPServer : SocketChannel {
17 	mixin Forward!"_socket";
18 
19 	AcceptHandler onAccept;
20 	SimpleHandler onClosed;
21 
22 	this(EventLoop loop, AddrFamily family = AddrFamily.IPv4) {
23 		super(loop, WT.Accept);
24 		flags |= WF.Read;
25 		socket = tcpSocket(family);
26 	}
27 
28 	override void close() {
29 		super.close();
30 		if (onClosed)
31 			onClosed();
32 	}
33 
34 	void start() {
35 		_loop.register(this);
36 		_isReg = true;
37 		version (Windows)
38 			accept();
39 	}
40 
41 	override void onRead() {
42 		debug (Log)
43 			trace("start listening");
44 		if (!tryAccept((Socket socket) {
45 				debug (Log)
46 					info("new connection from ", socket.remoteAddress, ", fd=", socket.handle);
47 
48 				if (onAccept)
49 					onAccept(socket);
50 				else
51 					new TCPClient(_loop, socket).start();
52 			})) {
53 			close();
54 		}
55 	}
56 
57 private:
58 	bool tryAccept(scope AcceptHandler handler) {
59 		version (Posix) {
60 			import core.sys.posix.sys.socket : accept;
61 
62 			const clientFd = accept(handle, null, null);
63 			if (clientFd < 0)
64 				return false;
65 
66 			debug (Log)
67 				trace("server fd=", handle, ", client fd=", clientFd);
68 
69 			handler(Socket(clientFd, _socket.addressFamily));
70 			return true;
71 		}
72 
73 		version (Windows) {
74 			import core.sys.windows.mswsock;
75 
76 			debug (Log)
77 				trace("server fd=", handle, ", client fd=", _clientSock.handle);
78 			socket_t[1] fd = [handle];
79 			_clientSock.setOption(SocketOptionLevel.SOCKET,
80 				cast(SocketOption)SO_UPDATE_ACCEPT_CONTENT, fd);
81 			handler(_clientSock);
82 
83 			debug (Log)
84 				trace("accept next connection");
85 			return _isReg && accept();
86 		}
87 	}
88 
89 	version (Windows)  : bool accept() @trusted {
90 		_clientSock = tcpSocket(_socket.addressFamily);
91 		//_clientSock = new Socket(WSASocket(_socket.addressFamily, SocketType.STREAM,
92 		//ProtocolType.TCP, null, 0, WSA_FLAG_OVERLAPPED), _socket.addressFamily);
93 		uint dwBytesReceived = void;
94 
95 		debug (Log)
96 			trace("client socket=", _clientSock.handle, ", server socket=", handle);
97 		return !checkErr(AcceptEx(handle, _clientSock.handle, _buf.ptr, 0, size, size,
98 				&dwBytesReceived, &_ctx.overlapped), "accept");
99 	}
100 
101 	mixin checkErr;
102 	enum size = sockaddr_in.sizeof + 16;
103 	Socket _clientSock;
104 	IocpContext _ctx = {operation: IocpOperation.accept};
105 	ubyte[size * 4] _buf;
106 }
107 
108 @property:
109 bool reusePort(Socket socket) {
110 	int result = void;
111 	socket.getOption(SocketOptionLevel.SOCKET, SocketOption.reuseAddr, result);
112 	return result != 0;
113 }
114 
115 bool reusePort(Socket socket, bool enabled) {
116 	socket.setOption(SocketOptionLevel.SOCKET, SocketOption.reuseAddr, enabled);
117 
118 	version (Posix) {
119 		socket.setOption(SocketOptionLevel.SOCKET, SocketOption.reusePort, enabled);
120 	} else version (Windows) {
121 		if (!enabled) {
122 			socket.setOption(SocketOptionLevel.SOCKET, SocketOption.exclusiveAddr, true);
123 		}
124 	}
125 
126 	return enabled;
127 }