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 "Protocol.h"

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

45)   m_fileProtocol(dirBase.getFile("protocol")),
46)   m_dirDests(dirBase.getSubdir("destinations")),

47)   m_pInStream(NULL),

48)   m_pSock(NULL),
49)   m_haveProtocol(false)

50) {

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

53) 
54)   // get input stream and attach to it
55)   getInStream();
56)   // create and bind socket
57)   createSock();
58) 
59)   // load static destinations

60)   updateDestsFull();

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

71)   while (!m_destList.empty()) {
72)     delete m_destList.back().m_pDest;
73)     m_destList.pop_back();
74)   }

75) 
76)   // destroy socket
77)   destroySock();
78)   // detach from input stream and release it
79)   releaseInStream();
80)   // cancel time callback
81)   m_callMgr.cancelTimeCall(this);
82) }
83) 
84) /// check for update of configuration
85) template<typename ADDR, typename SOCK>
86) void Sender<ADDR, SOCK>::updateConfig()
87) {
88)   // input stream name file was modified -> re-get input stream
89)   if (m_fileInStream.checkModified()) {
90)     releaseInStream();
91)     getInStream();
92)   }
93) 
94)   // bind address file was modified -> re-create socket
95)   if (m_fileBind.checkModified()) {
96)     destroySock();
97)     createSock();
98)   }
99) 

100)   // protocol file was modified -> re-read protocol
101)   if (m_fileProtocol.checkModified())
102)     readProto();
103) 

104)   // static destinations update
105)   // (directory modified -> full, otherwise -> light)

106)   if (m_dirDests.checkModified())
107)     updateDestsFull();

108)   else

109)     updateDestsLight();

110) }
111) 
112) /**
113)  * @brief set current frame
114)  * @param[in] stream stream name

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

116)  */
117) template<typename ADDR, typename SOCK>
118) void Sender<ADDR, SOCK>::setFrame(const std::string &stream,
119)                                   stBlinkenFrame *pFrame)
120) {

121)   // convert new frame to protocol data

122)   frame2data(pFrame, m_data);

123) 
124)   // send new protocol data to all destinations
125)   sendAllProto();

126) 
127)   (void)stream; // unused
128) }
129) 
130) /// callback when requsted time reached
131) template<typename ADDR, typename SOCK>
132) void Sender<ADDR, SOCK>::timeCall()
133) {

134)   // repeat current protocol data to all destinations
135)   sendAllProto();

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

160) /// (re-)read protocol

161) template<typename ADDR, typename SOCK>

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

163) {

164)   std::string strProto;
165)   stBlinkenFrame *pFrame;
166) 
167)   // send "no frame" to all destinations
168)   // (stream with old protocol will stop now)
169)   sendAllNoFrame();
170) 
171)   // clear dynamic destinations
172)   // (they registered with old protocol, which is out of service now)
173)   m_dynDests.clear();
174) 
175)   // clear old frame data and old no frame data
176)   m_noFrameData.clear();
177)   m_data.clear();
178) 
179)   // read new protocol from file
180)   m_haveProtocol = m_fileProtocol.getStr(strProto) &&
181)                    m_protocol.fromStr(strProto);
182) 
183)   // create new no frame protocol data and new protocol data
184)   if (m_haveProtocol) {
185)     frame2data(NULL, m_noFrameData);
186)     pFrame = NULL;
187)     if (m_pInStream)
188)       m_pInStream->getCurFrame(pFrame);
189)     frame2data(pFrame, m_data);

190)   }

191) 
192)   // send current protocol data to all destinations
193)   sendAllProto();

