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 "IoCallee.h"
20) #include "Module.h"
21) #include "Sender.h"
22) #include "SenderDest.h"
23) #include "SenderDest_impl.h"
24) #include "SettingFile.h"
25) #include "StreamMgr.h"
26) #include "StreamRecv.h"
27) #include "Time.h"
28) #include "TimeCallee.h"
29) 
30) namespace Blinker {
31) 
32) /**
33)  * @brief constructor
34)  * @param[in] callMgr callback manager
35)  * @param[in] streamMgr stream manager
36)  * @param[in] dirBase base directory
37)  */
38) template<typename ADDR, typename SOCK>
39) Sender<ADDR, SOCK>::Sender(CallMgr &callMgr, StreamMgr &streamMgr,
40)                            const Directory &dirBase):
41)   Module(callMgr, streamMgr, dirBase),
42)   m_fileInStream(dirBase.getFile("instream")),
43)   m_fileBind(dirBase.getFile("bind")),
44)   m_dirDestsBlp(dirBase.getSubdir("blp")),
45)   m_dirDestsEblp(dirBase.getSubdir("eblp")),
46)   m_dirDestsMcuf(dirBase.getSubdir("mcuf")),
47)   m_pInStream(NULL),
48)   m_pSock(NULL)
49) {
50)   // initialize protocol data buffers
51)   noFrame2data(BlinkenProtoBlp, m_noFrameDataBlp);
52)   noFrame2data(BlinkenProtoEblp, m_noFrameDataEblp);
53)   noFrame2data(BlinkenProtoMcuf, m_noFrameDataMcuf);
54)   m_dataBlp = m_noFrameDataBlp;
55)   m_dataEblp = m_noFrameDataEblp;
56)   m_dataMcuf = m_noFrameDataMcuf;
57) 
58)   // get input stream and attach to it
59)   getInStream();
60)   // create and bind socket
61)   createSock();
62) 
63)   // load static destinations
64)   updateDestsFull(m_dirDestsBlp, m_destListBlp, &m_noFrameDataBlp);
65)   updateDestsFull(m_dirDestsEblp, m_destListEblp, &m_noFrameDataEblp);
66)   updateDestsFull(m_dirDestsMcuf, m_destListMcuf, &m_noFrameDataMcuf);
67) }
68) 
69) /// virtual destructor
70) template<typename ADDR, typename SOCK>
71) Sender<ADDR, SOCK>::~Sender()
72) {
73)   // send "no frame" to all destinations
74)   sendAllNoFrame();
75) 
76)   // free static destination lists
77)   freeDestList(m_destListBlp);
78)   freeDestList(m_destListEblp);
79)   freeDestList(m_destListMcuf);
80) 
81)   // destroy socket
82)   destroySock();
83)   // detach from input stream and release it
84)   releaseInStream();
85)   // cancel time callback
86)   m_callMgr.cancelTimeCall(this);
87) }
88) 
89) /// check for update of configuration
90) template<typename ADDR, typename SOCK>
91) void Sender<ADDR, SOCK>::updateConfig()
92) {
93)   // input stream name file was modified -> re-get input stream
94)   if (m_fileInStream.checkModified()) {
95)     releaseInStream();
96)     getInStream();
97)   }
98) 
99)   // bind address file was modified -> re-create socket
100)   if (m_fileBind.checkModified()) {
101)     destroySock();
102)     createSock();
103)   }
104) 
105)   // static destinations update
106)   // (directory modified -> full, otherwise -> light)
107)   if (m_dirDestsBlp.checkModified())
108)     updateDestsFull(m_dirDestsBlp, m_destListBlp, &m_noFrameDataBlp);
109)   else
110)     updateDestsLight(m_destListBlp);
111)   if (m_dirDestsEblp.checkModified())
112)     updateDestsFull(m_dirDestsEblp, m_destListEblp, &m_noFrameDataEblp);
113)   else
114)     updateDestsLight(m_destListEblp);
115)   if (m_dirDestsMcuf.checkModified())
116)     updateDestsFull(m_dirDestsMcuf, m_destListMcuf, &m_noFrameDataMcuf);
117)   else
118)     updateDestsLight(m_destListMcuf);
119) }
120) 
121) /**
122)  * @brief set current frame
123)  * @param[in] stream stream name
124)  * @param[in] pFrame current frame
125)  */
126) template<typename ADDR, typename SOCK>
127) void Sender<ADDR, SOCK>::setFrame(const std::string &stream,
128)                                   stBlinkenFrame *pFrame)
129) {

130)   // convert new frame to protocol data
131)   frame2data(pFrame, BlinkenProtoBlp, m_dataBlp);
132)   frame2data(pFrame, BlinkenProtoEblp, m_dataEblp);
133)   frame2data(pFrame, BlinkenProtoMcuf, m_dataMcuf);
134) 
135)   // send new protocol data to all destinations
136)   sendAllProto();

