S3FC project page

S3FC home page

---

               

---

s3_txport_local.tcc

Go to the documentation of this file.

/*
 * Stone Three Foundation Class (s3fc) provides a number of utility classes.
 * Copyright (C) 2001 by Stone Three Signal Processing (Pty) Ltd.
 *
 * Authored by Stone Three Signal Processing (Pty) Ltd.
 *
 * This library is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
 * License as published by the Free Software Foundation; either
 * version 2.1 of the License, or (at your option) any later version.
 *
 * This library is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * Lesser General Public License for more details.
 *
 * You should have received a copy of the GNU Lesser General Public
 * License along with this library; if not, write to the Free Software
 * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 *
 * Please see the file 'COPYING' in the source root directory.
 */

template<class T_Data>
s3_txport_local<T_Data>::s3_txport_local( unsigned int queue_size)
  : rx_queue(queue_size), tx_queue(queue_size),connect_id(0), pn(500)
{
   start();
}

template<class T_Data>
s3_txport_local<T_Data>::~s3_txport_local()
{
   if (is_running())
   {
      terminate.post();
      wait_on_exit();
   }
   disconnect_all();
}

template<class T_Data>
s3_fifo_queue<s3_txport_data<T_Data> >& s3_txport_local<T_Data>::get_tx_queue()
{
   return tx_queue;
}

template<class T_Data>
s3_fifo_queue<s3_txport_data<T_Data> >& s3_txport_local<T_Data>::get_rx_queue()
{
   return rx_queue;
}

template<class T_Data>
std::list<s3_txport_client_id>
s3_txport_local<T_Data>::get_client_id_list() const
{
   std::list<s3_txport_client_id> ids;
   typename c_map::const_iterator ii;
   connection_mutex.lock();
   for ( ii = connections.begin(); ii!= connections.end(); ii++)
   {
      ids.push_back(ii->first);
   }
   connection_mutex.unlock();
   return ids;
}

template<class T_Data>
void s3_txport_local<T_Data>::send_metadata(s3_txport_client_id id,
                   const std::string& metadata)
{
   connection_mutex.lock();
   if (connections.count(id) != 0)
   {
       connections[id]->dispatch_event(
     s3_txport_event(s3_txport_event::metadata, id,
           metadata));
   }
   connection_mutex.unlock();

}

template<class T_Data>
bool s3_txport_local<T_Data>::is_connected_by_id(s3_txport_client_id id)
{
   bool ret;
   connection_mutex.lock();
   ret = connections.count(id) != 0;
   connection_mutex.unlock();
   return ret;
}

template<class T_Data>
bool s3_txport_local<T_Data>::disconnect_by_id(s3_txport_client_id id)
{
   bool ret;
   connection_mutex.lock();
   if (connections.count(id) != 0 )
   {
       // Remove us from the offending remote
       connections[id]->deregister_me(this);
       connections.erase(id);
       dispatch_event(s3_txport_event(s3_txport_event::disconnect,id,
            "Disconnected locally"));
       ret = true;
   }
   else
   {
      ret = false;
   }
   connection_mutex.unlock();
   return ret;
}

template<class T_Data>
void s3_txport_local<T_Data>::disconnect_all()
{
   typename c_map::iterator ii = connections.begin();
   typename c_map::iterator next_ii;
   
   while ( ii != connections.end() ) 
   {
      next_ii = ii;
      next_ii++;
      disconnect_by_id(ii->first);
      ii = next_ii;
   }
}

template<class T_Data>
bool s3_txport_local<T_Data>::listening() const
{
   return is_running();
}