194) }
195) 
196) /// get input stream and attach to it
197) template<typename ADDR, typename SOCK>
198) void Sender<ADDR, SOCK>::getInStream()
199) {
200)   // get input stream
201)   m_fileInStream.getStr(m_nameInStream);
202)   m_pInStream = &m_streamMgr.refStream(m_nameInStream);
203) 
204)   // attach to input stream
205)   if (m_pInStream)
206)     m_pInStream->attach(this);
207) }
208) 
209) /// detach from input stream and release it
210) template<typename ADDR, typename SOCK>
211) void Sender<ADDR, SOCK>::releaseInStream()
212) {
213)   // detach from input stream
214)   if (m_pInStream)
215)     m_pInStream->detach(this);
216) 
217)   // unreference stream
218)   m_pInStream = NULL;
219)   m_streamMgr.unrefStream(m_nameInStream);
220) }
221) 
222) /// create socket and bind it
223) template<typename ADDR, typename SOCK>
224) void Sender<ADDR, SOCK>::createSock()
225) {
226)   std::string strAddr;
227)   ADDR addr;
228) 
229)   // create socket
230)   if (!m_pSock) {
231)     m_pSock = new SOCK();
232)     if (!m_pSock)
233)       return;
234)   }
235) 
236)   // get bind address from bind address setting file
237)   if (!m_fileBind.getStr(strAddr) || !addr.fromStr(strAddr)) {
238)     delete m_pSock;
239)     m_pSock = NULL;
240)     return;
241)   }
242) 
243)   // bind socket
244)   if (!m_pSock->bind(addr)) {
245)     delete m_pSock;
246)     m_pSock = NULL;
247)     return;
248)   }
249) 

250)   // request callback on recepetion

251)   m_callMgr.requestIoReadCall(this, m_pSock);

252) 
253)   // send current protocol data to all destinations
254)   sendAllProto();

255) }
256) 
257) /// destroy socket
258) template<typename ADDR, typename SOCK>
259) void Sender<ADDR, SOCK>::destroySock()
260) {
261)   // send "no frame" to all destinations
262)   // (stream from this socket will stop now)
263)   sendAllNoFrame();
264) 
265)   // clear dynamic destinations

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

267)   m_dynDests.clear();

268) 
269)   // cancel callback request
270)   m_callMgr.cancelIoReadCall(this, m_pSock);
271) 
272)   // destroy socket
273)   if (m_pSock) {
274)     delete m_pSock;
275)     m_pSock = NULL;
276)   }
277) }
278) 
279) /**
280)  * @brief light update of static destinations,
281)  *        i.e. stat all files in current static destination directory
282)  */
283) template<typename ADDR, typename SOCK>

284) void Sender<ADDR, SOCK>::updateDestsLight()

285) {
286)   // walk through all files in static dest dir and check for modification
287)   typename DestList::iterator itDest;

288)   for (itDest = m_destList.begin(); itDest != m_destList.end(); ++itDest)

289)     itDest->m_pDest->updateConfig();
290) }
291) 
292) /**
293)  * @brief full update of static destinations,
294)  *        i.e. scan files in playlist directory
295)  */
296) template<typename ADDR, typename SOCK>

297) void Sender<ADDR, SOCK>::updateDestsFull()

298) {
299)   // get list of subdirs in input directory
300)   typedef std::list<std::string> Subdirlist;
301)   Subdirlist curSubdirs;

302)   m_dirDests.getEntries(Directory::TypeSubdir, curSubdirs);

303) 
304)   // walk through current static destinations and subdir list simultaneously
305)   Subdirlist::const_iterator  itSubdir = curSubdirs.begin();

306)   typename DestList::iterator itDest   = m_destList.begin();
307)   while (itSubdir != curSubdirs.end() || itDest != m_destList.end()) {

308) 
309)     // new static destination inserted

310)     if (itDest == m_destList.end() ||

311)         (itSubdir != curSubdirs.end() && *itSubdir < itDest->m_name)) {
312)       // create static destination object
313)       DestEntry destEntry(*itSubdir);

314)       destEntry.m_pDest = new Dest(*this, m_dirDests.getSubdir(*itSubdir),
315)                                    &m_noFrameData);

316)       if (destEntry.m_pDest)
317)         // insert static destination entry

318)         m_destList.insert(itDest, destEntry);

319)       // advance to next subdir
320)       ++itSubdir;
321)     }
322) 
323)     // static destination removed
324)     else if (itSubdir == curSubdirs.end() || *itSubdir > itDest->m_name) {
325)       // remove static destination
326)       delete itDest->m_pDest;

327)       itDest = m_destList.erase(itDest);

328)       // do not advance to next subdir
329)     }
330) 
331)     // static sestination stayed in input list
332)     else {
333)       // check for update
334)       itDest->m_pDest->updateConfig();
335)       // advance to next file and next entry
336)       ++itSubdir;
337)       ++itDest;
338)     }
339) 
340)   } // while itSubdir itDest
341) }
342) 
343) /// remove timed-out dynamic destinations
344) template<typename ADDR, typename SOCK>
345) void Sender<ADDR, SOCK>::removeTimedOutDynDests()
346) {
347)   Time now, timeout;
348)   typename DynDests::iterator itDyn;
349) 
350)   now = Time::now();
351)   timeout = Time(30);
352) 

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

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

