geario.net.channel.AbstractSocketChannel source code

1 module geario.net.channel.AbstractSocketChannel;
2 
3 import geario.event.selector.Selector;
4 import geario.net.channel.AbstractChannel;
5 import geario.net.channel.Types;
6 import geario.logging;
7 
8 import core.time;
9 import std.functional;
10 import std.socket;
11 import core.stdc.stdint;
12 
13 /**
14  * 
15  */
16 abstract class AbstractSocketChannel : AbstractChannel {
17 
18     protected shared bool _isWritting = false; // keep a data write operation atomic
19 
20     protected Socket _socket;
21 
22     this(Selector loop, ChannelType type) {
23         super(loop, type);
24     }
25 
26     // Busy with reading or writting
27     protected bool IsBusy() {
28         return false;
29     }
30 
31     protected @property void socket(Socket s) {
32         this.handle = s.handle();
33         version (Posix) {
34             s.blocking = false;
35         }
36         _socket = s;
37         version (GEAR_DEBUG_MORE)
38             log.info("new socket: fd=%d", this.handle);
39     }
40 
41     @property Socket socket() {
42         return _socket;
43     }
44 
45     override void Close()
46     {
47         // if (_isClosing) {
48         //     // debug log.warn("already closed [fd=%d]", this.handle);
49         //     return;
50         // }
51         // _isClosing = true;
52         version (GEAR_IO_MORE) log.trace("socket channel closing [fd=%d]...", this.handle);
53 
54         version (HAVE_IOCP)
55         {
56             super.Close();
57         }
58         else
59         {
60             if (IsBusy()) {
61                 import std.parallelism;
62 
63                 version (GEAR_DEBUG) log.warn("Close operation delayed");
64 
65                 auto theTask = task(() {
66                     super.Close();
67                     while (IsBusy())
68                     {
69                         version (GEAR_DEBUG)
70                             log.info("waitting for idle [fd=%d]...", this.handle);
71                         // Thread.sleep(20.msecs);
72                     }
73                 });
74 
75                 taskPool.put(theTask);
76             } else {
77                 super.Close();
78             }
79         }
80     }
81 
82     /// Get a socket option.
83     /// Returns: The number of bytes written to $(D result).
84     ///    returns the length, in bytes, of the actual result - very different from getsockopt()
85     pragma(inline) final int getOption(SocketOptionLevel level, SocketOption option, void[] result) @trusted {
86         return this._socket.getOption(level, option, result);
87     }
88 
89     /// Common case of getting integer and boolean options.
90     pragma(inline) final int getOption(SocketOptionLevel level,
91             SocketOption option, ref int32_t result) @trusted {
92         return this._socket.getOption(level, option, result);
93     }
94 
95     /// Get the linger option.
96     pragma(inline) final int getOption(SocketOptionLevel level, SocketOption option,
97             ref Linger result) @trusted {
98         return this._socket.getOption(level, option, result);
99     }
100 
101     /// Get a timeout (duration) option.
102     pragma(inline) final void getOption(SocketOptionLevel level,
103             SocketOption option, ref Duration result) @trusted {
104         this._socket.getOption(level, option, result);
105     }
106 
107     /// Set a socket option.
108     pragma(inline) final void setOption(SocketOptionLevel level, SocketOption option, void[] value) @trusted {
109         this._socket.setOption(forward!(level, option, value));
110     }
111 
112     /// Common case for setting integer and boolean options.
113     pragma(inline) final void setOption(SocketOptionLevel level, SocketOption option, int32_t value) @trusted {
114         this._socket.setOption(forward!(level, option, value));
115     }
116 
117     /// Set the linger option.
118     pragma(inline) final void setOption(SocketOptionLevel level, SocketOption option, Linger value) @trusted {
119         this._socket.setOption(forward!(level, option, value));
120     }
121 
122     pragma(inline) final void setOption(SocketOptionLevel level, SocketOption option, Duration value) @trusted {
123         this._socket.setOption(forward!(level, option, value));
124     }
125 
126     final @trusted Address RemoteAddress() {
127         return _remoteAddress;
128     }
129 
130     protected Address _remoteAddress;
131 
132     final @trusted Address LocalAddress() {
133         return _localAddress;
134     }
135 
136     protected Address _localAddress;
137 
138     version (HAVE_IOCP) {
139         void setRead(size_t bytes) {
140             readLen = bytes;
141         }
142 
143         protected size_t readLen;
144     }
145 
146     void Start();
147 
148     void OnWriteDone()
149     {
150         assert(false, "unimplemented");
151     }
152 }