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

22) #include "Protocol.h"

23) #include "ProtocolFile.h"

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

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

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

47)   m_fileProtocol(dirBase.getFile("protocol")),
48)   m_dirDests(dirBase.getSubdir("destinations")),

49)   m_pSock(NULL)

50) {

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

53) 

54)   // attac to input stream
55)   m_fileInStream.setStreamRecv(this);

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
79)   m_fileInStream.setStreamRecv(NULL);

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)     m_fileInStream.update();

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

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

102)   // static destinations update
103)   // (directory modified -> full, otherwise -> light)

104)   if (m_dirDests.checkModified())
105)     updateDestsFull();

106)   else

107)     updateDestsLight();

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

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

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

119)   // convert new frame to protocol data

120)   frame2data(pFrame, m_data);

121) 
122)   // send new protocol data to all destinations
123)   sendAllProto();

124) 
125)   (void)stream; // unused
126) }
127) 

128) /// callback when requested time reached

129) template<typename ADDR, typename SOCK>
130) void Sender<ADDR, SOCK>::timeCall()
131) {

132)   // repeat current protocol data to all destinations
133)   sendAllProto();

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

158) /// (re-)read protocol

159) template<typename ADDR, typename SOCK>

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

161) {

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

175)   m_fileProtocol.update();

176) 
177)   // create new no frame protocol data and new protocol data

178)   frame2data(NULL, m_noFrameData);
179)   frame2data(m_fileInStream.getCurFrame(), m_data);

180) 
181)   // send current protocol data to all destinations
182)   sendAllProto();

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

197)   m_fileBind.update();
198)   if (!m_fileBind.m_valid) {

199)     delete m_pSock;
200)     m_pSock = NULL;
201)     return;
202)   }
203) 
204)   // bind socket

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

206)     delete m_pSock;
207)     m_pSock = NULL;
208)     return;
209)   }
210) 

211)   // request callback on recepetion

212)   m_callMgr.requestIoReadCall(this, m_pSock);

213) 
214)   // send current protocol data to all destinations
215)   sendAllProto();

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

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

228)   m_dynDests.clear();

229) 
230)   // cancel callback request
231)   m_callMgr.cancelIoReadCall(this, m_pSock);
232) 
233)   // destroy socket
234)   if (m_pSock) {
235)     delete m_pSock;
236)     m_pSock = NULL;
237)   }
238) }
239) 
240) /**
241)  * @brief light update of static destinations,
242)  *        i.e. stat all files in current static destination directory
243)  */
244) template<typename ADDR, typename SOCK>

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

246) {
247)   // walk through all files in static dest dir and check for modification
248)   typename DestList::iterator itDest;

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

250)     itDest->m_pDest->updateConfig();
251) }
252) 
253) /**
254)  * @brief full update of static destinations,
255)  *        i.e. scan files in playlist directory
256)  */
257) template<typename ADDR, typename SOCK>

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

259) {
260)   // get list of subdirs in input directory
261)   typedef std::list<std::string> Subdirlist;
262)   Subdirlist curSubdirs;

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

264) 
265)   // walk through current static destinations and subdir list simultaneously
266)   Subdirlist::const_iterator  itSubdir = curSubdirs.begin();

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

269) 
270)     // new static destination inserted

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

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

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

277)       if (destEntry.m_pDest)
278)         // insert static destination entry

279)         m_destList.insert(itDest, destEntry);

280)       // advance to next subdir
281)       ++itSubdir;
282)     }
283) 
284)     // static destination removed
285)     else if (itSubdir == curSubdirs.end() || *itSubdir > itDest->m_name) {
286)       // remove static destination
287)       delete itDest->m_pDest;

288)       itDest = m_destList.erase(itDest);

289)       // do not advance to next subdir
290)     }
291) 
292)     // static sestination stayed in input list
293)     else {
294)       // check for update
295)       itDest->m_pDest->updateConfig();
296)       // advance to next file and next entry
297)       ++itSubdir;
298)       ++itDest;
299)     }
300) 
301)   } // while itSubdir itDest
302) }
303) 
304) /// remove timed-out dynamic destinations
305) template<typename ADDR, typename SOCK>
306) void Sender<ADDR, SOCK>::removeTimedOutDynDests()
307) {
308)   Time now, timeout;
309)   typename DynDests::iterator itDyn;
310) 
311)   now = Time::now();
312)   timeout = Time(30);
313) 

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

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