355)       m_dynDests.erase(itDyn++);

356)     else
357)       ++itDyn;
358) }
359) 

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

361) template<typename ADDR, typename SOCK>

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

363) {
364)   // remove timed-out dynamic destinations
365)   removeTimedOutDynDests();
366) 

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

368)   sendDests(&m_data);

369) 
370)   // request time callback in one second
371)   m_callMgr.requestTimeCall(this, Time::now() + Time(1));
372) }
373) 
374) /// send "no frame" to all destinations
375) template<typename ADDR, typename SOCK>
376) void Sender<ADDR, SOCK>::sendAllNoFrame()
377) {
378)   // remove timed-out dynamic destinations
379)   removeTimedOutDynDests();
380) 
381)   // get "no frame" protocol data and send to all static/dynamic destinations

382)   sendDests(&m_noFrameData);

383) 
384)   // request time callback in one second
385)   m_callMgr.requestTimeCall(this, Time::now() + Time(1));
386) }
387) 
388) /**
389)  * @brief send data to static/dynamic destinations
390)  * @param[in] data *pData protocol data to send
391)  */
392) template<typename ADDR, typename SOCK>

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

394) {
395)   // send data to static destinations
396)   typename DestList::const_iterator itDest;

397)   for (itDest = m_destList.begin(); itDest != m_destList.end(); ++itDest)

398)     itDest->m_pDest->setProtoData(pData);
399) 
400)   // send data to all dynamic destinations
401)   typename DynDests::const_iterator itDyn;

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

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

404) }
405) 
406) /**

407)  * @brief send protocol data to address

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

409)  * @param[in] addr address to send to
410)  */
411) template<typename ADDR, typename SOCK>

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

413)                                    const ADDR &addr) const
414) {

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

416)     m_pSock->send(data, addr);
417) }
418) 
419) /**
420)  * @brief convert frame to protocol data

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

422)  * @param[out] data protocol data
423)  */
424) template<typename ADDR, typename SOCK>
425) void Sender<ADDR, SOCK>::frame2data(stBlinkenFrame *pFrame,

426)                                     std::string &data) const

427) {
428)   char buf[65536];
429)   int len;
430) 

431)   // no protocol -> leave with empty data
432)   if (!m_haveProtocol) {
433)     data.clear();
434)     return;
435)   }
436) 

437)   // convert frame to protcol data

438)   if (pFrame)

439)     len = BlinkenFrameToNetwork(pFrame, m_protocol.m_proto, buf, sizeof(buf));

440)   else

441)     len = BlinkenProtoMakePacket(m_protocol.m_proto, BlinkenPacketStreamEnd,

442)                                  buf, sizeof(buf));

443)   if (len < 0)
444)     len = 0;
445)   data.assign(buf, len);
446) }
447) 
448) /// receive data from socket
449) template<typename ADDR, typename SOCK>
450) void Sender<ADDR, SOCK>::receiveFromSock()
451) {
452)   etBlinkenProto proto;
453)   etBlinkenPacket packet;
454)   std::string data;
455)   ADDR addr;
456) 
457)   // make sure socket exists
458)   if (!m_pSock)
459)     return;
460) 
461)   // receive (leave if no reception)
462)   if (!m_pSock->recv(data, addr))
463)     return;
464) 
465)   // detect packet type and protocol
466)   BlinkenProtoDetectPacket(data.c_str(), data.size(), &proto, &packet);
467) 

468)   if (m_haveProtocol && proto == m_protocol.m_proto) {
469)     switch (packet) {
470) 
471)       // request -> add to dynamic destinations and send current frame
472)       case BlinkenPacketRequest:
473)         m_dynDests[addr] = Time::now();
474)         sendProto(m_data, addr);
475)         break;
476) 
477)       // end request -> remove from dynamic destinations
478)       case BlinkenPacketEndRequest:
479)         m_dynDests.erase(addr);
480)         break;
481) 
482)       default:
483)         break;
484) 
485)     } // switch (packet)
486)   } // if (m_haveProtocol ...