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 BLINKER_SENDER_IMPL_H
7) #define BLINKER_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 "Directory.h"
17) #include "File.h"

18) #include "InStreamFile.h"

19) #include "IoCallee.h"

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

22) #include "Mgrs.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 "StreamRecv.h"
31) #include "Time.h"
32) #include "TimeCallee.h"
33) 
34) namespace Blinker {
35) 
36) /**
37)  * @brief constructor

38)  * @param[in] name module name

39)  * @param[in] mgrs managers

40)  * @param[in] dirBase base directory
41)  */
42) template<typename ADDR, typename SOCK>

43) Sender<ADDR, SOCK>::Sender(const std::string &name, Mgrs &mgrs,
44)                            const Directory &dirBase):
45)   Module(name, mgrs, dirBase),

46)   m_fileInStream(dirBase.getFile("instream"), mgrs.m_streamMgr),

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

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

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

51) {

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

54) 

55)   // attach to input stream

56)   m_fileInStream.setStreamRecv(this);

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

61)   m_destListTracker.init();

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

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

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

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

78)   // cancel time callback

79)   m_mgrs.m_callMgr.cancelTimeCall(this);

80) }
81) 
82) /// check for update of configuration
83) template<typename ADDR, typename SOCK>
84) void Sender<ADDR, SOCK>::updateConfig()
85) {
86)   // input stream name file was modified -> re-get input stream

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

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

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

100)   // static destinations update

101)   m_destListTracker.updateConfig();

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

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

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

113)   // convert new frame to protocol data

114)   frame2data(pFrame, m_data);

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

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

122) /// callback when requested time reached

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

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

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

152) /// (re-)read protocol

153) template<typename ADDR, typename SOCK>

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

155) {

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

169)   m_fileProtocol.update();

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

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

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

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

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

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

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

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

205)   // request callback on recepetion

206)   m_mgrs.m_callMgr.requestIoReadCall(this, m_pSock);

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

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

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

222)   m_dynDests.clear();

223) 
224)   // cancel callback request

225)   m_mgrs.m_callMgr.cancelIoReadCall(this, m_pSock);

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

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

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

246)       m_dynDests.erase(itDyn++);

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

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

252) template<typename ADDR, typename SOCK>

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

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

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

259)   sendDests(&m_data);

260) 
261)   // request time callback in one second

262)   m_mgrs.m_callMgr.requestTimeCall(this, Time::now() + Time(1));

263) }
264) 
265) /// send "no frame" to all destinations
266) template<typename ADDR, typename SOCK>
267) void Sender<ADDR, SOCK>::sendAllNoFrame()
268) {
269)   // remove timed-out dynamic destinations
270)   removeTimedOutDynDests();
271) 
272)   // get "no frame" protocol data and send to all static/dynamic destinations

273)   sendDests(&m_noFrameData);

274) 
275)   // request time callback in one second

276)   m_mgrs.m_callMgr.requestTimeCall(this, Time::now() + Time(1));

277) }
278) 
279) /**
280)  * @brief send data to static/dynamic destinations
281)  * @param[in] data *pData protocol data to send
282)  */
283) template<typename ADDR, typename SOCK>

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

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

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

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

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

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

296) }
297) 
298) /**

299)  * @brief send protocol data to address

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

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

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

305)                                    const ADDR &addr) const
306) {

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

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

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

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

318)                                     std::string &data) const

319) {

320)   etBlinkenProto proto;

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

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

325)   if (!m_fileProtocol.m_valid) {

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

329)   proto = m_fileProtocol.m_obj.m_proto;

330) 

331)   // convert frame to protocol data

332)   if (pFrame)

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

334)   else

335)     len = BlinkenProtoMakePacket(proto, BlinkenPacketStreamEnd,

336)                                  buf, sizeof(buf));

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

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

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

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

381) }
382) 
383) } // namespace Blinker
384) 

385) #endif // #ifndef BLINKER_SENDER_H