Blinker

1) /* Blinker
2)    Copyright 2011 Stefan Schuermans <stefan@blinkenarea.org>
3)    Copyleft GNU public license - http://www.gnu.org/copyleft/gpl.html
4)    a blinkenarea.org project */
5) 
6) #ifndef SENDER_IMPL_H
7) #define SENDER_IMPL_H
8) 
9) #include <list>
10) #include <map>
11) #include <string>
12) 
13) #include <BlinkenLib/BlinkenProto.h>
14) #include <BlinkenLib/BlinkenFrame.h>
15) 
16) #include "CallMgr.h"
17) #include "Directory.h"
18) #include "File.h"

19) #include "InStreamFile.h"

20) #include "IoCallee.h"

21) #include "ListTracker.h"
22) #include "ListTracker_impl.h"

23) #include "Module.h"

24) #include "Protocol.h"

25) #include "ProtocolFile.h"

26) #include "Sender.h"
27) #include "SenderDest.h"
28) #include "SenderDest_impl.h"
29) #include "SettingFile.h"
30) #include "StreamMgr.h"
31) #include "StreamRecv.h"
32) #include "Time.h"
33) #include "TimeCallee.h"
34) 
35) namespace Blinker {
36) 
37) /**
38)  * @brief constructor
39)  * @param[in] callMgr callback manager
40)  * @param[in] streamMgr stream manager
41)  * @param[in] dirBase base directory
42)  */
43) template<typename ADDR, typename SOCK>
44) Sender<ADDR, SOCK>::Sender(CallMgr &callMgr, StreamMgr &streamMgr,
45)                            const Directory &dirBase):
46)   Module(callMgr, streamMgr, dirBase),

47)   m_fileInStream(dirBase.getFile("instream"), streamMgr),

48)   m_fileBind(dirBase.getFile("bind")),

49)   m_fileProtocol(dirBase.getFile("protocol")),

50)   m_pSock(NULL),
51)   m_destListTracker(*this, dirBase.getSubdir("destinations"))

52) {

53)   // read protocol
54)   readProto();

55) 

56)   // attac to input stream
57)   m_fileInStream.setStreamRecv(this);

58)   // create and bind socket
59)   createSock();
60) 
61)   // load static destinations

62)   m_destListTracker.init();

63) }
64) 
65) /// virtual destructor
66) template<typename ADDR, typename SOCK>
67) Sender<ADDR, SOCK>::~Sender()
68) {
69)   // send "no frame" to all destinations
70)   sendAllNoFrame();
71) 

72)   // free static destinations
73)   m_destListTracker.clear();

74) 
75)   // destroy socket
76)   destroySock();

77)   // detach from input stream
78)   m_fileInStream.setStreamRecv(NULL);

79)   // cancel time callback
80)   m_callMgr.cancelTimeCall(this);
81) }
82) 
83) /// check for update of configuration
84) template<typename ADDR, typename SOCK>
85) void Sender<ADDR, SOCK>::updateConfig()
86) {
87)   // input stream name file was modified -> re-get input stream

88)   if (m_fileInStream.checkModified())
89)     m_fileInStream.update();

90) 
91)   // bind address file was modified -> re-create socket
92)   if (m_fileBind.checkModified()) {
93)     destroySock();
94)     createSock();
95)   }
96) 

97)   // protocol file was modified -> re-read protocol
98)   if (m_fileProtocol.checkModified())
99)     readProto();
100) 

101)   // static destinations update

102)   m_destListTracker.updateConfig();

103) }
104) 
105) /**
106)  * @brief set current frame
107)  * @param[in] stream stream name

108)  * @param[in] pFrame current frame (NULL for none)

109)  */
110) template<typename ADDR, typename SOCK>
111) void Sender<ADDR, SOCK>::setFrame(const std::string &stream,
112)                                   stBlinkenFrame *pFrame)
113) {

114)   // convert new frame to protocol data

115)   frame2data(pFrame, m_data);

116) 
117)   // send new protocol data to all destinations
118)   sendAllProto();

119) 
120)   (void)stream; // unused
121) }
122) 

123) /// callback when requested time reached

124) template<typename ADDR, typename SOCK>
125) void Sender<ADDR, SOCK>::timeCall()
126) {

127)   // repeat current protocol data to all destinations
128)   sendAllProto();

