added the call watcher, useful class to monitor/interface call status to external applications/network etc

git-svn-id: http://svn.berlios.de/svnroot/repos/sems/trunk@403 8eb893ce-cfd4-0310-b710-fb5ebe64c474

core/AmCallWatcher.cpp

@@ -0,0 +1,203 @@
+/*
+ * $Id: AmCallWatcher.cpp 279 2007-03-23 21:30:44Z sayer $
+ *
+ *  (c) 2007 iptego GmbH
+ *
+ * This file is part of sems, a free SIP media server.
+ *
+ * sems is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation; either version 2 of the License, or
+ * (at your option) any later version
+ *
+ * For a license to use the ser software under conditions
+ * other than those described here, or to purchase support for this
+ * software, please contact iptel.org by e-mail at the following addresses:
+ *    info@iptel.org
+ *
+ * sems 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 General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License 
+ * along with this program; if not, write to the Free Software 
+ * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
+ */
+
+#include <sys/time.h>
+
+#include "log.h"
+
+#include "AmCallWatcher.h"
+
+AmCallWatcher::AmCallWatcher() 
+  : garbage_collector(new AmCallWatcherGarbageCollector(soft_states_mut, soft_states)),
+    AmEventQueue(this)
+{ 
+}
+
+AmCallWatcher::~AmCallWatcher() 
+{ 
+}
+
+void AmCallWatcher::run() {
+  DBG("starting call watcher.\n");
+  garbage_collector->start();
+  while (true) {
+    waitForEvent();
+    processEvents();
+  }
+}
+
+void AmCallWatcher::on_stop() {
+  ERROR("The call watcher cannot be stopped.\n");
+}
+
+void AmCallWatcher::process(AmEvent* ev) {
+  CallStatusUpdateEvent* csu = dynamic_cast<CallStatusUpdateEvent*>(ev);
+  if (NULL == csu) {
+    ERROR("received invalid event!\n");
+    return;
+  }
+
+
+  switch (csu->event_id) {
+  case CallStatusUpdateEvent::Initialize: {
+    states_mut.lock();
+    DBG("adding  call state '%s'\n", 
+	csu->get_call_id().c_str());      
+
+    // check whether already there
+    CallStatusMap::iterator it = states.find(csu->get_call_id());
+    if (it != states.end()) {
+      WARN("implementation error: state '%s' already in list!\n",
+	   csu->get_call_id().c_str());
+      // avoid leak - delete the old state
+      delete it->second;
+    }
+
+    // insert the new one
+    states[csu->get_call_id()] = csu->get_init_status();
+    states_mut.unlock();
+  } break;
+
+  case CallStatusUpdateEvent::Update: {
+    states_mut.lock();
+    CallStatusMap::iterator it = states.find(csu->get_call_id());
+    if (it != states.end()) {
+      it->second->update(csu);      
+      it->second->dump();
+      states_mut.unlock();
+    } else {      
+      states_mut.unlock();
+
+      soft_states_mut.lock();
+      CallStatusTimedMap::iterator it = 
+	soft_states.find(csu->get_call_id());
+      if (it != soft_states.end()) {
+	it->second.first->update(csu);
+	it->second.first->dump();
+      } else {
+	DBG("received update event for inexistent call '%s'\n", 
+	    csu->get_call_id().c_str());
+      }
+      soft_states_mut.unlock();
+    }
+  } break;
+
+  case CallStatusUpdateEvent::Obsolete: {
+    states_mut.lock();
+    CallStatusMap::iterator it = states.find(csu->get_call_id());
+    if (it != states.end()) {
+
+      CallStatus* cs = it->second;
+      states.erase(it);
+      size_t s_size = states.size();
+      states_mut.unlock();
+
+      struct timeval now;
+      gettimeofday(&now, NULL);
+      
+      soft_states_mut.lock();
+      soft_states[csu->get_call_id()] = 
+	std::make_pair(cs, now.tv_sec + WATCHER_SOFT_EXPIRE_SECONDS);
+      size_t soft_size = soft_states.size();
+      soft_states_mut.unlock();
+      
+      DBG("moved call state '%s' to soft-state map (%u states, %u soft-states)\n", 
+	  csu->get_call_id().c_str(), s_size, soft_size);      
+      
+    } else {      
+      DBG("received obsolete event for inexistent call '%s'\n", 
+	  csu->get_call_id().c_str());
+      states_mut.unlock();
+    }
+  }break;
+  }
+}
+
+void AmCallWatcher::dump() {
+  states_mut.lock();
+  for (CallStatusMap::iterator it = states.begin(); 
+       it != states.end(); it++) {
+    it->second->dump();
+  }
+  states_mut.unlock();
+}
+
+CallStatus* AmCallWatcher::getStatus(const string& call_id) {
+  CallStatus* res = NULL;
+
+  states_mut.lock();
+
+  CallStatusMap::iterator it = states.find(call_id);
+  if (it != states.end()) {
+    res = it->second->copy();
+    states_mut.unlock();
+  } else {
+    states_mut.unlock();
+
+    // check obsolete states
+    soft_states_mut.lock();
+    CallStatusTimedMap::iterator it = 
+      soft_states.find(call_id);
+    if (it != soft_states.end()) {
+      // got it. return and remove from map
+      res = it->second.first;
+      soft_states.erase(it);
+      DBG("erased call state '%s' (%u in list).\n",
+	  call_id.c_str(), soft_states.size());
+    } else {
+      DBG("state for call '%s' not found.\n",
+	  call_id.c_str());
+    }
+    soft_states_mut.unlock();
+  }
+  return res;
+}
+
+void AmCallWatcherGarbageCollector::run() {
+  DBG("AmCallWatcherGarbageCollector started.\n");
+  while (true) {
+    sleep(2);
+    struct timeval now;
+    gettimeofday(&now, NULL);
+
+    bool erased = false;
+
+    mut.lock();
+    for (AmCallWatcher::CallStatusTimedMap::iterator it = garbage.begin(); 
+	 it != garbage.end(); it++) {
+      if (it->second.second < now.tv_sec) {
+	garbage.erase(it); // map::erase does not invalidate map::iterator
+	erased = true;
+      }
+    }
+    if (erased){
+      DBG("cleared old soft-states (%u soft-states remaining)\n", 
+	  garbage.size());      
+    }
+    mut.unlock();
+  }
+}