137) 
138)   (void)stream; // unused
139) }
140) 
141) /**
142)  * @brief set current frame to none
143)  * @param[in] stream stream name
144)  */
145) template<typename ADDR, typename SOCK>
146) void Sender<ADDR, SOCK>::setNoFrame(const std::string &stream)
147) {

148)   // set protocol data
149)   m_dataBlp = m_noFrameDataBlp;
150)   m_dataEblp = m_noFrameDataEblp;
151)   m_dataMcuf = m_noFrameDataMcuf;
152) 
153)   // send new protocol data to all destinations
154)   sendAllProto();

155) 
156)   (void)stream; // unused
157) }
158) 
159) /// callback when requsted time reached
160) template<typename ADDR, typename SOCK>
161) void Sender<ADDR, SOCK>::timeCall()
162) {

163)   // repeat current protocol data to all destinations
164)   sendAllProto();

165) }
166) 
167) /**
168)  * @brief callback when I/O object is readable
169)  * @param[in] io I/O object that is readable
170)  */
171) template<typename ADDR, typename SOCK>
172) void Sender<ADDR, SOCK>::ioReadCall(Io *io)
173) {
174)   // reception on socket
175)   if (io == m_pSock)
176)     receiveFromSock();
177) }
178) 
179) /**
180)  * @brief callback when I/O object is writable
181)  * @param[in] io I/O object that is writable
182)  */
183) template<typename ADDR, typename SOCK>
184) void Sender<ADDR, SOCK>::ioWriteCall(Io *io)
185) {
186)   (void)io; // unused
187) }
188) 
189) /**
190)  * @brief free static destiation list
191)  * @param[in] destList static destination list to free
192)  */
193) template<typename ADDR, typename SOCK>
194) void Sender<ADDR, SOCK>::freeDestList(DestList &destList)
195) {
196)   while (!destList.empty()) {
197)     delete destList.back().m_pDest;
198)     destList.pop_back();
199)   }
200) }
201) 
202) /// get input stream and attach to it
203) template<typename ADDR, typename SOCK>
204) void Sender<ADDR, SOCK>::getInStream()
205) {
206)   // get input stream
207)   m_fileInStream.getStr(m_nameInStream);
208)   m_pInStream = &m_streamMgr.refStream(m_nameInStream);
209) 
210)   // attach to input stream
211)   if (m_pInStream)
212)     m_pInStream->attach(this);
213) }
214) 
215) /// detach from input stream and release it
216) template<typename ADDR, typename SOCK>
217) void Sender<ADDR, SOCK>::releaseInStream()
218) {
219)   // detach from input stream
220)   if (m_pInStream)
221)     m_pInStream->detach(this);
222) 
223)   // unreference stream
224)   m_pInStream = NULL;
225)   m_streamMgr.unrefStream(m_nameInStream);
226) }
227) 
228) /// create socket and bind it
229) template<typename ADDR, typename SOCK>
230) void Sender<ADDR, SOCK>::createSock()
231) {
232)   std::string strAddr;
233)   ADDR addr;
234) 
235)   // create socket
236)   if (!m_pSock) {
237)     m_pSock = new SOCK();
238)     if (!m_pSock)
239)       return;
240)   }
241) 
242)   // get bind address from bind address setting file
243)   if (!m_fileBind.getStr(strAddr) || !addr.fromStr(strAddr)) {
244)     delete m_pSock;
245)     m_pSock = NULL;
246)     return;
247)   }
248) 
249)   // bind socket
250)   if (!m_pSock->bind(addr)) {
251)     delete m_pSock;
252)     m_pSock = NULL;
253)     return;
254)   }
255) 
256)   // request callback on recpetion
257)   m_callMgr.requestIoReadCall(this, m_pSock);
258) }
259) 
260) /// destroy socket
261) template<typename ADDR, typename SOCK>
262) void Sender<ADDR, SOCK>::destroySock()
263) {
264)   // send "no frame" to all destinations
265)   // (stream from this socket will stop now)
266)   sendAllNoFrame();
267) 
268)   // clear dynamic destinations

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