316)       m_dynDests.erase(itDyn++);

317)     else
318)       ++itDyn;
319) }
320) 

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

322) template<typename ADDR, typename SOCK>

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

324) {
325)   // remove timed-out dynamic destinations
326)   removeTimedOutDynDests();
327) 

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

329)   sendDests(&m_data);

330) 
331)   // request time callback in one second
332)   m_callMgr.requestTimeCall(this, Time::now() + Time(1));
333) }
334) 
335) /// send "no frame" to all destinations
336) template<typename ADDR, typename SOCK>
337) void Sender<ADDR, SOCK>::sendAllNoFrame()
338) {
339)   // remove timed-out dynamic destinations
340)   removeTimedOutDynDests();
341) 
342)   // get "no frame" protocol data and send to all static/dynamic destinations

343)   sendDests(&m_noFrameData);

344) 
345)   // request time callback in one second
346)   m_callMgr.requestTimeCall(this, Time::now() + Time(1));
347) }
348) 
349) /**
350)  * @brief send data to static/dynamic destinations
351)  * @param[in] data *pData protocol data to send
352)  */
353) template<typename ADDR, typename SOCK>

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

355) {
356)   // send data to static destinations
357)   typename DestList::const_iterator itDest;

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

359)     itDest->m_pDest->setProtoData(pData);
360) 
361)   // send data to all dynamic destinations
362)   typename DynDests::const_iterator itDyn;

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

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

365) }
366) 
367) /**

368)  * @brief send protocol data to address

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

370)  * @param[in] addr address to send to
371)  */
372) template<typename ADDR, typename SOCK>

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

374)                                    const ADDR &addr) const
375) {

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

377)     m_pSock->send(data, addr);
378) }
379) 
380) /**
381)  * @brief convert frame to protocol data

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

383)  * @param[out] data protocol data
384)  */
385) template<typename ADDR, typename SOCK>
386) void Sender<ADDR, SOCK>::frame2data(stBlinkenFrame *pFrame,

387)                                     std::string &data) const

388) {

389)   etBlinkenProto proto;

390)   char buf[65536];
391)   int len;
392) 

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

394)   if (!m_fileProtocol.m_valid) {

395)     data.clear();
396)     return;
397)   }

398)   proto = m_fileProtocol.m_obj.m_proto;

399) 

400)   // convert frame to protcol data

401)   if (pFrame)

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

403)   else

404)     len = BlinkenProtoMakePacket(proto, BlinkenPacketStreamEnd,

405)                                  buf, sizeof(buf));

406)   if (len < 0)
407)     len = 0;
408)   data.assign(buf, len);
409) }
410) 
411) /// receive data from socket
412) template<typename ADDR, typename SOCK>
413) void Sender<ADDR, SOCK>::receiveFromSock()
414) {
415)   etBlinkenProto proto;
416)   etBlinkenPacket packet;
417)   std::string data;
418)   ADDR addr;
419) 
420)   // make sure socket exists
421)   if (!m_pSock)
422)     return;
423) 
424)   // receive (leave if no reception)
425)   if (!m_pSock->recv(data, addr))
426)     return;
427) 
428)   // detect packet type and protocol
429)   BlinkenProtoDetectPacket(data.c_str(), data.size(), &proto, &packet);
430) 

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

432)     switch (packet) {
433) 
434)       // request -> add to dynamic destinations and send current frame
435)       case BlinkenPacketRequest:
436)         m_dynDests[addr] = Time::now();
437)         sendProto(m_data, addr);
438)         break;
439) 
440)       // end request -> remove from dynamic destinations
441)       case BlinkenPacketEndRequest:
442)         m_dynDests.erase(addr);
443)         break;
444) 
445)       default:
446)         break;
447) 
448)     } // switch (packet)

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