template<class T_Data>
s3_txport_client_id s3_txport_local<T_Data>::connect(
   s3_txport_local<T_Data> *listener)
{
   if (listener == 0)
   {
       throw s3_generic_exception(
     "s3_txport_local::connect(s3_txport_local *)",
     "Remote listener does not exist (null pointer)");
   }
   s3_txport_local<T_Data> *remote = listener->register_me(this);
   if (remote == 0)
   {
      // Already registered so buzz off.
      dispatch_event(s3_txport_event(s3_txport_event::error,
                 0,"Already connected to this remote"));
      return static_cast<s3_txport_client_id>(0);
   }
   s3_txport_client_id id = next_id();
   connection_mutex.lock();
   connections[id] = remote;
   connection_mutex.unlock();
   dispatch_event(s3_txport_event(s3_txport_event::connect,
              id,"Locally established connection"));
   return id;
}

template<class T_Data>
s3_txport_local<T_Data>* s3_txport_local<T_Data>::register_me(
   s3_txport_local<T_Data> *remote)
{
   // Check if we are already registered!!
   typename c_map::iterator ii;
   bool registered = false;
   connection_mutex.lock();
   for ( ii = connections.begin(); ii!= connections.end(); ii++)
   {
      if (ii->second == remote)
      {
         registered = true;
      }
   }
   connection_mutex.unlock();
   if (registered)
   {
      dispatch_event(s3_txport_event(s3_txport_event::error,0,
         "Remote attempted duplicate connect"));
      return 0;
   }

   s3_txport_client_id id = next_id();
   connection_mutex.lock();
   connections[id] = remote;
   connection_mutex.unlock();
   dispatch_event(s3_txport_event(s3_txport_event::connect,id,
              "Remote connetion received"));
   return this;
}

template<class T_Data>
void s3_txport_local<T_Data>::deregister_me(
   const s3_txport_local<T_Data> *remote)
{
   typename c_map::iterator ii;
   connection_mutex.lock();

   ii = connections.begin();
   while( ii != connections.end() )
   {
      typename c_map::iterator next_ii;
      next_ii = ii;
      next_ii++;

      // Remove if correct value (this invalidates ii).
      if (ii->second == remote)
      {
         dispatch_event(s3_txport_event(s3_txport_event::disconnect,
               ii->first,
               "Remote has deregistered"));
         connections.erase(ii);
      }
      ii = next_ii;      
   }
   connection_mutex.unlock();
}

template<class T_Data>
bool s3_txport_local<T_Data>::listen()
{
   return true;
}

template<class T_Data>
void s3_txport_local<T_Data>::cleanup()
{
   pn.disable();
   pn.unsubscribe(sem);
   tx_queue.unsubscribe_consumer(sem);
   S3FC_DBG(std::cout << "s3_txport_local:: cleanup() cleaning up"
       << std::endl);
}

template<class T_Data>
void s3_txport_local<T_Data>::main_loop()
{
   typename s3_fifo_queue<s3_txport_data<T_Data> >::s3_queue_data_t *d;
   std::list<s3_txport_client_id> id_list;
   pn.subscribe(sem);
   pn.enable();
   tx_queue.subscribe_consumer(sem);

   while (1)
   {
      sem.wait();
      if (terminate.try_wait())
      {
         return; // kill thread.
      }
      d = tx_queue.nbl_open_output();
      if (d == 0)  continue; // Loop because the hit was something else

      id_list = d->get_id_list();

      if (!id_list.empty())
      {
         // Now send the same data to each client in the list
         for ( std::list<s3_txport_client_id>::iterator ii = id_list.begin();
               ii != id_list.end(); ii++)
         {
            connection_mutex.lock();
            if (connections.count(*ii) != 0 )
            {
               connections[*ii]->get_rx_queue().push(*d);
               connection_mutex.unlock();
            }
            else
            {
               connection_mutex.unlock();
               dispatch_event(s3_txport_event(s3_txport_event::error,
         *ii, "Data sent to non-existant client ID"));
            }
         }
      }
      tx_queue.close_output(d);
   }
}
template<class T_Data>
int s3_txport_local<T_Data>::num_connections() const
{
   int c;
   connection_mutex.lock();
   c = connections.size();
   connection_mutex.unlock();
   return c;
}

---

Send comments to: s3fc@stonethree.com