|
0
|
1 /*******************************************************************************
|
|
|
2
|
|
|
3 copyright: Copyright (c) 2008 Rick Richardson. All rights reserved
|
|
|
4
|
|
|
5 license: BSD style: $(LICENSE)
|
|
|
6
|
|
|
7 version: Initial release v0.1 : May 2008
|
|
|
8
|
|
|
9 author: Rick Richardson
|
|
|
10
|
|
|
11 *******************************************************************************/
|
|
|
12
|
|
5
|
13 module dreactor.core.Vat;
|
|
0
|
14
|
|
|
15 import tango.io.selector.Selector;
|
|
|
16 import tango.io.selector.model.ISelector;
|
|
|
17 import tango.core.Exception;
|
|
|
18 import tango.core.Thread;
|
|
|
19 import tango.core.Atomic;
|
|
|
20 import tango.util.collection.LinkSeq;
|
|
|
21 import tango.util.log.Log;
|
|
|
22
|
|
3
|
23 import dreactor.transport.AsyncSocketConduit;
|
|
6
|
24 import dreactor.core.Dispatcher;
|
|
3
|
25 import dreactor.util.ThreadSafeQueue;
|
|
0
|
26
|
|
3
|
27 Logger log;
|
|
|
28
|
|
|
29 enum : int {UNREGISTER = -1, REMAIN = 0, REGISTER = 1, REREGISTER = 2};
|
|
0
|
30
|
|
5
|
31 static char[] version_string = "Vat.d 0.1 2008-05-31";
|
|
0
|
32
|
|
5
|
33 class Vat
|
|
0
|
34 {
|
|
|
35 private
|
|
|
36 Thread thread;
|
|
|
37 bool running;
|
|
|
38 Atomic!(int) pending;
|
|
3
|
39
|
|
6
|
40 ThreadSafeQueue!(Dispatcher) freshList;
|
|
|
41 ThreadSafeQueue!(Dispatcher) remList;
|
|
3
|
42 public
|
|
0
|
43 this()
|
|
|
44 {
|
|
6
|
45 freshList = new ThreadSafeQueue!(Dispatcher);
|
|
|
46 remList = new ThreadSafeQueue!(Dispatcher);
|
|
5
|
47 log = Log.lookup("dreactor.core.Vat");
|
|
0
|
48 }
|
|
|
49
|
|
|
50 void run()
|
|
|
51 {
|
|
|
52 running = true;
|
|
|
53 thread = new Thread(&eventLoop);
|
|
|
54 thread.start();
|
|
|
55 }
|
|
|
56
|
|
|
57 void exit()
|
|
|
58 {
|
|
|
59 running = false;
|
|
|
60 }
|
|
|
61
|
|
6
|
62 bool addConnection(Dispatcher handler)
|
|
0
|
63 {
|
|
3
|
64 log.trace("adding handler");
|
|
0
|
65 return freshList.push(handler);
|
|
|
66 }
|
|
3
|
67
|
|
6
|
68 bool remConnection(Dispatcher handler)
|
|
3
|
69 {
|
|
|
70 return remList.push(handler);
|
|
|
71 }
|
|
0
|
72
|
|
|
73 private
|
|
|
74 void eventLoop()
|
|
|
75 {
|
|
|
76 auto selector = new Selector();
|
|
|
77 selector.open();
|
|
|
78 do
|
|
|
79 {
|
|
3
|
80 auto eventCount = selector.select(0.01);
|
|
0
|
81
|
|
|
82 if (eventCount > 0)
|
|
|
83 {
|
|
|
84 // process events
|
|
|
85 foreach (SelectionKey key; selector.selectedSet())
|
|
|
86 {
|
|
|
87 if (key.isReadable())
|
|
|
88 {
|
|
|
89 // incoming data
|
|
3
|
90 log.trace("Read event fired");
|
|
6
|
91 auto conn = cast(Dispatcher) key.attachment;
|
|
|
92 if ( Dispatcher.State.listening == conn.getState() )
|
|
3
|
93 conn.handleConnection(conn.transport, &addConnection);
|
|
|
94 else
|
|
|
95 processReturn(conn.handleIncoming(), selector, conn);
|
|
0
|
96 }
|
|
|
97 else if (key.isWritable())
|
|
|
98 {
|
|
3
|
99 log.trace("Write event fired");
|
|
6
|
100 auto conn = cast(Dispatcher) key.attachment;
|
|
3
|
101 processReturn(conn.handleOutgoing(), selector, conn);
|
|
0
|
102 }
|
|
|
103 else if (key.isHangup())
|
|
|
104 {
|
|
5
|
105 log.trace("Hangup event fired");
|
|
6
|
106 auto conn = cast(Dispatcher) key.attachment;
|
|
3
|
107 processReturn(conn.handleDisconnect(), selector, conn);
|
|
0
|
108 }
|
|
|
109 else if (key.isError() || key.isInvalidHandle())
|
|
|
110 {
|
|
3
|
111 log.trace("Error event fired");
|
|
0
|
112 // error, close connection
|
|
6
|
113 auto conn = cast(Dispatcher) key.attachment;
|
|
3
|
114 conn.handleError(&remConnection);
|
|
0
|
115 }
|
|
|
116 }
|
|
|
117 }
|
|
|
118 else if (eventCount == 0)
|
|
|
119 {
|
|
|
120 /* can't think of anything useful to do here. */
|
|
|
121 }
|
|
|
122 else
|
|
|
123 {
|
|
3
|
124 log.error("Selector.select returned {}", eventCount);
|
|
0
|
125 }
|
|
|
126 //add Conduits to listener
|
|
6
|
127 freshList.processAll( (ref Dispatcher h)
|
|
0
|
128 {
|
|
3
|
129 selector.reregister(h.transport, h.events(), h);
|
|
0
|
130 return 1;
|
|
|
131 });
|
|
6
|
132 remList.processAll( (ref Dispatcher h)
|
|
3
|
133 {
|
|
|
134 selector.unregister(h.transport);
|
|
|
135 return 1;
|
|
|
136 });
|
|
0
|
137
|
|
|
138 } while (running)
|
|
3
|
139
|
|
|
140 }
|
|
|
141
|
|
6
|
142 void processReturn(int result, Selector s, Dispatcher h)
|
|
3
|
143 {
|
|
|
144 switch(result)
|
|
|
145 {
|
|
|
146 case UNREGISTER:
|
|
|
147 s.unregister(h.transport);
|
|
|
148 break;
|
|
|
149 case REMAIN:
|
|
|
150 //this space intentially left blank
|
|
|
151 break;
|
|
|
152 case REGISTER:
|
|
|
153 s.register(h.transport, h.events(), h);
|
|
|
154 break;
|
|
|
155 case REREGISTER:
|
|
|
156 s.reregister(h.transport, h.events(), h);
|
|
|
157 break;
|
|
|
158 default:
|
|
5
|
159 log.error("processReturn: unknown return value");
|
|
3
|
160 }
|
|
0
|
161 }
|
|
|
162 }
|
