sipwise
/
sems
mirror of https://github.com/sipwise/sems.git
1
0
Fork 0
Code Issues Releases Wiki Activity

Merge branch 'master' of ssh://git.sip-router.org/sems

Browse Source
sayer/1.4-spce2.6
Peter Lemenkov 16 years ago
parent 3d25784410 791132ad78
commit 3d6244ebc0
12 changed files with 345 additions and 51 deletions
Show Stats Download Patch File Download Diff File

71
apps/msg_storage/MsgStorage.cpp
Unescape View File

@ -21,7 +21,9 @@ MsgStorage* MsgStorage::_instance = 0;
EXPORT_PLUGIN_CLASS_FACTORY(MsgStorage, MOD_NAME);
MsgStorage::MsgStorage(const string& name)
: AmDynInvokeFactory(name) {
: AmDynInvokeFactory(name),
listeners()
{
_instance = this;
}
@ -52,7 +54,7 @@ int MsgStorage::onLoad() {
status = mkdir(path.c_str(),
S_IRWXU | S_IRWXG | S_IROTH | S_IXOTH);
if (status && (errno != EEXIST)) {
ERROR("creating test path in storage '%s': %s\n",
ERROR("Write permission check failed. Could not create '%s': %s\n",
path.c_str(),strerror(errno));
return -1;
}
@ -91,6 +93,7 @@ void MsgStorage::invoke(const string& method,
userdir_open(args.get(0).asCStr(),
args.get(1).asCStr(),
ret);
#if 0
} else if(method == "userdir_close"){
ret.push(userdir_close(args.get(0).asCStr(),
args.get(1).asCStr()));
@ -98,6 +101,12 @@ void MsgStorage::invoke(const string& method,
userdir_getcount(args.get(0).asCStr(),
args.get(1).asCStr(),
ret);
#endif
} else if(method == "events_subscribe"){
events_subscribe(args.get(0).asDynInv(),
args.get(1).asCStr());
} else if(method == "events_unsubscribe"){
events_unsubscribe(args.get(0).asDynInv());
} else if(method == "_list"){
ret.push("msg_new");
ret.push("msg_get");
@ -105,8 +114,13 @@ void MsgStorage::invoke(const string& method,
ret.push("msg_delete");
ret.push("userdir_open");
#if 0
ret.push("userdir_close");
ret.push("userdir_getcount");
#endif
ret.push("events_subscribe");
ret.push("events_unsubscribe");
}
else
throw AmDynInvoke::NotImplemented(method);
@ -146,6 +160,9 @@ int MsgStorage::msg_new(string domain, string user,
if (data)
filecopy(data, fp);
fclose(fp);
event_notify(domain,user,"msg_new");
return MSG_OK;
}
@ -185,6 +202,8 @@ int MsgStorage::msg_markread(string domain, string user, string msg_name) {
return MSG_EREADERROR;
}
event_notify(domain,user,"msg_markread");
return MSG_OK;
}
@ -196,6 +215,9 @@ int MsgStorage::msg_delete(string domain, string user, string msg_name) {
path.c_str(),strerror(errno));
return MSG_EMSGNOTFOUND;
}
event_notify(domain,user,"msg_delete");
return MSG_OK;
}
@ -246,12 +268,55 @@ void MsgStorage::userdir_open(string domain, string user, AmArg& ret) {
ret.push(msglist);
}
#if 0
int MsgStorage::userdir_close(string domain, string user) {
// TODO: unblock the directory from delete (decrease lock)
return 0;
}
void MsgStorage::userdir_getcount(string domain, string user, AmArg& ret) { }
void MsgStorage::userdir_getcount(string domain, string user, AmArg& ret) {
// TODO: return some useful value
}
#endif
void MsgStorage::events_subscribe(AmDynInvoke* event_sink, string method)
{
listeners_mut.lock();
listeners.insert(make_pair(event_sink,method));
listeners_mut.unlock();
}
void MsgStorage::events_unsubscribe(AmDynInvoke* event_sink)
{
listeners_mut.lock();
listeners.erase(event_sink);
listeners_mut.unlock();
}
void MsgStorage::event_notify(const string& domain,
const string& user,
const string& event)
{
AmArg args,ret;
args.push(domain);
args.push(user);
args.push(event);
listeners_mut.lock();
for(Listeners::iterator it = listeners.begin();
it != listeners.end(); ++it) {
try {
it->first->invoke(it->second, args, ret);
}
catch(...){
DBG("Unexpected exception while notifying event subscribers");
}
ret.clear();
}
listeners_mut.unlock();
}
// copies ifp to ofp, blockwise
void MsgStorage::filecopy(FILE* ifp, FILE* ofp) {

15
apps/msg_storage/MsgStorage.h
Unescape View File

@ -3,6 +3,9 @@
#include "AmApi.h"
#include <map>
using std::map;
class MsgStorage : public AmDynInvokeFactory,
public AmDynInvoke
{
@ -11,6 +14,10 @@ class MsgStorage : public AmDynInvokeFactory,
string msg_dir;
typedef map<AmDynInvoke*,string> Listeners;
Listeners listeners;
AmMutex listeners_mut;
int msg_new(string domain, string user, string msg_name, FILE* data);
void msg_get(string domain, string user, string msg_name, AmArg& ret);
int msg_markread(string domain, string user, string msg_name);
@ -20,7 +27,15 @@ class MsgStorage : public AmDynInvokeFactory,
int userdir_close(string domain, string user);
void userdir_getcount(string domain, string user, AmArg& ret);
void events_subscribe(AmDynInvoke* event_sink, string method);
void events_unsubscribe(AmDynInvoke* event_sink);
void event_notify(const string& domain,
const string& user,
const string& event);
inline void filecopy(FILE* ifp, FILE* ofp);
public:
MsgStorage(const string& name);
~MsgStorage();

7
apps/mwi/CMakeLists.txt
Unescape View File

@ -0,0 +1,7 @@
set (mwi_SRCS
mwi.cpp
)
SET(sems_module_name mwi)
INCLUDE(${CMAKE_SOURCE_DIR}/cmake/module.rules.txt)
INCLUDE(${CMAKE_SOURCE_DIR}/cmake/config.rules.txt)

7
apps/mwi/Makefile
Unescape View File

@ -0,0 +1,7 @@
plug_in_name = mwi
module_ldflags =
module_cflags = -DMOD_NAME=\"$(plug_in_name)\"
COREPATH ?=../../core
include $(COREPATH)/plug-in/Makefile.app_module

2
apps/mwi/etc/mwi.conf
Unescape View File

@ -0,0 +1,2 @@
# Presence Server:
presence_server=127.0.0.1

126
apps/mwi/mwi.cpp
Unescape View File

@ -0,0 +1,126 @@
/*
Copyright (C) Anton Zagorskiy amberovsky@gmail.com
Oyster-Telecom Laboratory
Published under BSD License
*/
#include "AmPlugIn.h"
#include "AmSession.h"
#include "AmConfigReader.h"
#include "AmUtils.h"
#include "log.h"
#include "mwi.h"
#include <string>
MWI* MWI::_instance = 0;
AmDynInvoke* MWI::MessageStorage = 0;
EXPORT_PLUGIN_CLASS_FACTORY(MWI, MOD_NAME);
MWI::MWI(const string& name)
: AmDynInvokeFactory(name) {
_instance = this;
};
MWI::~MWI() { };
int MWI::onLoad()
{
AmDynInvokeFactory* ms_fact =
AmPlugIn::instance()->getFactory4Di("msg_storage");
if(!ms_fact || !(MessageStorage = ms_fact->getInstance())) {
ERROR("could not load msg_storage. Load a msg_storage implementation module.\n");
return -1;
};
// register the publish method as event sink for msg_storage events
AmArg es_args,ret;
es_args.push(this);
es_args.push("publish");
MessageStorage->invoke("events_subscribe",es_args,ret);
AmConfigReader cfg;
if(cfg.loadFile(AmConfig::ModConfigPath + "mwi.conf")) {
ERROR("can not load configuration file\n");
return -1;
};
presence_server = cfg.getParameter("presence_server");
if (presence_server.length())
DBG("set presence server '%s'\n", presence_server.c_str());
else {
ERROR("parameter 'presence_server' did not found in the configuration file\n");
return -1;
}
DBG("MWI module loaded.\n");
return 0;
};
void MWI::publish(const string& user, const string& domain)
{
int new_msgs = 0;
int all_msgs = 0;
string headers, body;
AmArg di_args, ret;
di_args.push(domain.c_str());
di_args.push(user.c_str());
MessageStorage->invoke("userdir_open",di_args,ret);
if (!ret.size() || !isArgInt(ret.get(0))) {
ERROR("userdir_open for user '%s' domain '%s' returned no (valid) result.\n", user.c_str(), domain.c_str());
return;
};
all_msgs = ret.get(1).size();
for (size_t i = 0; i < ret.get(1).size(); i++) {
AmArg& elem = ret.get(1).get(i);
if (elem.get(2).asInt()) // skip empty messages
new_msgs += elem.get(1).asInt();
else
all_msgs--;
};
DBG("Found %d new and %d old messages\n", new_msgs, all_msgs - new_msgs);
string vm_buf = int2str(new_msgs) + "/" + int2str(all_msgs - new_msgs);
headers = "Event: message-summary\r\n";
headers += "Subscription-State: active\r\n";
if (new_msgs > 0)
body = "Messages-Waiting: yes\r\n";
else
body = "Messages-Waiting: no\r\n";
body += "Message-Account: sip:" + user + "@" + domain + "\r\n";
body += "Voice-Message: " + vm_buf + " (" + vm_buf + ")\r\n";
AmSipDialog tmp_d(NULL);
tmp_d.local_party = string("<sip:mwi-publisher@") + presence_server + ">";
tmp_d.remote_party = domain.c_str();
tmp_d.route = "sip:" + presence_server;
tmp_d.remote_uri = "sip:" + user + "@" + domain;
tmp_d.callid = AmSession::getNewId() + "@" + presence_server;
tmp_d.local_tag = AmSession::getNewId();
tmp_d.sendRequest("PUBLISH", "application/simple-message-summary", body, headers);
};
void MWI::invoke(const string& method, const AmArg& args, AmArg& ret)
{
if (method == "publish") {
string user, domain;
user = args.get(1).asCStr();
domain = args.get(0).asCStr();
publish(user, domain);
ret.push(0);
}
else
throw AmDynInvoke::NotImplemented(method);
};

41
apps/mwi/mwi.h
Unescape View File

@ -0,0 +1,41 @@
/*
Copyright (C) Anton Zagorskiy amberovsky@gmail.com
Oyster-Telecom Laboratory
Published under BSD License
*/
#ifndef _MWI_H
#define _MWI_H
#include "AmApi.h"
#include <string>
class MWI : public AmDynInvokeFactory, public AmDynInvoke
{
private:
static MWI* _instance;
static AmDynInvoke* MessageStorage;
string presence_server;
typedef struct
{
unsigned int new_msgs;
unsigned int saved_msgs;
} msg_info_struct;
void getMsgInfo (const string& name, const string& domain, msg_info_struct& msg_info);
void publish (const string& name, const string& domain);
public:
MWI(const string& name);
~MWI();
AmDynInvoke* getInstance(){ return _instance; }
int onLoad();
void invoke(const string& method, const AmArg& args, AmArg& ret);
};
#endif

8
core/AmArg.cpp
Unescape View File

@ -37,6 +37,7 @@ const char* AmArg::t2str(int type) {
case AmArg::Double: return "Double";
case AmArg::CStr: return "CStr";
case AmArg::AObject: return "AObject";
case AmArg::ADynInv: return "ADynInv";
case AmArg::Blob: return "Blob";
case AmArg::Array: return "Array";
case AmArg::Struct: return "Struct";
@ -62,6 +63,7 @@ AmArg& AmArg::operator=(const AmArg& v) {
case Double: { v_double = v.v_double; } break;
case CStr: { v_cstr = strdup(v.v_cstr); } break;
case AObject:{ v_obj = v.v_obj; } break;
case ADynInv:{ v_inv = v.v_inv; } break;
case Array: { v_array = new ValueArray(*v.v_array); } break;
case Struct: { v_struct = new ValueStruct(*v.v_struct); } break;
case Blob: { v_blob = new ArgBlob(*v.v_blob); } break;
@ -306,6 +308,7 @@ bool operator==(const AmArg& lhs, const AmArg& rhs) {
case AmArg::Double: { return lhs.v_double == rhs.v_double; } break;
case AmArg::CStr: { return !strcmp(lhs.v_cstr,rhs.v_cstr); } break;
case AmArg::AObject:{ return lhs.v_obj == rhs.v_obj; } break;
case AmArg::ADynInv:{ return lhs.v_inv == rhs.v_inv; } break;
case AmArg::Array: { return lhs.v_array == rhs.v_array; } break;
case AmArg::Struct: { return lhs.v_struct == rhs.v_struct; } break;
case AmArg::Blob: { return (lhs.v_blob->len == rhs.v_blob->len) &&
@ -363,6 +366,7 @@ void AmArg::assertArrayFmt(const char* format) const {
case 'f': assertArgDouble(get(i)); got+='f'; break;
case 's': assertArgCStr(get(i)); got+='s'; break;
case 'o': assertArgAObject(get(i)); got+='o'; break;
case 'd': assertArgADynInv(get(i)); got+='d'; break;
case 'a': assertArgArray(get(i)); got+='a'; break;
case 'b': assertArgBlob(get(i)); got+='b'; break;
case 'u': assertArgStruct(get(i)); got+='u'; break;
@ -415,6 +419,10 @@ string AmArg::print(const AmArg &a) {
return double2str(a.asDouble());
case CStr:
return "'" + string(a.asCStr()) + "'";
case AObject:
return "<Object>";
case ADynInv:
return "<DynInv>";
case Array:
s = "[";
for (size_t i = 0; i < a.size(); i ++)

25
core/AmArg.h
Unescape View File

@ -49,11 +49,13 @@ class ArgObject {
virtual ~ArgObject() { }
};
struct ArgBlob { char* data;
struct ArgBlob {
char* data;
int len;
ArgBlob()
: data(NULL),len(0)
ArgBlob()
: data(NULL),len(0)
{
}
@ -74,6 +76,8 @@ ArgBlob()
~ArgBlob() { if (data) free(data); }
};
class AmDynInvoke;
/** \brief variable type argument for DynInvoke APIs */
class AmArg
: public ArgObject
@ -87,7 +91,8 @@ class AmArg
Bool,
Double,
CStr,
AObject, // for passing pointers to objects not owned by AmArg
AObject, // pointer to an object not owned by AmArg
ADynInv, // pointer to a AmDynInvoke (useful for call backs)
Blob,
Array,
@ -116,6 +121,7 @@ class AmArg
double v_double;
const char* v_cstr;
ArgObject* v_obj;
AmDynInvoke* v_inv;
ArgBlob* v_blob;
ValueArray* v_array;
ValueStruct* v_struct;
@ -169,6 +175,11 @@ class AmArg
v_obj(v)
{ }
AmArg(AmDynInvoke* v)
: type(ADynInv),
v_inv(v)
{ }
// convenience constructors
AmArg(vector<std::string>& v);
AmArg(const vector<int>& v );
@ -196,6 +207,7 @@ class AmArg
#define isArgBool(a) (AmArg::Bool == a.getType())
#define isArgCStr(a) (AmArg::CStr == a.getType())
#define isArgAObject(a) (AmArg::AObject == a.getType())
#define isArgADynInv(a) (AmArg::ADynInv == a.getType())
#define isArgBlob(a) (AmArg::Blob == a.getType())
#define _THROW_TYPE_MISMATCH(exp,got) \
@ -222,6 +234,9 @@ class AmArg
#define assertArgAObject(a) \
if (!isArgAObject(a)) \
_THROW_TYPE_MISMATCH(AObject,a);
#define assertArgADynInv(a) \
if (!isArgADynInv(a)) \
_THROW_TYPE_MISMATCH(ADynInv,a);
#define assertArgBlob(a) \
if (!isArgBlob(a)) \
_THROW_TYPE_MISMATCH(Blob,a);
@ -240,6 +255,7 @@ class AmArg
double asDouble() const { return v_double; }
const char* asCStr() const { return v_cstr; }
ArgObject* asObject() const { return v_obj; }
AmDynInvoke* asDynInv() const { return v_inv; }
ArgBlob* asBlob() const { return v_blob; }
ValueStruct* asStruct() const { return v_struct; }
@ -304,6 +320,7 @@ class AmArg
* f - double
* s - cstr
* o - object
* d - dyninvoke
* b - blob
* a - array
* u - struct

88
core/AmConfig.cpp
Unescape View File

@ -221,6 +221,7 @@ int AmConfig::readConfiguration()
DBG("Reading configuration...\n");
AmConfigReader cfg;
int ret=0;
if(cfg.loadFile(AmConfig::ConfigurationFile.c_str())){
ERROR("while loading main configuration file\n");
@ -230,6 +231,15 @@ int AmConfig::readConfiguration()
// take values from global configuration file
// they will be overwritten by command line args
// stderr
if(cfg.hasParameter("stderr")){
if(!setLogStderr(cfg.getParameter("stderr"), true)){
ERROR("invalid stderr value specified,"
" valid are only yes or no\n");
ret = -1;
}
}
#ifndef DISABLE_SYSLOG_LOG
if (cfg.hasParameter("syslog_facility")) {
set_syslog_facility(cfg.getParameter("syslog_facility").c_str());
@ -251,7 +261,7 @@ int AmConfig::readConfiguration()
if(cfg.hasParameter("sip_port")){
if(!setSIPPort(cfg.getParameter("sip_port").c_str())){
ERROR("invalid sip port specified\n");
return -1;
ret = -1;
}
}
if(cfg.hasParameter("media_ip")) {
@ -313,7 +323,7 @@ int AmConfig::readConfiguration()
if(cfg.hasParameter("loglevel")){
if(!setLogLevel(cfg.getParameter("loglevel"))){
ERROR("invalid log level specified\n");
return -1;
ret = -1;
}
}
@ -350,29 +360,30 @@ int AmConfig::readConfiguration()
if (!appcfg.good()) {
ERROR("could not load application mapping file at '%s'\n",
appcfg_fname.c_str());
return -1;
ret = -1;
}
while (!appcfg.eof()) {
string entry;
getline (appcfg,entry);
if (!entry.length() || entry[0] == '#')
continue;
vector<string> re_v = explode(entry, "=>");
if (re_v.size() != 2) {
ERROR("Incorrect line '%s' in %s: expected format 'regexp=>app_name'\n",
entry.c_str(), appcfg_fname.c_str());
return -1;
else {
while (!appcfg.eof()) {
string entry;
getline (appcfg,entry);
if (!entry.length() || entry[0] == '#')
continue;
vector<string> re_v = explode(entry, "=>");
if (re_v.size() != 2) {
ERROR("Incorrect line '%s' in %s: expected format 'regexp=>app_name'\n",
entry.c_str(), appcfg_fname.c_str());
ret = -1;
}
regex_t app_re;
if (regcomp(&app_re, re_v[0].c_str(), REG_EXTENDED | REG_NOSUB)) {
ERROR("compiling regex '%s' in %s.\n",
re_v[0].c_str(), appcfg_fname.c_str());
ret = -1;
}
DBG("adding application mapping '%s' => '%s'\n",
re_v[0].c_str(),re_v[1].c_str());
AppMapping.push_back(make_pair(app_re, re_v[1]));
}
regex_t app_re;
if (regcomp(&app_re, re_v[0].c_str(), REG_EXTENDED | REG_NOSUB)) {
ERROR("compiling regex '%s' in %s.\n",
re_v[0].c_str(), appcfg_fname.c_str());
return -1;
}
DBG("adding application mapping '%s' => '%s'\n",
re_v[0].c_str(),re_v[1].c_str());
AppMapping.push_back(make_pair(app_re, re_v[1]));
}
} else {
AppSelect = App_SPECIFIED;
@ -385,7 +396,7 @@ int AmConfig::readConfiguration()
if(!setDaemonMode(cfg.getParameter("fork"))){
ERROR("invalid fork value specified,"
" valid are only yes or no\n");
return -1;
ret = -1;
}
}
@ -394,7 +405,7 @@ int AmConfig::readConfiguration()
if(!setDaemonMode(cfg.getParameter("daemon"))){
ERROR("invalid daemon value specified,"
" valid are only yes or no\n");
return -1;
ret = -1;
}
}
@ -408,15 +419,6 @@ int AmConfig::readConfiguration()
#endif /* !DISABLE_DAEMON_MODE */
// stderr
if(cfg.hasParameter("stderr")){
if(!setLogStderr(cfg.getParameter("stderr"), false)){
ERROR("invalid stderr value specified,"
" valid are only yes or no\n");
return -1;
}
}
MaxShutdownTime = cfg.getParameterInt("max_shutdown_time",
DEFAULT_MAX_SHUTDOWN_TIME);
@ -424,7 +426,7 @@ int AmConfig::readConfiguration()
if(cfg.hasParameter("rtp_low_port")){
if(!setRtpLowPort(cfg.getParameter("rtp_low_port"))){
ERROR("invalid rtp low port specified\n");
return -1;
ret = -1;
}
}
@ -432,7 +434,7 @@ int AmConfig::readConfiguration()
if(cfg.hasParameter("rtp_high_port")){
if(!setRtpHighPort(cfg.getParameter("rtp_high_port"))){
ERROR("invalid rtp high port specified\n");
return -1;
ret = -1;
}
}
@ -440,12 +442,12 @@ int AmConfig::readConfiguration()
#ifdef SESSION_THREADPOOL
if(!setSessionProcessorThreads(cfg.getParameter("session_processor_threads"))){
ERROR("invalid session_processor_threads value specified\n");
return -1;
ret = -1;
}
if (SessionProcessorThreads<1) {
ERROR("invalid session_processor_threads value specified."
" need at least one thread\n");
return -1;
ret = -1;
}
#else
WARN("session_processor_threads specified in sems.conf,\n");
@ -458,14 +460,14 @@ int AmConfig::readConfiguration()
if(cfg.hasParameter("media_processor_threads")){
if(!setMediaProcessorThreads(cfg.getParameter("media_processor_threads"))){
ERROR("invalid media_processor_threads value specified");
return -1;
ret = -1;
}
}
if(cfg.hasParameter("sip_server_threads")){
if(!setSIPServerThreads(cfg.getParameter("sip_server_threads"))){
ERROR("invalid sip_server_threads value specified");
return -1;
ret = -1;
}
}
@ -486,7 +488,7 @@ int AmConfig::readConfiguration()
if(cfg.hasParameter("dead_rtp_time")){
if(!setDeadRtpTime(cfg.getParameter("dead_rtp_time"))){
ERROR("invalid dead_rtp_time value specified");
return -1;
ret = -1;
}
}
@ -534,11 +536,11 @@ int AmConfig::readConfiguration()
} else {
ERROR("unknown setting for '100rel' config option: '%s'.\n",
rel100s.c_str());
return -1;
ret = -1;
}
}
INFO("100rel: %d.\n", AmConfig::rel100);
return 0;
return ret;
}

0
core/etc/app_mapping.conf → core/etc/app_mapping.conf.sample
Unescape View File

6
core/sems.cpp
Unescape View File

@ -435,7 +435,11 @@ int main(int argc, char* argv[])
init_logging();
/* load and apply configuration file */
AmConfig::readConfiguration();
if(AmConfig::readConfiguration()){
ERROR("Errors occured while reading configuration file: exiting.");
return -1;
}
log_level = AmConfig::LogLevel;
log_stderr = AmConfig::LogStderr;

Write Preview
Loading…
Cancel
Save