Mercurial > projects > dreactor

comparison dreactor/protocol/RawTcp.d @ 4:f8b01c9f7114

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
adding basic protocols
author rick@minifunk
date Tue, 08 Jul 2008 11:22:39 -0400
parents
children f875a1f278b8
comparison
equal deleted inserted replaced
3:e3dbc9208822 4:f8b01c9f7114
1 module dreactor.protocol.RawTcp;
2
3 import tango.io.Conduit;
4 import tango.io.selector.model.ISelector;
5 import tango.net.Socket;
6 import tango.util.collection.CircularSeq;
7 import tango.util.log.Log;
8 import tango.util.log.Config;
9
10 import dreactor.transport.AsyncSocketConduit;
11 import dreactor.core.SelectLoop;
12 import dreactor.core.ConnectionHandler;
13
14 /******************************************************************************
15
16 Basic TCP server or client routines for sending raw data.
17
18 ******************************************************************************/
19 class RawTCPListener
20 {
21 public
22 Logger log;
23 this(ConnectionHandler mgr, SelectLoop sel, IPv4Address addr)
24 {
25 manager = mgr;
26 mgr.events(Event.Read);
27 mgr.setOutgoingHandler(&Handlers.onSend);
28 mgr.setIncomingHandler(&Handlers.onReceive);
29 mgr.setConnectHandler(&accept);
30 mgr.listen(addr);
31
32 sel.addConnection(mgr);
33 select = sel;
34 log = Log.lookup("dreactor.protocol.RawTcpServer");
35 log.info("log initialized");
36 children = new CircularSeq!(ConnectionHandler);
37 }
38
39 int accept(Conduit cond, RegisterD reg)
40 {
41 AsyncSocketConduit newcond = new AsyncSocketConduit;
42 (cast(AsyncSocketConduit)cond).socket().accept(newcond.socket);
43 ConnectionHandler h = ConnectionHandler.New(newcond, manager);
44 h.events(Event.Read);
45 select.addConnection(h);
46 children.append(h);
47 log.info("accepted new connection");
48 return 0;
49 }
50
51 int broadcast(char[] outbuf)
52 {
53 foreach(ConnectionHandler h; children)
54 {
55 if (h.appendOutBuffer(outbuf))
56 {
57 h.addEvent(Event.Write);
58 select.addConnection(h);
59 }
60 }
61 return 0;
62 }
63
64 void close()
65 {
66
67 }
68
69 private
70 ConnectionHandler manager;
71 SelectLoop select;
72 CircularSeq!(ConnectionHandler) children;
73 }
74
75 class RawTCPClient
76 {
77 public
78 Logger log;
79 this(ConnectionHandler mgr, SelectLoop sel, Event evts = Event.Read)
80 {
81 manager = mgr;
82 manager.events(evts);
83 connected = false;
84 mgr.setOutgoingHandler(&Handlers.onSend);
85 mgr.setIncomingHandler(&Handlers.onReceive);
86 select = sel;
87 log = Log.lookup("dreactor.protocol.RawTcpClient");
88 }
89
90 int connect(IPv4Address addr)
91 {
92 (cast(AsyncSocketConduit) manager.transport()).connect(addr);
93 select.addConnection(manager);
94 connected = true;
95 log.info("connected to {}", addr);
96 return 0;
97 }
98
99 /**************************************************************************
100
101 send
102 User-called function to send data to the counterpart at the other
103 end of the connection. This sets up the connection manager to send
104 data as the socket becomes free.
105
106 **************************************************************************/
107 int send(char[] outbuf, IPv4Address addr = null)
108 {
109 log.info("sending buffer: {}", outbuf);
110 if (!connected)
111 {
112 log.info("not connected, connecting");
113 if (addr !is null)
114 {
115 if (0 > connect(addr))
116 {
117 log.error("unable to connect");
118 return -1;
119 }
120 }
121 }
122 if (manager.appendOutBuffer(outbuf))
123 {
124 manager.addEvent(Event.Write);
125 if (!select.addConnection(manager))
126 {
127 log.error("unable to register mgr");
128 }
129 }
130 return 0;
131 }
132
133 private
134 ConnectionHandler manager;
135 SelectLoop select;
136 bool connected;
137 }
138
139
140 /******************************************************************************
141
142 Default Event handlers common to both listener/clients
143
144 ******************************************************************************/
145 class Handlers
146 {
147 /**************************************************************************
148
149 onSend
150 OutgoingHandlerD
151 To be registered as the response to socket writable event.
152 Sends data, returns amount sent. Unregisters Handler for sending
153 if there is no more data left to send.
154
155 ***************************************************************************/
156 public static int onSend(ConnectionHandler h)
157 {
158 Logger log = Log.lookup("Handlers.onSend");
159
160 log.info("top of onSend");
161 char[] outbuf = h.nextBuffer();
162 if (outbuf !is null)
163 {
164 int sent = h.transport.write(outbuf);
165 if (sent > 0)
166 {
167 if (! h.addOffset(sent))
168 {
169 h.remEvent(Event.Write);
170 return REREGISTER;
171 }
172 }
173 else if (sent == AsyncSocketConduit.Eof)
174 {
175 log.error("Select said socket was writable, but sent 0 bytes");
176 }
177 else
178 {
179 log.error("Socket send return ERR");
180 }
181 return REMAIN;
182 }
183 else
184 {
185 h.remEvent(Event.Write);
186 return REREGISTER;
187 }
188 }
189 /**************************************************************************
190
191 receive
192 IncomingHandlerD
193 Default incoming data handler. Should be replaced with something useful.
194
195 **************************************************************************/
196 public static int onReceive(ConnectionHandler h)
197 {
198 Logger log = Log.lookup("Handlers.onReceive");
199
200 char inbuf[8192];
201 int amt;
202 if((amt = h.transport.read(inbuf)) > 0)
203 log.info("Received Buffer: {}", inbuf[0 .. amt]);
204 else
205 log.error("Received no data, err = {}", amt);
206
207 return REMAIN;
208 }
209 }