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] mgrs managers

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

42) Sender<ADDR, SOCK>::Sender(Mgrs &mgrs, const Directory &dirBase):
43)   Module(mgrs, dirBase),
44)   m_fileInStream(dirBase.getFile("instream"), mgrs.m_streamMgr),

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

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

47)   m_pSock(NULL),
48)   m_destListTracker(*this, dirBase.getSubdir("destinations"))

49) {

50)   // read protocol
51)   readProto();

52) 

53)   // attach to input stream

54)   m_fileInStream.setStreamRecv(this);

55)   // create and bind socket
56)   createSock();
57) 
58)   // load static destinations

59)   m_destListTracker.init();

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

69)   // free static destinations
70)   m_destListTracker.clear();

71) 
72)   // destroy socket
73)   destroySock();

74)   // detach from input stream
75)   m_fileInStream.setStreamRecv(NULL);

76)   // cancel time callback

77)   m_mgrs.m_callMgr.cancelTimeCall(this);

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

85)   if (m_fileInStream.checkModified())
86)     m_fileInStream.update();

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

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

98)   // static destinations update

99)   m_destListTracker.updateConfig();

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

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

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

111)   // convert new frame to protocol data

112)   frame2data(pFrame, m_data);

113) 
114)   // send new protocol data to all destinations
115)   sendAllProto();

116) 
117)   (void)stream; // unused
118) }
119) 

120) /// callback when requested time reached

121) template<typename ADDR, typename SOCK>
122) void Sender<ADDR, SOCK>::timeCall()
123) {

124)   // repeat current protocol data to all destinations
125)   sendAllProto();

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

150) /// (re-)read protocol

151) template<typename ADDR, typename SOCK>

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

153) {

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

167)   m_fileProtocol.update();

168) 
169)   // create new no frame protocol data and new protocol data

170)   frame2data(NULL, m_noFrameData);
171)   frame2data(m_fileInStream.getCurFrame(), m_data);

172) 
173)   // send current protocol data to all destinations
174)   sendAllProto();

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

189)   m_fileBind.update();
190)   if (!m_fileBind.m_valid) {

191)     delete m_pSock;
192)     m_pSock = NULL;
193)     return;
194)   }
195) 
196)   // bind socket

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

198)     delete m_pSock;
199)     m_pSock = NULL;
200)     return;
201)   }
202) 

203)   // request callback on recepetion

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

205) 
206)   // send current protocol data to all destinations
207)   sendAllProto();

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

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

220)   m_dynDests.clear();

221) 
222)   // cancel callback request

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

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

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

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

244)       m_dynDests.erase(itDyn++);

245)     else
246)       ++itDyn;
247) }
248) 

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

250) template<typename ADDR, typename SOCK>

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

252) {
253)   // remove timed-out dynamic destinations
254)   removeTimedOutDynDests();
255) 

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

257)   sendDests(&m_data);

258) 
259)   // request time callback in one second

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

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

271)   sendDests(&m_noFrameData);

272) 
273)   // request time callback in one second

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

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

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

283) {
284)   // send data to static destinations

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

289) 
290)   // send data to all dynamic destinations
291)   typename DynDests::const_iterator itDyn;

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

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

294) }
295) 
296) /**

297)  * @brief send protocol data to address

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

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

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

303)                                    const ADDR &addr) const
304) {

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

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

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

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

316)                                     std::string &data) const

317) {

318)   etBlinkenProto proto;

319)   char buf[65536];
320)   int len;
321) 

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

323)   if (!m_fileProtocol.m_valid) {

324)     data.clear();
325)     return;
326)   }

327)   proto = m_fileProtocol.m_obj.m_proto;

328) 

329)   // convert frame to protocol data

330)   if (pFrame)

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

332)   else

333)     len = BlinkenProtoMakePacket(proto, BlinkenPacketStreamEnd,

334)                                  buf, sizeof(buf));

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

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

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

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

379) }
380) 
381) } // namespace Blinker
382) 

383) #endif // #ifndef BLINKER_SENDER_H