Mercurial > projects > dreactor

comparison dreactor/protocol/RawTcp.d @ 11:5836613d16ac

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
reorg! reorg!
author rick@minifunk
date Tue, 12 Aug 2008 16:59:56 -0400
parents e75a2e506b1d
children
comparison
equal deleted inserted replaced
10:e75a2e506b1d 11:5836613d16ac
17 17
18 ******************************************************************************/ 18 ******************************************************************************/
19 class RawTCPListener 19 class RawTCPListener
20 { 20 {
21 public 21 public
22 this(Dispatcher mgr, Vat sel, IPv4Address addr) 22
23 { 23 this(AsyncSocketConduit cond)
24 manager = mgr; 24 {
25 mgr.events(Event.Read);
26 mgr.setOutgoingHandler(&RawTCPHandler.onSend);
27 mgr.setIncomingHandler(&onReceive);
28 mgr.setConnectHandler(&accept);
29 mgr.setErrorHandler(&RawTCPHandler.onError);
30 mgr.setDisconnectHandler(&RawTCPHandler.onHangup);
31 mgr.listen(addr);
32
33 sel.addConnection(mgr);
34 vat = sel;
35 log = Log.lookup("dreactor.protocol.RawTcpServer"); 25 log = Log.lookup("dreactor.protocol.RawTcpServer");
36 log.info("log initialized"); 26 log.info("log initialized");
37 children = new CircularSeq!(Dispatcher); 27 children = new CircularSeq!(Dispatcher);
38 } 28 }
29
39 this(Vat sel, IPv4Address addr) 30 this(Vat sel, IPv4Address addr)
40 { 31 {
41 AsyncSocketConduit cond = new AsyncSocketConduit; 32 AsyncSocketConduit cond = new AsyncSocketConduit;
42 cond.socket().setAddressReuse(true); 33 cond.socket().setAddressReuse(true);
43 Dispatcher lh = new Dispatcher(cond, true); 34 this(cond);
44 this(lh, sel, addr);
45 } 35 }
46 36
47 ~this() 37 ~this()
48 { 38 {
49 close(); 39 close();
50 } 40 }
51 41
52 int accept(Conduit cond, RegisterD reg) 42 AsyncSocketConduit accept(Conduit cond, RegisterD reg)
53 { 43 {
54 AsyncSocketConduit newcond = new AsyncSocketConduit; 44 AsyncSocketConduit newcond = new AsyncSocketConduit;
55 (cast(AsyncSocketConduit)cond).socket().accept(newcond.socket); 45 (cast(AsyncSocketConduit)cond).socket().accept(newcond.socket);
56 Dispatcher h = Dispatcher.New(newcond, manager);
57 h.events(Event.Read); 46 h.events(Event.Read);
58 vat.addConnection(h); 47 vat.addConnection(h);
59 children.append(h); 48 children.append(h);
60 log.info("accepted new connection"); 49 log.info("accepted new connection");
61 return 0; 50 return newcond;
62 } 51 }
63 52
64 int broadcast(char[] outbuf, Dispatcher[] excluded = null) 53 int broadcast(char[] outbuf, AsyncSocketConduit[] recips)
65 { 54 {
66 foreach(Dispatcher h; children) 55 foreach(AsyncSocketConduit c; recips)
67 { 56 {
68 if (excluded && excluded.includes(h)) 57 if (c.appendOutBuffer(outbuf))
69 continue;
70 if (h.appendOutBuffer(outbuf))
71 { 58 {
72 h.addEvent(Event.Write); 59 h.addEvent(Event.Write);
73 vat.addConnection(h); 60 vat.addConnection(h);
74 } 61 }
75 } 62 }
261 248
262 void setDataHandler(void delegate(char[], Dispatcher) h) 249 void setDataHandler(void delegate(char[], Dispatcher) h)
263 { 250 {
264 dataHandler = h; 251 dataHandler = h;
265 } 252 }
253
266 private 254 private
267 void delegate(char[], Dispatcher) dataHandler; 255 void delegate(char[], Dispatcher) dataHandler;
268 Dispatcher manager; 256 Dispatcher manager;
269 Vat vat; 257 Vat vat;
270 bool connected; 258 bool connected;