src/noarch/Sender.h (fd00bff) - Blinker · BlinkenArea

Sender.h

/* Blinker
   Copyright 2011 Stefan Schuermans <stefan@blinkenarea.org>
   Copyleft GNU public license - http://www.gnu.org/copyleft/gpl.html
   a blinkenarea.org project */

#ifndef SENDER_H
#define SENDER_H

#include <list>
#include <map>
#include <string>

#include <BlinkenLib/BlinkenProto.h>
#include <BlinkenLib/BlinkenFrame.h>

#include "CallMgr.h"
#include "Directory.h"
#include "File.h"
#include "IoCallee.h"
#include "Module.h"
#include "SettingFile.h"
#include "StreamMgr.h"
#include "StreamRecv.h"
#include "Time.h"
#include "TimeCallee.h"

namespace Blinker {

/// stream sender
template<typename ADDR, typename SOCK>
class Sender: public IoCallee, public Module, public StreamRecv,
              public TimeCallee
{
protected:
  /// static destination
  struct Dest {
    std::string m_name; ///< name of static destination
    SettingFile m_file; ///< setting file object
    ADDR        m_addr; ///< parsed address
    Dest(const std::string &name, const File &file); ///< constructor
    bool parse();       ///< parse address from current file
  };

/// static destinations
  typedef std::list<Dest> Dests;

/// dynamic destinations: address -> time of last request
  typedef std::map<ADDR, Time> DynDests;

public:
  /**
   * @brief constructor
   * @param[in] callMgr callback manager
   * @param[in] streamMgr stream manager
   * @param[in] dirBase base directory
   */
  Sender(CallMgr &callMgr, StreamMgr &streamMgr, const Directory &dirBase);

/// virtual destructor
  virtual ~Sender();

private:
  /// copy constructor disabled
  Sender(const Sender &that);

/// assignment operator disabled
  const Sender & operator=(const Sender &that);

public:
  /// check for update of configuration
  virtual void updateConfig();

/**
   * @brief set current frame
   * @param[in] stream stream name
   * @param[in] pFrame current frame
   */
  virtual void setFrame(const std::string &stream, stBlinkenFrame *pFrame);

/**
   * @brief set current frame to none
   * @param[in] stream stream name
   */
  virtual void setNoFrame(const std::string &stream);

/// callback when requsted time reached
  virtual void timeCall();

/**
   * @brief callback when I/O object is readable
   * @param[in] io I/O object that is readable
   */
  virtual void ioReadCall(Io *io);

/**
   * @brief callback when I/O object is writable
   * @param[in] io I/O object that is writable
   */
  virtual void ioWriteCall(Io *io);

protected:
  /// get input stream and attach to it
  void getInStream();

/// detach from input stream and release it
  void releaseInStream();

/// create socket and bind it
  void createSock();

/// destroy socket
  void destroySock();

/**
   * @brief light update of static destinations,
   *        i.e. stat all files in current static destination directory
   * @param[in] dests static destinations for protocol
   * @param[in] proto Blinken protocol identifier
   */
  void updateDestsLight(Dests &dests, etBlinkenProto proto);

/// remove timed-out dynamic destinations
  void removeTimedOutDynDests();

/**
   * @brief send frame to all destinations
   * @param[in] pFrame frame to send
   */
  void sendAllFrame(stBlinkenFrame *pFrame);

/// send "no frame" to all destinations
  void sendAllNoFrame();

/**
   * @brief send data to static/dynamic destinations
   * @param[in] data data to send
   * @param[in] dests static destinations
   * @param[in] dynDests dynamic destinations
   */
  void sendDests(const std::string &data, const Dests dests,
                 const DynDests dynDests);

/**
   * @brief send current frame to address
   * @param[in] addr address to send to
   * @param[in] proto Blinken protocol identifier
   */
  void sendCurFrame(const ADDR &addr, etBlinkenProto proto) const;

/**
   * @brief send "no frame" to address
   * @param[in] addr address to send to
   * @param[in] proto Blinken protocol identifier
   */
  void sendNoFrame(const ADDR &addr, etBlinkenProto proto) const;

/**
   * @brief send frame to address
   * @param[in] data protcol data of frame
   * @param[in] addr address to send to
   */
  void sendFrame(const std::string &data, const ADDR &addr) const;

/**
   * @brief convert frame to protocol data
   * @param[in] pFrame frame
   * @param[in] proto Blinken protocol identifier
   * @param[out] data protcol data
   */
  static void frame2data(stBlinkenFrame *pFrame, etBlinkenProto proto,
                         std::string &data);

/**
   * @brief get "no frame" protocol data
   * @param[in] proto Blinken protocol identifier
   * @param[out] data protcol data
   */
  static void noFrame2data(etBlinkenProto proto, std::string &data);

/// receive data from socket
  void receiveFromSock();

protected:
  SettingFile m_fileInStream; ///< input stream name file
  SettingFile m_fileBind;     ///< bind address file
  Directory   m_dirDestsBlp;  ///< static BLP destinations directory
  Directory   m_dirDestsEblp; ///< static EBLP destinations directory
  Directory   m_dirDestsMcuf; ///< static MCUF destinations directory
  std::string m_nameInStream; ///< name of input stream
  Stream      *m_pInStream;   ///< input stream
  SOCK        *m_pSock;       ///< socket to use for sending streams
  Dests       m_destsBlp;     ///< static BLP destinations
  Dests       m_destsEblp;    ///< static EBLP destinations
  Dests       m_destsMcuf;    ///< static MCUF destinations
  DynDests    m_dynDestsBlp;  ///< dynamic BLP destinations
  DynDests    m_dynDestsEblp; ///< dynamic EBLP destinations
  DynDests    m_dynDestsMcuf; ///< dynamic MCUF destinations
}; // class Sender

/* ##########
   # Sender #
   ########## */

/**
 * @brief constructor
 * @param[in] callMgr callback manager
 * @param[in] streamMgr stream manager
 * @param[in] dirBase base directory
 */
template<typename ADDR, typename SOCK>
Sender<ADDR, SOCK>::Sender(CallMgr &callMgr, StreamMgr &streamMgr,
                           const Directory &dirBase):
  Module(callMgr, streamMgr, dirBase),
  m_fileInStream(dirBase.getFile("instream")),
  m_fileBind(dirBase.getFile("bind")),
  m_dirDestsBlp(dirBase.getSubdir("blp")),
  m_dirDestsEblp(dirBase.getSubdir("eblp")),
  m_dirDestsMcuf(dirBase.getSubdir("mcuf")),
  m_pInStream(NULL),
  m_pSock(NULL)
{
  // get input stream and attach to it
  getInStream();
  // create and bind socket
  createSock();

// load static destinations
  updateDestsFull(m_dirDestsBlp, m_destsBlp, BlinkenProtoBlp);
  updateDestsFull(m_dirDestsEblp, m_destsEblp, BlinkenProtoEblp);
  updateDestsFull(m_dirDestsMcuf, m_destsMcuf, BlinkenProtoMcuf);
}

/// virtual destructor
template<typename ADDR, typename SOCK>
Sender<ADDR, SOCK>::~Sender()
{
  // send "no frame" to all destinations
  sendAllNoFrame();

// destroy socket
  destroySock();
  // detach from input stream and release it
  releaseInStream();
  // cancel time callback
  m_callMgr.cancelTimeCall(this);
}

/// check for update of configuration
template<typename ADDR, typename SOCK>
void Sender<ADDR, SOCK>::updateConfig()
{
  // input stream name file was modified -> re-get input stream
  if (m_fileInStream.checkModified()) {
    releaseInStream();
    getInStream();
  }

// bind address file was modified -> re-create socket
  if (m_fileBind.checkModified()) {
    destroySock();
    createSock();
  }

// static destinations update
  // (directory modified -> full, otherwise -> light)
  if (m_dirDestsBlp.checkModified())
    updateDestsFull(m_dirDestsBlp, m_destsBlp, BlinkenProtoBlp);
  else
    updateDestsLight(m_destsBlp, BlinkenProtoBlp);
  if (m_dirDestsEblp.checkModified())
    updateDestsFull(m_dirDestsEblp, m_destsEblp, BlinkenProtoEblp);
  else
    updateDestsLight(m_destsEblp, BlinkenProtoEblp);
  if (m_dirDestsMcuf.checkModified())
    updateDestsFull(m_dirDestsMcuf, m_destsMcuf, BlinkenProtoMcuf);
  else
    updateDestsLight(m_destsMcuf, BlinkenProtoMcuf);
}

/**
 * @brief set current frame
 * @param[in] stream stream name
 * @param[in] pFrame current frame
 */
template<typename ADDR, typename SOCK>
void Sender<ADDR, SOCK>::setFrame(const std::string &stream,
                                  stBlinkenFrame *pFrame)
{
  // send frame to all destinations
  sendAllFrame(pFrame);

(void)stream; // unused
}

/**
 * @brief set current frame to none
 * @param[in] stream stream name
 */
template<typename ADDR, typename SOCK>
void Sender<ADDR, SOCK>::setNoFrame(const std::string &stream)
{
  // send "no frame" to all destinations
  sendAllNoFrame();

(void)stream; // unused
}

/// callback when requsted time reached
template<typename ADDR, typename SOCK>
void Sender<ADDR, SOCK>::timeCall()
{
  stBlinkenFrame *pFrame;
  std::string data;

// get current frame from stream
  if (m_pInStream && m_pInStream->getCurFrame(pFrame))
    // repeat frame to all destinations
    sendAllFrame(pFrame);

// no stream of no current frame
  else
    // repeat "no frame" to all destinations
    sendAllNoFrame();
}

/**
 * @brief callback when I/O object is readable
 * @param[in] io I/O object that is readable
 */
template<typename ADDR, typename SOCK>
void Sender<ADDR, SOCK>::ioReadCall(Io *io)
{
  // reception on socket
  if (io == m_pSock)
    receiveFromSock();
}

/**
 * @brief callback when I/O object is writable
 * @param[in] io I/O object that is writable
 */
template<typename ADDR, typename SOCK>
void Sender<ADDR, SOCK>::ioWriteCall(Io *io)
{
  (void)io; // unused
}

/// get input stream and attach to it
template<typename ADDR, typename SOCK>
void Sender<ADDR, SOCK>::getInStream()
{
  // get input stream
  m_fileInStream.getStr(m_nameInStream);
  m_pInStream = &m_streamMgr.refStream(m_nameInStream);

// attach to input stream
  if (m_pInStream)
    m_pInStream->attach(this);
}

/// detach from input stream and release it
template<typename ADDR, typename SOCK>
void Sender<ADDR, SOCK>::releaseInStream()
{
  // detach from input stream
  if (m_pInStream)
    m_pInStream->detach(this);

// unreference stream
  m_pInStream = NULL;
  m_streamMgr.unrefStream(m_nameInStream);
}

/// create socket and bind it
template<typename ADDR, typename SOCK>
void Sender<ADDR, SOCK>::createSock()
{
  std::string strAddr;
  ADDR addr;

// create socket
  if (!m_pSock) {
    m_pSock = new SOCK();
    if (!m_pSock)
      return;
  }

// get bind address from bind address setting file
  if (!m_fileBind.getStr(strAddr) || !addr.fromStr(strAddr)) {
    delete m_pSock;
    m_pSock = NULL;
    return;
  }

// bind socket
  if (!m_pSock->bind(addr)) {
    delete m_pSock;
    m_pSock = NULL;
    return;
  }

// request callback on recpetion
  m_callMgr.requestIoReadCall(this, m_pSock);
}

/// destroy socket
template<typename ADDR, typename SOCK>
void Sender<ADDR, SOCK>::destroySock()
{
  // send "no frame" to all destinations
  // (stream from this socket will stop now)
  sendAllNoFrame();

// clear dynamic destinations
  // (they registered with this socket adn this socket is gone)
  m_dynDestsBlp.clear();
  m_dynDestsEblp.clear();
  m_dynDestsMcuf.clear();

// cancel callback request
  m_callMgr.cancelIoReadCall(this, m_pSock);

// destroy socket
  if (m_pSock) {
    delete m_pSock;
    m_pSock = NULL;
  }
}

/**
 * @brief light update of static destinations,
 *        i.e. stat all files in current static destination directory
 * @param[in] dests static destinations for one protocol
 * @param[in] proto Blinken protocol identifier
 */
template<typename ADDR, typename SOCK>
void Sender<ADDR, SOCK>::updateDestsLight(Dests &dests, etBlinkenProto proto)
{
  // walk through all files in static dest dir and check for modification
  typename Dests::iterator itDest = dests.begin();
  while (itDest != dests.end()) {

// address changed
    if (itDest->m_file.checkModified()) {
      // send "no frame" to old address
      sendNoFrame(itDest->m_addr, proto);
      // re-load address
      if (!itDest->parse()) {
        // parsing address failed -> remove destination
        itDest = dests.erase(itDest);
        // do not advance to next destination
        continue;
      }
      // send current frame to new address
      sendCurFrame(itDest->m_addr, proto);
    }

// advance to next destination
    ++itDest;

} // while itDest
}

/**
 * @brief full update of static destinations,
 *        i.e. scan files in playlist directory
 * @param[in] dirDests static destinations directory for protocol
 * @param[in] dests static destinations for protocol
 * @param[in] proto Blinken protocol identifier
 */
template<typename ADDR, typename SOCK>
void Sender<ADDR, SOCK>::updateDestsFull(Directory &dirDests, Dests &dests,
                                         etBlinkenProto proto)
{
  // get list of files in static destinations directory
  typedef std::list<std::string> Filelist;
  Filelist curFiles;
  dirDests.getEntries(Directory::TypeFile, curFiles);

// walk through current static destinations and file list simultaneously
  Filelist::const_iterator itFile = curFiles.begin();
  typename Dests::iterator itDest = dests.begin();
  while (itFile != curFiles.end() || itDest != dests.end()) {

// new static destination inserted
    if (itDest == dests.end() ||
        (itFile != curFiles.end() && *itFile < itDest->m_name)) {
      // parse new address
      Dest dest(*itFile, dirDests.getFile(*itFile));
      if (dest.parse()) {
        // insert new static destination
        dests.insert(itDest, dest);
        // send current frame to new static destination
        sendCurFrame(itDest->m_addr, proto);
      }
      // advance to next file
      ++itFile;
    }

// static destination removed
    // static destination changed (=> remove and re-insert in next iteration)
    else if (itFile == curFiles.end() || *itFile > itDest->m_name ||
             itDest->m_file.checkModified()) {
      // send "no frame" to old static destination
      sendNoFrame(itDest->m_addr, proto);
      // remove static destination
      itDest = dests.erase(itDest);
      // do not advance to next file
    }

// static destination stayed in playlist and did not change
    else {
      // advance to next file and next static destination
      ++itFile;
      ++itDest;
    }

} // while itFile itDest
}

/// remove timed-out dynamic destinations
template<typename ADDR, typename SOCK>
void Sender<ADDR, SOCK>::removeTimedOutDynDests()
{
  Time now, timeout;
  typename DynDests::iterator itDyn;

now = Time::now();
  timeout = Time(30);

for (itDyn = m_dynDestsBlp.begin(); itDyn != m_dynDestsBlp.end(); )
    if (itDyn->second + timeout < now)
      m_dynDestsBlp.erase(itDyn++);
    else
      ++itDyn;
  for (itDyn = m_dynDestsEblp.begin(); itDyn != m_dynDestsEblp.end(); )
    if (itDyn->second + timeout < now)
      m_dynDestsEblp.erase(itDyn++);
    else
      ++itDyn;
  for (itDyn = m_dynDestsMcuf.begin(); itDyn != m_dynDestsMcuf.end(); )
    if (itDyn->second + timeout < now)
      m_dynDestsMcuf.erase(itDyn++);
    else
      ++itDyn;
}

/**
 * @brief send frame to all destinations
 * @param[in] pFrame frame to send
 */
template<typename ADDR, typename SOCK>
void Sender<ADDR, SOCK>::sendAllFrame(stBlinkenFrame *pFrame)
{
  std::string data;

// remove timed-out dynamic destinations
  removeTimedOutDynDests();

// convert frame to protocol data and send to all static/dynamic destinations
  if (!m_destsBlp.empty() || !m_dynDestsBlp.empty()) {
    frame2data(pFrame, BlinkenProtoBlp, data);
    sendDests(data, m_destsBlp, m_dynDestsBlp);
  }
  if (!m_destsEblp.empty() || !m_dynDestsEblp.empty()) {
    frame2data(pFrame, BlinkenProtoEblp, data);
    sendDests(data, m_destsEblp, m_dynDestsEblp);
  }
  if (!m_destsMcuf.empty() || !m_dynDestsMcuf.empty()) {
    frame2data(pFrame, BlinkenProtoMcuf, data);
    sendDests(data, m_destsMcuf, m_dynDestsMcuf);
  }

// request time callback in one second
  m_callMgr.requestTimeCall(this, Time::now() + Time(1));
}

/// send "no frame" to all destinations
template<typename ADDR, typename SOCK>
void Sender<ADDR, SOCK>::sendAllNoFrame()
{
  std::string data;

// remove timed-out dynamic destinations
  removeTimedOutDynDests();

// get "no frame" protocol data and send to all static/dynamic destinations
  if (!m_destsBlp.empty() || !m_dynDestsBlp.empty()) {
    noFrame2data(BlinkenProtoBlp, data);
    sendDests(data, m_destsBlp, m_dynDestsBlp);
  }
  if (!m_destsEblp.empty() || !m_dynDestsEblp.empty()) {
    noFrame2data(BlinkenProtoEblp, data);
    sendDests(data, m_destsEblp, m_dynDestsEblp);
  }
  if (!m_destsMcuf.empty() || !m_dynDestsMcuf.empty()) {
    noFrame2data(BlinkenProtoMcuf, data);
    sendDests(data, m_destsMcuf, m_dynDestsMcuf);
  }

// request time callback in one second
  m_callMgr.requestTimeCall(this, Time::now() + Time(1));
}

/**
 * @brief send data to static/dynamic destinations
 * @param[in] data data to send
 * @param[in] dests static destinations
 * @param[in] dynDests dynamic destinations
 */
template<typename ADDR, typename SOCK>
void Sender<ADDR, SOCK>::sendDests(const std::string &data, const Dests dests,
                                   const DynDests dynDests)
{
  // send data to static destinations
  typename Dests::const_iterator itDest;
  for (itDest = dests.begin(); itDest != dests.end(); ++itDest)
    sendFrame(data, itDest->m_addr);

// send data to all dynamic destinations
  typename DynDests::const_iterator itDyn;
  for (itDyn = dynDests.begin(); itDyn != dynDests.end(); ++itDyn)
    sendFrame(data, itDyn->first);
}

/**
 * @brief send current frame to address
 * @param[in] addr address to send to
 * @param[in] proto Blinken protocol identifier
 */
template<typename ADDR, typename SOCK>
void Sender<ADDR, SOCK>::sendCurFrame(const ADDR &addr,
                                      etBlinkenProto proto) const
{
  stBlinkenFrame *pFrame;
  std::string data;

// get current frame from stream
  if (m_pInStream && m_pInStream->getCurFrame(pFrame)) {
    // convert frame to protocal data
    frame2data(pFrame, proto, data);
    // send frame to address
    sendFrame(data, addr);
  }

// no stream of no current frame
  else {
    // get "no frame" ad protocal data
    noFrame2data(proto, data);
    // send "no frame" to address
    sendFrame(data, addr);
  }
}

/**
 * @brief send "no frame" to address
 * @param[in] addr address to send to
 * @param[in] proto Blinken protocol identifier
 */
template<typename ADDR, typename SOCK>
void Sender<ADDR, SOCK>::sendNoFrame(const ADDR &addr,
                                     etBlinkenProto proto) const
{
  std::string data;

// get "no frame" ad protocal data
  noFrame2data(proto, data);

// send "no frame" to address
  sendFrame(data, addr);
}

/**
 * @brief send frame to address
 * @param[in] data protocol data of frame
 * @param[in] addr address to send to
 */
template<typename ADDR, typename SOCK>
void Sender<ADDR, SOCK>::sendFrame(const std::string &data,
                                   const ADDR &addr) const
{
  if (m_pSock) {
    m_pSock->send(data, addr);
  }
}

/**
 * @brief convert frame to protocol data
 * @param[in] pFrame frame
 * @param[in] proto Blinken protocol identifier
 * @param[out] data protocol data
 */
template<typename ADDR, typename SOCK>
void Sender<ADDR, SOCK>::frame2data(stBlinkenFrame *pFrame,
                                    etBlinkenProto proto, std::string &data)
{
  char buf[65536];
  int len;

// convert frame to protcol data
  len = BlinkenFrameToNetwork(pFrame, proto, buf, sizeof(buf));
  if (len < 0)
    len = 0;
  data.assign(buf, len);
}

/**
 * @brief get "no frame" protocol data
 * @param[in] proto Blinken protocol identifier
 * @param[out] data protcol data
 */
template<typename ADDR, typename SOCK>
void Sender<ADDR, SOCK>::noFrame2data(etBlinkenProto proto, std::string &data)
{
  char buf[16];
  int len;

// obtain "no frame" protcol data
  len = BlinkenProtoMakePacket(proto, BlinkenPacketStreamEnd,
        buf, sizeof(buf));
  if (len < 0)
    len = 0;
  data.assign(buf, len);
}

/// receive data from socket
template<typename ADDR, typename SOCK>
void Sender<ADDR, SOCK>::receiveFromSock()
{
  etBlinkenProto proto;
  etBlinkenPacket packet;
  std::string data;
  ADDR addr;

// make sure socket exists
  if (!m_pSock)
    return;

// receive (leave if no reception)
  if (!m_pSock->recv(data, addr))
    return;

// detect packet type and protocol
  BlinkenProtoDetectPacket(data.c_str(), data.size(), &proto, &packet);

switch (packet) {

// request -> add to dynamic destinations and send current frame
    case BlinkenPacketRequest:
      switch (proto) {
        case BlinkenProtoBlp:
          m_dynDestsBlp[addr] = Time::now();
          sendCurFrame(addr, BlinkenProtoBlp);
          break;
        case BlinkenProtoEblp:
          m_dynDestsEblp[addr] = Time::now();
          sendCurFrame(addr, BlinkenProtoEblp);
          break;
        case BlinkenProtoMcuf:
          m_dynDestsMcuf[addr] = Time::now();
          sendCurFrame(addr, BlinkenProtoMcuf);
          break;
        default:
          break;
      }
      break;

// end request -> remove from dynamic destinations
    case BlinkenPacketEndRequest:
      switch (proto) {
        case BlinkenProtoBlp:
          m_dynDestsBlp.erase(addr);
          break;
        case BlinkenProtoEblp:
          m_dynDestsEblp.erase(addr);
          break;
        case BlinkenProtoMcuf:
          m_dynDestsMcuf.erase(addr);
          break;
        default:
          break;
      }
      break;

default:
      break;

} // switch (packet)

}

/* ################
   # Sender::Dest #
   ################ */

/// constructor
template<typename ADDR, typename SOCK>
Sender<ADDR, SOCK>::Dest::Dest(const std::string &name, const File &file):
  m_name(name),
  m_file(file)
{
}

/// parse address from current file
template<typename ADDR, typename SOCK>
bool Sender<ADDR, SOCK>::Dest::parse()
{
  std::string strAddr;
  return m_file.getStr(strAddr) && m_addr.fromStr(strAddr);
}

} // namespace Blinker

#endif // #ifndef SENDER_H

Blinker

added stream name to stream receiver interface callbacks