129) }
130) 
131) /**
132)  * @brief callback when I/O object is readable
133)  * @param[in] io I/O object that is readable
134)  */
135) template<typename ADDR, typename SOCK>
136) void Sender<ADDR, SOCK>::ioReadCall(Io *io)
137) {
138)   // reception on socket
139)   if (io == m_pSock)
140)     receiveFromSock();
141) }
142) 
143) /**
144)  * @brief callback when I/O object is writable
145)  * @param[in] io I/O object that is writable
146)  */
147) template<typename ADDR, typename SOCK>
148) void Sender<ADDR, SOCK>::ioWriteCall(Io *io)
149) {
150)   (void)io; // unused
151) }
152) 

153) /// (re-)read protocol

154) template<typename ADDR, typename SOCK>

155) void Sender<ADDR, SOCK>::readProto()

156) {

157)   // send "no frame" to all destinations
158)   // (stream with old protocol will stop now)
159)   sendAllNoFrame();
160) 
161)   // clear dynamic destinations
162)   // (they registered with old protocol, which is out of service now)
163)   m_dynDests.clear();
164) 
165)   // clear old frame data and old no frame data
166)   m_noFrameData.clear();
167)   m_data.clear();
168) 
169)   // read new protocol from file

170)   m_fileProtocol.update();

171) 
172)   // create new no frame protocol data and new protocol data

173)   frame2data(NULL, m_noFrameData);
174)   frame2data(m_fileInStream.getCurFrame(), m_data);

175) 
176)   // send current protocol data to all destinations
177)   sendAllProto();

178) }
179) 
180) /// create socket and bind it
181) template<typename ADDR, typename SOCK>
182) void Sender<ADDR, SOCK>::createSock()
183) {
184)   // create socket
185)   if (!m_pSock) {
186)     m_pSock = new SOCK();
187)     if (!m_pSock)
188)       return;
189)   }
190) 
191)   // get bind address from bind address setting file

192)   m_fileBind.update();
193)   if (!m_fileBind.m_valid) {

194)     delete m_pSock;
195)     m_pSock = NULL;
196)     return;
197)   }
198) 
199)   // bind socket

200)   if (!m_pSock->bind(m_fileBind.m_obj)) {

201)     delete m_pSock;
202)     m_pSock = NULL;
203)     return;
204)   }
205) 

206)   // request callback on recepetion

207)   m_callMgr.requestIoReadCall(this, m_pSock);

208) 
209)   // send current protocol data to all destinations
210)   sendAllProto();

211) }
212) 
213) /// destroy socket
214) template<typename ADDR, typename SOCK>
215) void Sender<ADDR, SOCK>::destroySock()
216) {
217)   // send "no frame" to all destinations
218)   // (stream from this socket will stop now)
219)   sendAllNoFrame();
220) 
221)   // clear dynamic destinations

222)   // (they registered with this socket and this socket is gone)

223)   m_dynDests.clear();

224) 
225)   // cancel callback request
226)   m_callMgr.cancelIoReadCall(this, m_pSock);
227) 
228)   // destroy socket
229)   if (m_pSock) {
230)     delete m_pSock;
231)     m_pSock = NULL;
232)   }
233) }
234) 
235) /// remove timed-out dynamic destinations
236) template<typename ADDR, typename SOCK>
237) void Sender<ADDR, SOCK>::removeTimedOutDynDests()
238) {
239)   Time now, timeout;
240)   typename DynDests::iterator itDyn;
241) 
242)   now = Time::now();
243)   timeout = Time(30);
244) 

245)   for (itDyn = m_dynDests.begin(); itDyn != m_dynDests.end(); )

246)     if (itDyn->second + timeout < now)

247)       m_dynDests.erase(itDyn++);

248)     else
249)       ++itDyn;
250) }
251) 

252) /// send current protocol data to all destinations

253) template<typename ADDR, typename SOCK>

254) void Sender<ADDR, SOCK>::sendAllProto()

255) {
256)   // remove timed-out dynamic destinations
257)   removeTimedOutDynDests();
258) 

259)   // send current protocol data to all static/dynamic destinations

260)   sendDests(&m_data);