core/AmCallWatcher.h

@@ -0,0 +1,162 @@
+/*
+ * $Id: AmCallWatcher.h 279 2007-03-23 21:30:44Z sayer $
+ *
+ *  (c) 2007 iptego GmbH
+ *
+ * This file is part of sems, a free SIP media server.
+ *
+ * sems is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation; either version 2 of the License, or
+ * (at your option) any later version
+ *
+ * For a license to use the ser software under conditions
+ * other than those described here, or to purchase support for this
+ * software, please contact iptel.org by e-mail at the following addresses:
+ *    info@iptel.org
+ *
+ * sems 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 General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License 
+ * along with this program; if not, write to the Free Software 
+ * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
+ */
+
+#ifndef _AM_CALL_WATCHER_H
+#define _AM_CALL_WATCHER_H
+
+//
+// States are put into map on an Initialize event.
+// States are held in a map identified by call_id
+// (opaque identifier) and updated with Update events.
+// Once an Obsolete event is received, the states are 
+// moved to soft-state map, where they are held until
+// queried to a maximum of WATCHER_SOFT_EXPIRE_SECONDS
+
+#define WATCHER_SOFT_EXPIRE_SECONDS  5
+
+#include <string>
+using std::string;
+
+#include <map>
+using std::map;
+
+#include <utility>
+using std::pair;
+
+#include "AmEventQueue.h"
+#include "AmEvent.h"
+#include "AmThread.h"
+
+/** 
+ * event that carries out call status update
+ */
+class CallStatus;
+
+class CallStatusUpdateEvent : public AmEvent {
+  string call_id;
+
+  CallStatus* init_status;
+
+ public:
+  enum UpdateType {
+    Initialize = 0,
+    Update,
+    Obsolete
+  };
+
+  CallStatusUpdateEvent(UpdateType t, const string& call_id)
+    : call_id(call_id), AmEvent(t)  { }
+
+  // implicit: initialize
+  CallStatusUpdateEvent(const string& call_id, CallStatus* init_status)
+    : call_id(call_id), init_status(init_status), AmEvent(Initialize)  { }
+
+  ~CallStatusUpdateEvent() { }
+
+  string get_call_id() { return call_id; }
+  CallStatus* get_init_status() { return init_status; }
+};
+
+/** 
+ * interface for an update-able call status
+ */
+class CallStatus
+{
+ public:
+  CallStatus()  { }
+  virtual ~CallStatus() { }
+
+  /** update from an event */
+  virtual void update(CallStatusUpdateEvent* e) = 0;
+
+  /** get a copy of self with relevant data */
+  virtual CallStatus* copy() = 0;
+  virtual void dump() { }
+};
+
+class AmCallWatcherGarbageCollector;
+/**
+ * call watcher is an entity for managing call status
+ * via events that change status. Events are executed in a 
+ * separate thread serially by processing the event queue, 
+ * so synchronous status queries do not block the thread 
+ * reporting the status change.
+ */
+class AmCallWatcher
+: public AmThread, 
+  public AmEventQueue,
+  public AmEventHandler
+{
+ public:
+  typedef map<string, CallStatus*> CallStatusMap;
+  typedef map<string, pair<CallStatus*, time_t> > CallStatusTimedMap;
+
+ private:
+  CallStatusMap states;
+  AmMutex       states_mut;
+
+
+  CallStatusTimedMap soft_states;
+  AmMutex            soft_states_mut;
+  AmCallWatcherGarbageCollector* garbage_collector;
+
+ public:
+  AmCallWatcher();
+  ~AmCallWatcher();
+
+  // thread
+  void run();
+  void on_stop();
+
+  // eventhandler
+  void process(AmEvent*);
+
+  CallStatus* getStatus(const string& call_id);
+
+  // dump all states
+  void dump();
+};
+
+/** 
+ * checks garbage every two seconds. 
+ * A bit inefficient with two threads, but AmCallWatcher
+ * shouldn't be blocked by event.
+ */
+class AmCallWatcherGarbageCollector 
+: public AmThread
+{
+  AmMutex& mut;
+  AmCallWatcher::CallStatusTimedMap& garbage;
+ public: 
+  AmCallWatcherGarbageCollector(AmMutex& mut,
+				AmCallWatcher::CallStatusTimedMap& garbage)
+    : mut(mut), garbage(garbage) {}
+  void run();
+  void on_stop() { }
+};
+
+#endif

core/AmDtmfDetector.cpp

@@ -87,7 +87,8 @@ void AmSipDtmfEvent::parseLine(const string& line)
       line.substr(0, KeySignal.length()) == KeySignal)
     {
       string event(line.substr(KeySignal.length(), string::npos));
-      switch (event.c_str()[0])
+
+      switch (event.c_str()[event.find_first_not_of(" \t")])
         {
         case '*':
 	  m_event = 10;