270)   m_dynDestsBlp.clear();
271)   m_dynDestsEblp.clear();
272)   m_dynDestsMcuf.clear();
273) 
274)   // cancel callback request
275)   m_callMgr.cancelIoReadCall(this, m_pSock);
276) 
277)   // destroy socket
278)   if (m_pSock) {
279)     delete m_pSock;
280)     m_pSock = NULL;
281)   }
282) }
283) 
284) /**
285)  * @brief light update of static destinations,
286)  *        i.e. stat all files in current static destination directory
287)  * @param[in] destList static destinations for one protocol
288)  */
289) template<typename ADDR, typename SOCK>
290) void Sender<ADDR, SOCK>::updateDestsLight(DestList &destList)
291) {
292)   // walk through all files in static dest dir and check for modification
293)   typename DestList::iterator itDest;
294)   for (itDest = destList.begin(); itDest != destList.end(); ++itDest)
295)     itDest->m_pDest->updateConfig();
296) }
297) 
298) /**
299)  * @brief full update of static destinations,
300)  *        i.e. scan files in playlist directory
301)  * @param[in] dirDests static destinations directory for protocol
302)  * @param[in] destList static destinations for protocol
303)  * @param[in] pNoFrameData "no frame" protocaol data
304)  */
305) template<typename ADDR, typename SOCK>
306) void Sender<ADDR, SOCK>::updateDestsFull(Directory &dirDests,
307)                                          DestList &destList,
308)                                          const std::string *pNoFrameData)
309) {
310)   // get list of subdirs in input directory
311)   typedef std::list<std::string> Subdirlist;
312)   Subdirlist curSubdirs;
313)   dirDests.getEntries(Directory::TypeSubdir, curSubdirs);
314) 
315)   // walk through current static destinations and subdir list simultaneously
316)   Subdirlist::const_iterator  itSubdir = curSubdirs.begin();
317)   typename DestList::iterator itDest   = destList.begin();
318)   while (itSubdir != curSubdirs.end() || itDest != destList.end()) {
319) 
320)     // new static destination inserted
321)     if (itDest == destList.end() ||
322)         (itSubdir != curSubdirs.end() && *itSubdir < itDest->m_name)) {
323)       // create static destination object
324)       DestEntry destEntry(*itSubdir);
325)       destEntry.m_pDest = new Dest(*this, dirDests.getSubdir(*itSubdir),
326)                                    pNoFrameData);
327)       if (destEntry.m_pDest)
328)         // insert static destination entry
329)         destList.insert(itDest, destEntry);
330)       // advance to next subdir
331)       ++itSubdir;
332)     }
333) 
334)     // static destination removed
335)     else if (itSubdir == curSubdirs.end() || *itSubdir > itDest->m_name) {
336)       // remove static destination
337)       delete itDest->m_pDest;
338)       itDest = destList.erase(itDest);
339)       // do not advance to next subdir
340)     }
341) 
342)     // static sestination stayed in input list
343)     else {
344)       // check for update
345)       itDest->m_pDest->updateConfig();
346)       // advance to next file and next entry
347)       ++itSubdir;
348)       ++itDest;
349)     }
350) 
351)   } // while itSubdir itDest
352) }
353) 
354) /// remove timed-out dynamic destinations
355) template<typename ADDR, typename SOCK>
356) void Sender<ADDR, SOCK>::removeTimedOutDynDests()
357) {
358)   Time now, timeout;
359)   typename DynDests::iterator itDyn;
360) 
361)   now = Time::now();
362)   timeout = Time(30);
363) 
364)   for (itDyn = m_dynDestsBlp.begin(); itDyn != m_dynDestsBlp.end(); )
365)     if (itDyn->second + timeout < now)
366)       m_dynDestsBlp.erase(itDyn++);
367)     else
368)       ++itDyn;
369)   for (itDyn = m_dynDestsEblp.begin(); itDyn != m_dynDestsEblp.end(); )
370)     if (itDyn->second + timeout < now)
371)       m_dynDestsEblp.erase(itDyn++);
372)     else
373)       ++itDyn;
374)   for (itDyn = m_dynDestsMcuf.begin(); itDyn != m_dynDestsMcuf.end(); )
375)     if (itDyn->second + timeout < now)
376)       m_dynDestsMcuf.erase(itDyn++);
377)     else
378)       ++itDyn;
379) }
380) 

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

382) template<typename ADDR, typename SOCK>

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

384) {
385)   // remove timed-out dynamic destinations
386)   removeTimedOutDynDests();
387) 

388)   // send current protocol data to all static/dynamic destinations
389)   sendDests(&m_dataBlp, m_destListBlp, m_dynDestsBlp);
390)   sendDests(&m_dataEblp, m_destListEblp, m_dynDestsEblp);
391)   sendDests(&m_dataMcuf, m_destListMcuf, m_dynDestsMcuf);

