Mercurial > projects > dreactor

comparison dreactor/core/Vat.d @ 6:287ba7de97c4

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
more housekeeping
author rick@minifunk
date Tue, 08 Jul 2008 12:23:26 -0400
parents f875a1f278b8
children 60cf25102fb2
comparison
equal deleted inserted replaced
5:f875a1f278b8 6:287ba7de97c4
19 import tango.core.Atomic; 19 import tango.core.Atomic;
20 import tango.util.collection.LinkSeq; 20 import tango.util.collection.LinkSeq;
21 import tango.util.log.Log; 21 import tango.util.log.Log;
22 22
23 import dreactor.transport.AsyncSocketConduit; 23 import dreactor.transport.AsyncSocketConduit;
24 import dreactor.core.ConnectionHandler; 24 import dreactor.core.Dispatcher;
25 import dreactor.util.ThreadSafeQueue; 25 import dreactor.util.ThreadSafeQueue;
26 26
27 Logger log; 27 Logger log;
28 28
29 enum : int {UNREGISTER = -1, REMAIN = 0, REGISTER = 1, REREGISTER = 2}; 29 enum : int {UNREGISTER = -1, REMAIN = 0, REGISTER = 1, REREGISTER = 2};
35 private 35 private
36 Thread thread; 36 Thread thread;
37 bool running; 37 bool running;
38 Atomic!(int) pending; 38 Atomic!(int) pending;
39 39
40 ThreadSafeQueue!(ConnectionHandler) freshList; 40 ThreadSafeQueue!(Dispatcher) freshList;
41 ThreadSafeQueue!(ConnectionHandler) remList; 41 ThreadSafeQueue!(Dispatcher) remList;
42 public 42 public
43 this() 43 this()
44 { 44 {
45 freshList = new ThreadSafeQueue!(ConnectionHandler); 45 freshList = new ThreadSafeQueue!(Dispatcher);
46 remList = new ThreadSafeQueue!(ConnectionHandler); 46 remList = new ThreadSafeQueue!(Dispatcher);
47 log = Log.lookup("dreactor.core.Vat"); 47 log = Log.lookup("dreactor.core.Vat");
48 } 48 }
49 49
50 void run() 50 void run()
51 { 51 {
57 void exit() 57 void exit()
58 { 58 {
59 running = false; 59 running = false;
60 } 60 }
61 61
62 bool addConnection(ConnectionHandler handler) 62 bool addConnection(Dispatcher handler)
63 { 63 {
64 log.trace("adding handler"); 64 log.trace("adding handler");
65 return freshList.push(handler); 65 return freshList.push(handler);
66 } 66 }
67 67
68 bool remConnection(ConnectionHandler handler) 68 bool remConnection(Dispatcher handler)
69 { 69 {
70 return remList.push(handler); 70 return remList.push(handler);
71 } 71 }
72 72
73 private 73 private
86 { 86 {
87 if (key.isReadable()) 87 if (key.isReadable())
88 { 88 {
89 // incoming data 89 // incoming data
90 log.trace("Read event fired"); 90 log.trace("Read event fired");
91 auto conn = cast(ConnectionHandler) key.attachment; 91 auto conn = cast(Dispatcher) key.attachment;
92 if ( ConnectionHandler.State.listening == conn.getState() ) 92 if ( Dispatcher.State.listening == conn.getState() )
93 conn.handleConnection(conn.transport, &addConnection); 93 conn.handleConnection(conn.transport, &addConnection);
94 else 94 else
95 processReturn(conn.handleIncoming(), selector, conn); 95 processReturn(conn.handleIncoming(), selector, conn);
96 } 96 }
97 else if (key.isWritable()) 97 else if (key.isWritable())
98 { 98 {
99 log.trace("Write event fired"); 99 log.trace("Write event fired");
100 auto conn = cast(ConnectionHandler) key.attachment; 100 auto conn = cast(Dispatcher) key.attachment;
101 processReturn(conn.handleOutgoing(), selector, conn); 101 processReturn(conn.handleOutgoing(), selector, conn);
102 } 102 }
103 else if (key.isHangup()) 103 else if (key.isHangup())
104 { 104 {
105 log.trace("Hangup event fired"); 105 log.trace("Hangup event fired");
106 auto conn = cast(ConnectionHandler) key.attachment; 106 auto conn = cast(Dispatcher) key.attachment;
107 processReturn(conn.handleDisconnect(), selector, conn); 107 processReturn(conn.handleDisconnect(), selector, conn);
108 } 108 }
109 else if (key.isError() || key.isInvalidHandle()) 109 else if (key.isError() || key.isInvalidHandle())
110 { 110 {
111 log.trace("Error event fired"); 111 log.trace("Error event fired");
112 // error, close connection 112 // error, close connection
113 auto conn = cast(ConnectionHandler) key.attachment; 113 auto conn = cast(Dispatcher) key.attachment;
114 conn.handleError(&remConnection); 114 conn.handleError(&remConnection);
115 } 115 }
116 } 116 }
117 } 117 }
118 else if (eventCount == 0) 118 else if (eventCount == 0)
122 else 122 else
123 { 123 {
124 log.error("Selector.select returned {}", eventCount); 124 log.error("Selector.select returned {}", eventCount);
125 } 125 }
126 //add Conduits to listener 126 //add Conduits to listener
127 freshList.processAll( (ref ConnectionHandler h) 127 freshList.processAll( (ref Dispatcher h)
128 { 128 {
129 selector.reregister(h.transport, h.events(), h); 129 selector.reregister(h.transport, h.events(), h);
130 return 1; 130 return 1;
131 }); 131 });
132 remList.processAll( (ref ConnectionHandler h) 132 remList.processAll( (ref Dispatcher h)
133 { 133 {
134 selector.unregister(h.transport); 134 selector.unregister(h.transport);
135 return 1; 135 return 1;
136 }); 136 });
137 137
138 } while (running) 138 } while (running)
139 139
140 } 140 }
141 141
142 void processReturn(int result, Selector s, ConnectionHandler h) 142 void processReturn(int result, Selector s, Dispatcher h)
143 { 143 {
144 switch(result) 144 switch(result)
145 { 145 {
146 case UNREGISTER: 146 case UNREGISTER:
147 s.unregister(h.transport); 147 s.unregister(h.transport);