261) 
262)   // request time callback in one second
263)   m_callMgr.requestTimeCall(this, Time::now() + Time(1));
264) }
265) 
266) /// send "no frame" to all destinations
267) template<typename ADDR, typename SOCK>
268) void Sender<ADDR, SOCK>::sendAllNoFrame()
269) {
270)   // remove timed-out dynamic destinations
271)   removeTimedOutDynDests();
272) 
273)   // get "no frame" protocol data and send to all static/dynamic destinations

274)   sendDests(&m_noFrameData);

275) 
276)   // request time callback in one second
277)   m_callMgr.requestTimeCall(this, Time::now() + Time(1));
278) }
279) 
280) /**
281)  * @brief send data to static/dynamic destinations
282)  * @param[in] data *pData protocol data to send
283)  */
284) template<typename ADDR, typename SOCK>

285) void Sender<ADDR, SOCK>::sendDests(const std::string *pData)

286) {
287)   // send data to static destinations

288)   typename DestListTracker::ListIt itDest;
289)   for (itDest = m_destListTracker.m_list.begin();
290)        itDest != m_destListTracker.m_list.end(); ++itDest)
291)     itDest->m_pObj->setProtoData(pData);

292) 
293)   // send data to all dynamic destinations
294)   typename DynDests::const_iterator itDyn;

295)   for (itDyn = m_dynDests.begin(); itDyn != m_dynDests.end(); ++itDyn)

296)     sendProto(*pData, itDyn->first);

297) }
298) 
299) /**

300)  * @brief send protocol data to address

301)  * @param[in] data protocol data of frame (empty if unknown)

302)  * @param[in] addr address to send to
303)  */
304) template<typename ADDR, typename SOCK>

305) void Sender<ADDR, SOCK>::sendProto(const std::string &data,

306)                                    const ADDR &addr) const
307) {

308)   if (m_pSock && !data.empty())

309)     m_pSock->send(data, addr);
310) }
311) 
312) /**
313)  * @brief convert frame to protocol data

314)  * @param[in] pFrame frame (NULL for none)

315)  * @param[out] data protocol data
316)  */
317) template<typename ADDR, typename SOCK>
318) void Sender<ADDR, SOCK>::frame2data(stBlinkenFrame *pFrame,

319)                                     std::string &data) const

320) {

321)   etBlinkenProto proto;

322)   char buf[65536];
323)   int len;
324) 

325)   // no protocol -> leave with empty data

326)   if (!m_fileProtocol.m_valid) {

327)     data.clear();
328)     return;
329)   }

330)   proto = m_fileProtocol.m_obj.m_proto;

331) 

332)   // convert frame to protcol data

333)   if (pFrame)

334)     len = BlinkenFrameToNetwork(pFrame, proto, buf, sizeof(buf));

335)   else

336)     len = BlinkenProtoMakePacket(proto, BlinkenPacketStreamEnd,

337)                                  buf, sizeof(buf));

338)   if (len < 0)
339)     len = 0;
340)   data.assign(buf, len);
341) }
342) 
343) /// receive data from socket
344) template<typename ADDR, typename SOCK>
345) void Sender<ADDR, SOCK>::receiveFromSock()
346) {
347)   etBlinkenProto proto;
348)   etBlinkenPacket packet;
349)   std::string data;
350)   ADDR addr;
351) 
352)   // make sure socket exists
353)   if (!m_pSock)
354)     return;
355) 
356)   // receive (leave if no reception)
357)   if (!m_pSock->recv(data, addr))
358)     return;
359) 
360)   // detect packet type and protocol
361)   BlinkenProtoDetectPacket(data.c_str(), data.size(), &proto, &packet);
362) 

363)   if (m_fileProtocol.m_valid && proto == m_fileProtocol.m_obj.m_proto) {

364)     switch (packet) {
365) 
366)       // request -> add to dynamic destinations and send current frame
367)       case BlinkenPacketRequest:
368)         m_dynDests[addr] = Time::now();
369)         sendProto(m_data, addr);
370)         break;
371) 
372)       // end request -> remove from dynamic destinations
373)       case BlinkenPacketEndRequest:
374)         m_dynDests.erase(addr);
375)         break;
376) 
377)       default:
378)         break;
379) 
380)     } // switch (packet)

381)   } // if (m_fileProtocol.m_valid ...