392) 
393)   // request time callback in one second
394)   m_callMgr.requestTimeCall(this, Time::now() + Time(1));
395) }
396) 
397) /// send "no frame" to all destinations
398) template<typename ADDR, typename SOCK>
399) void Sender<ADDR, SOCK>::sendAllNoFrame()
400) {
401)   // remove timed-out dynamic destinations
402)   removeTimedOutDynDests();
403) 
404)   // get "no frame" protocol data and send to all static/dynamic destinations

405)   sendDests(&m_noFrameDataBlp, m_destListBlp, m_dynDestsBlp);
406)   sendDests(&m_noFrameDataEblp, m_destListEblp, m_dynDestsEblp);
407)   sendDests(&m_noFrameDataMcuf, m_destListMcuf, m_dynDestsMcuf);

408) 
409)   // request time callback in one second
410)   m_callMgr.requestTimeCall(this, Time::now() + Time(1));
411) }
412) 
413) /**
414)  * @brief send data to static/dynamic destinations
415)  * @param[in] data *pData protocol data to send
416)  * @param[in] destList static destinations
417)  * @param[in] dynDests dynamic destinations
418)  */
419) template<typename ADDR, typename SOCK>
420) void Sender<ADDR, SOCK>::sendDests(const std::string *pData,
421)                                    const DestList destList,
422)                                    const DynDests dynDests)
423) {
424)   // send data to static destinations
425)   typename DestList::const_iterator itDest;
426)   for (itDest = destList.begin(); itDest != destList.end(); ++itDest)
427)     itDest->m_pDest->setProtoData(pData);
428) 
429)   // send data to all dynamic destinations
430)   typename DynDests::const_iterator itDyn;
431)   for (itDyn = dynDests.begin(); itDyn != dynDests.end(); ++itDyn)

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

433) }
434) 
435) /**

436)  * @brief send protocol data to address

437)  * @param[in] data protocol data of frame
438)  * @param[in] addr address to send to
439)  */
440) template<typename ADDR, typename SOCK>

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

442)                                    const ADDR &addr) const
443) {
444)   if (m_pSock) {
445)     m_pSock->send(data, addr);
446)   }
447) }
448) 
449) /**
450)  * @brief convert frame to protocol data
451)  * @param[in] pFrame frame
452)  * @param[in] proto Blinken protocol identifier
453)  * @param[out] data protocol data
454)  */
455) template<typename ADDR, typename SOCK>
456) void Sender<ADDR, SOCK>::frame2data(stBlinkenFrame *pFrame,
457)                                     etBlinkenProto proto, std::string &data)
458) {
459)   char buf[65536];
460)   int len;
461) 
462)   // convert frame to protcol data
463)   len = BlinkenFrameToNetwork(pFrame, proto, buf, sizeof(buf));
464)   if (len < 0)
465)     len = 0;
466)   data.assign(buf, len);
467) }
468) 
469) /**
470)  * @brief get "no frame" protocol data
471)  * @param[in] proto Blinken protocol identifier
472)  * @param[out] data protcol data
473)  */
474) template<typename ADDR, typename SOCK>
475) void Sender<ADDR, SOCK>::noFrame2data(etBlinkenProto proto, std::string &data)
476) {
477)   char buf[16];
478)   int len;
479) 
480)   // obtain "no frame" protcol data
481)   len = BlinkenProtoMakePacket(proto, BlinkenPacketStreamEnd,
482)         buf, sizeof(buf));
483)   if (len < 0)
484)     len = 0;
485)   data.assign(buf, len);
486) }
487) 
488) /// receive data from socket
489) template<typename ADDR, typename SOCK>
490) void Sender<ADDR, SOCK>::receiveFromSock()
491) {
492)   etBlinkenProto proto;
493)   etBlinkenPacket packet;
494)   std::string data;
495)   ADDR addr;
496) 
497)   // make sure socket exists
498)   if (!m_pSock)
499)     return;
500) 
501)   // receive (leave if no reception)
502)   if (!m_pSock->recv(data, addr))
503)     return;
504) 
505)   // detect packet type and protocol
506)   BlinkenProtoDetectPacket(data.c_str(), data.size(), &proto, &packet);
507) 
508)   switch (packet) {
509) 
510)     // request -> add to dynamic destinations and send current frame
511)     case BlinkenPacketRequest:
512)       switch (proto) {
513)         case BlinkenProtoBlp:
514)           m_dynDestsBlp[addr] = Time::now();

515)           sendProto(m_dataBlp, addr);

516)           break;
517)         case BlinkenProtoEblp:
518)           m_dynDestsEblp[addr] = Time::now();

519)           sendProto(m_dataEblp, addr);

520)           break;
521)         case BlinkenProtoMcuf:
522)           m_dynDestsMcuf[addr] = Time::now();

523)           sendProto(m_dataMcuf, addr);