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) 
166)   // send "no frame" to all destinations
167)   // (stream with old protocol will stop now)
168)   sendAllNoFrame();
169) 
170)   // clear dynamic destinations
171)   // (they registered with old protocol, which is out of service now)
172)   m_dynDests.clear();
173) 
174)   // clear old frame data and old no frame data
175)   m_noFrameData.clear();
176)   m_data.clear();
177) 
178)   // read new protocol from file
179)   m_haveProtocol = m_fileProtocol.getStr(strProto) &&
180)                    m_protocol.fromStr(strProto);
181) 
182)   // create new no frame protocol data and new protocol data
183)   if (m_haveProtocol) {
184)     frame2data(NULL, m_noFrameData);
185)     if (m_pInStream)

186)       frame2data(m_pInStream->getCurFrame(), m_data);
187)     else
188)       frame2data(NULL, m_data);

189)   }

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

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

249)   // request callback on recepetion

250)   m_callMgr.requestIoReadCall(this, m_pSock);

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

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

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

266)   m_dynDests.clear();

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

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

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

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

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

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

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

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

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

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

307) 
308)     // new static destination inserted

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

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

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

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

317)         m_destList.insert(itDest, destEntry);

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

326)       itDest = m_destList.erase(itDest);

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

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

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

354)       m_dynDests.erase(itDyn++);

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

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

360) template<typename ADDR, typename SOCK>

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

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

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

367)   sendDests(&m_data);

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

381)   sendDests(&m_noFrameData);

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

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

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

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

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

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

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

403) }
404) 
405) /**

406)  * @brief send protocol data to address

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

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

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

412)                                    const ADDR &addr) const
413) {

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

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

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

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

425)                                     std::string &data) const

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

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

436)   // convert frame to protcol data

437)   if (pFrame)

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

439)   else

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

441)                                  buf, sizeof(buf));

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

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