sysmocom
/
asterisk
11
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

major redesign of the channel spy infrastructure, increasing efficiency and reducing locking conflicts 

(nearly) complete rewrite of app_muxmon, renaming the application to MixMonitor and fixing a large number of bugs and inconsistencies
update app_chanspy to use new spy infrastructure


git-svn-id: https://origsvn.digium.com/svn/asterisk/trunk@6884 65c4cc65-6c06-0410-ace0-fbb531ad65f3
This commit is contained in:
Kevin P. Fleming 2005-10-28 23:01:13 +00:00
parent 97c9900b59
commit 846b39a9f1
6 changed files with 1102 additions and 1019 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
.cleancount
View File

@ -1 +1 @@
7
8

293
apps/app_chanspy.c
View File

@ -35,7 +35,6 @@ ASTERISK_FILE_VERSION(__FILE__, "$Revision$")
#include "asterisk/channel.h"
#include "asterisk/features.h"
#include "asterisk/options.h"
#include "asterisk/slinfactory.h"
#include "asterisk/app.h"
#include "asterisk/utils.h"
#include "asterisk/say.h"
@ -68,6 +67,8 @@ static const char *desc = " Chanspy([<scanspec>][|<options>])\n\n"
"(e.g. run Chanspy(Agent) and dial 1234# while spying to jump to channel Agent/1234)\n\n"
"";
static const char *chanspy_spy_type = "ChanSpy";
#define OPTION_QUIET (1 << 0) /* Quiet, no announcement */
#define OPTION_BRIDGED (1 << 1) /* Only look at bridged calls */
#define OPTION_VOLUME (1 << 2) /* Specify initial volume */
@ -88,39 +89,10 @@ LOCAL_USER_DECL;
struct chanspy_translation_helper {
/* spy data */
struct ast_channel_spy spy;
int volfactor;
int fd;
struct ast_slinfactory slinfactory[2];
int volfactor;
};
/* Prototypes */
static struct ast_channel *local_get_channel_begin_name(char *name);
static struct ast_channel *local_channel_walk(struct ast_channel *chan);
static void spy_release(struct ast_channel *chan, void *data);
static void *spy_alloc(struct ast_channel *chan, void *params);
static struct ast_frame *spy_queue_shift(struct ast_channel_spy *spy, int qnum);
static void ast_flush_spy_queue(struct ast_channel_spy *spy);
static int spy_generate(struct ast_channel *chan, void *data, int len, int samples);
static void start_spying(struct ast_channel *chan, struct ast_channel *spychan, struct ast_channel_spy *spy);
static void stop_spying(struct ast_channel *chan, struct ast_channel_spy *spy);
static int channel_spy(struct ast_channel *chan, struct ast_channel *spyee, int *volfactor, int fd);
static int chanspy_exec(struct ast_channel *chan, void *data);
#if 0
static struct ast_channel *local_get_channel_by_name(char *name)
{
struct ast_channel *ret;
ast_mutex_lock(&modlock);
if ((ret = ast_get_channel_by_name_locked(name))) {
ast_mutex_unlock(&ret->lock);
}
ast_mutex_unlock(&modlock);
return ret;
}
#endif
static struct ast_channel *local_channel_walk(struct ast_channel *chan)
{
struct ast_channel *ret;
@ -149,247 +121,85 @@ static struct ast_channel *local_get_channel_begin_name(char *name)
return ret;
}
static void spy_release(struct ast_channel *chan, void *data)
static void *spy_alloc(struct ast_channel *chan, void *data)
{
struct chanspy_translation_helper *csth = data;
ast_slinfactory_destroy(&csth->slinfactory[0]);
ast_slinfactory_destroy(&csth->slinfactory[1]);
return;
/* just store the data pointer in the channel structure */
return data;
}
static void *spy_alloc(struct ast_channel *chan, void *params)
static void spy_release(struct ast_channel *chan, void *data)
{
struct chanspy_translation_helper *csth = params;
ast_slinfactory_init(&csth->slinfactory[0]);
ast_slinfactory_init(&csth->slinfactory[1]);
return params;
/* nothing to do */
}
static struct ast_frame *spy_queue_shift(struct ast_channel_spy *spy, int qnum)
{
struct ast_frame *f;
if (qnum < 0 || qnum > 1)
return NULL;
f = spy->queue[qnum];
if (f) {
spy->queue[qnum] = f->next;
return f;
}
return NULL;
}
static void ast_flush_spy_queue(struct ast_channel_spy *spy)
{
struct ast_frame *f=NULL;
int x = 0;
ast_mutex_lock(&spy->lock);
for(x=0;x<2;x++) {
f = NULL;
while((f = spy_queue_shift(spy, x)))
ast_frfree(f);
}
ast_mutex_unlock(&spy->lock);
}
#if 0
static int extract_audio(short *buf, size_t len, struct ast_trans_pvt *trans, struct ast_frame *fr, int *maxsamp)
{
struct ast_frame *f;
int size, retlen = 0;
if (trans) {
if ((f = ast_translate(trans, fr, 0))) {
size = (f->datalen > len) ? len : f->datalen;
memcpy(buf, f->data, size);
retlen = f->datalen;
ast_frfree(f);
} else {
/* your guess is as good as mine why this will happen but it seems to only happen on iax and appears harmless */
ast_log(LOG_DEBUG, "Failed to translate frame from %s\n", ast_getformatname(fr->subclass));
}
} else {
size = (fr->datalen > len) ? len : fr->datalen;
memcpy(buf, fr->data, size);
retlen = fr->datalen;
}
if (retlen > 0 && (size = retlen / 2)) {
if (size > *maxsamp) {
*maxsamp = size;
}
}
return retlen;
}
static int spy_queue_ready(struct ast_channel_spy *spy)
{
int res = 0;
ast_mutex_lock(&spy->lock);
if (spy->status == CHANSPY_RUNNING) {
res = (spy->queue[0] && spy->queue[1]) ? 1 : 0;
} else {
res = (spy->queue[0] || spy->queue[1]) ? 1 : -1;
}
ast_mutex_unlock(&spy->lock);
return res;
}
#endif
static int spy_generate(struct ast_channel *chan, void *data, int len, int samples)
{
struct chanspy_translation_helper *csth = data;
struct ast_frame frame, *f;
int len0 = 0, len1 = 0, samp0 = 0, samp1 = 0, x, vf, maxsamp;
short buf0[1280], buf1[1280], buf[1280];
struct ast_frame *f;
if (csth->spy.status == CHANSPY_DONE) {
if (csth->spy.status != CHANSPY_RUNNING)
/* Channel is already gone more than likely */
return -1;
}
ast_mutex_lock(&csth->spy.lock);
while((f = csth->spy.queue[0])) {
csth->spy.queue[0] = f->next;
ast_slinfactory_feed(&csth->slinfactory[0], f);
ast_frfree(f);
}
ast_mutex_unlock(&csth->spy.lock);
ast_mutex_lock(&csth->spy.lock);
while((f = csth->spy.queue[1])) {
csth->spy.queue[1] = f->next;
ast_slinfactory_feed(&csth->slinfactory[1], f);
ast_frfree(f);
}
f = ast_channel_spy_read_frame(&csth->spy, samples);
ast_mutex_unlock(&csth->spy.lock);
if (csth->slinfactory[0].size < len || csth->slinfactory[1].size < len) {
if (!f)
return 0;
}
if ((len0 = ast_slinfactory_read(&csth->slinfactory[0], buf0, len))) {
samp0 = len0 / 2;
}
if ((len1 = ast_slinfactory_read(&csth->slinfactory[1], buf1, len))) {
samp1 = len1 / 2;
}
maxsamp = (samp0 > samp1) ? samp0 : samp1;
vf = get_volfactor(csth->volfactor);
for(x=0; x < maxsamp; x++) {
if (vf < 0) {
if (samp0) {
buf0[x] /= abs(vf);
}
if (samp1) {
buf1[x] /= abs(vf);
}
} else if (vf > 0) {
if (samp0) {
buf0[x] *= vf;
}
if (samp1) {
buf1[x] *= vf;
}
}
if (samp0 && samp1) {
if (x < samp0 && x < samp1) {
buf[x] = buf0[x] + buf1[x];
} else if (x < samp0) {
buf[x] = buf0[x];
} else if (x < samp1) {
buf[x] = buf1[x];
}
} else if (x < samp0) {
buf[x] = buf0[x];
} else if (x < samp1) {
buf[x] = buf1[x];
}
}
memset(&frame, 0, sizeof(frame));
frame.frametype = AST_FRAME_VOICE;
frame.subclass = AST_FORMAT_SLINEAR;
frame.data = buf;
frame.samples = x;
frame.datalen = x * 2;
if (ast_write(chan, &frame)) {
if (ast_write(chan, f)) {
ast_frfree(f);
return -1;
}
if (csth->fd) {
write(csth->fd, buf1, len1);
}
if (csth->fd)
write(csth->fd, f->data, f->datalen);
ast_frfree(f);
return 0;
}
static struct ast_generator spygen = {
alloc: spy_alloc,
release: spy_release,
generate: spy_generate,
.alloc = spy_alloc,
.release = spy_release,
.generate = spy_generate,
};
static void start_spying(struct ast_channel *chan, struct ast_channel *spychan, struct ast_channel_spy *spy)
static int start_spying(struct ast_channel *chan, struct ast_channel *spychan, struct ast_channel_spy *spy)
{
struct ast_channel_spy *cptr=NULL;
int res;
struct ast_channel *peer;
ast_log(LOG_WARNING, "Attaching %s to %s\n", spychan->name, chan->name);
ast_log(LOG_NOTICE, "Attaching %s to %s\n", spychan->name, chan->name);
ast_mutex_lock(&chan->lock);
if (chan->spiers) {
for(cptr=chan->spiers;cptr && cptr->next;cptr=cptr->next);
cptr->next = spy;
} else {
chan->spiers = spy;
}
res = ast_channel_spy_add(chan, spy);
ast_mutex_unlock(&chan->lock);
if ( ast_test_flag(chan, AST_FLAG_NBRIDGE) && (peer = ast_bridged_channel(chan))) {
if (!res && ast_test_flag(chan, AST_FLAG_NBRIDGE) && (peer = ast_bridged_channel(chan))) {
ast_softhangup(peer, AST_SOFTHANGUP_UNBRIDGE);
}
return res;
}
static void stop_spying(struct ast_channel *chan, struct ast_channel_spy *spy)
{
struct ast_channel_spy *cptr=NULL, *prev=NULL;
/* If our status has changed, then the channel we're spying on is gone....
DON'T TOUCH IT!!! RUN AWAY!!! */
if (spy->status != CHANSPY_RUNNING)
return;
ast_mutex_lock(&chan->lock);
for(cptr=chan->spiers; cptr; cptr=cptr->next) {
if (cptr == spy) {
if (prev) {
prev->next = cptr->next;
cptr->next = NULL;
} else
chan->spiers = NULL;
}
prev = cptr;
}
ast_mutex_unlock(&chan->lock);
if (!chan)
return;
}
ast_mutex_lock(&chan->lock);
ast_channel_spy_remove(chan, spy);
ast_mutex_unlock(&chan->lock);
};
/* Map 'volume' levels from -4 through +4 into
decibel (dB) settings for channel drivers
@ -414,35 +224,47 @@ static void set_volume(struct ast_channel *chan, struct chanspy_translation_help
{
signed char volume_adjust = volfactor_map[csth->volfactor + 4];
if (!ast_channel_setoption(chan, AST_OPTION_TXGAIN, &volume_adjust, sizeof(volume_adjust), 0)) {
if (!ast_channel_setoption(chan, AST_OPTION_TXGAIN, &volume_adjust, sizeof(volume_adjust), 0))
csth->volfactor = 0;
}
}
static int channel_spy(struct ast_channel *chan, struct ast_channel *spyee, int *volfactor, int fd)
{
struct chanspy_translation_helper csth;
int running = 1, res = 0, x = 0;
int running, res = 0, x = 0;
char inp[24];
char *name=NULL;
struct ast_frame *f;
if (chan && !ast_check_hangup(chan) && spyee && !ast_check_hangup(spyee)) {
running = (chan && !ast_check_hangup(chan) && spyee && !ast_check_hangup(spyee));
if (running) {
memset(inp, 0, sizeof(inp));
name = ast_strdupa(spyee->name);
if (option_verbose >= 2)
ast_verbose(VERBOSE_PREFIX_2 "Spying on channel %s\n", name);
memset(&csth, 0, sizeof(csth));
ast_set_flag(&csth.spy, CHANSPY_FORMAT_AUDIO);
ast_set_flag(&csth.spy, CHANSPY_TRIGGER_NONE);
ast_set_flag(&csth.spy, CHANSPY_MIXAUDIO);
csth.spy.type = chanspy_spy_type;
csth.spy.status = CHANSPY_RUNNING;
csth.spy.read_queue.format = AST_FORMAT_SLINEAR;
csth.spy.write_queue.format = AST_FORMAT_SLINEAR;
ast_mutex_init(&csth.spy.lock);
csth.volfactor = *volfactor;
set_volume(chan, &csth);
if (fd) {
csth.fd = fd;
}
start_spying(spyee, chan, &csth.spy);
csth.spy.read_vol_adjustment = csth.volfactor;
csth.spy.write_vol_adjustment = csth.volfactor;
csth.fd = fd;
if (start_spying(spyee, chan, &csth.spy))
running = 0;
}
if (running) {
running = 1;
ast_activate_generator(chan, &spygen, &csth);
while (csth.spy.status == CHANSPY_RUNNING &&
@ -487,6 +309,8 @@ static int channel_spy(struct ast_channel *chan, struct ast_channel *spyee, int
}
csth.volfactor = *volfactor;
set_volume(chan, &csth);
csth.spy.read_vol_adjustment = csth.volfactor;
csth.spy.write_vol_adjustment = csth.volfactor;
}
} else if (res >= 48 && res <= 57) {
inp[x++] = res;
@ -498,11 +322,12 @@ static int channel_spy(struct ast_channel *chan, struct ast_channel *spyee, int
if (option_verbose >= 2) {
ast_verbose(VERBOSE_PREFIX_2 "Done Spying on channel %s\n", name);
}
ast_flush_spy_queue(&csth.spy);
} else {
running = 0;
}
ast_mutex_destroy(&csth.spy.lock);
return running;
}

617
apps/app_mixmonitor.c
View File

@ -1,46 +1,64 @@
/*
* Asterisk -- A telephony toolkit for Linux.
*
* Asterisk -- An open source telephony toolkit.
*
* Copyright (C) 2005, Anthony Minessale II
* Copyright (C) 2005, Digium, Inc.
*
* Mark Spencer <markster@digium.com>
* Kevin P. Fleming <kpfleming@digium.com>
*
* Based on app_muxmon.c provided by
* Anthony Minessale II <anthmct@yahoo.com>
*
* See http://www.asterisk.org for more information about
* the Asterisk project. Please do not directly contact
* any of the maintainers of this project for assistance;
* the project provides a web site, mailing lists and IRC
* channels for your use.
*
* This program is free software, distributed under the terms of
* the GNU General Public License
* the GNU General Public License Version 2. See the LICENSE file
* at the top of the source tree.
*/
/*! \file
* \brief muxmon() - record a call natively
*/
#include <asterisk/file.h>
#include <asterisk/logger.h>
#include <asterisk/channel.h>
#include <asterisk/pbx.h>
#include <asterisk/module.h>
#include <asterisk/lock.h>
#include <asterisk/cli.h>
#include <asterisk/options.h>
#include <asterisk/app.h>
#include <asterisk/translate.h>
#include <asterisk/slinfactory.h>
#include <stdlib.h>
#include <unistd.h>
#include <string.h>
#define get_volfactor(x) x ? ((x > 0) ? (1 << x) : ((1 << abs(x)) * -1)) : 0
#define minmax(x,y) x ? (x > y) ? y : ((x < (y * -1)) ? (y * -1) : x) : 0
static char *tdesc = "Native Channel Monitoring Module";
static char *app = "MuxMon";
static char *synopsis = "Record A Call Natively";
static char *desc = ""
" MuxMon(<file>.<ext>[|<options>[|<command>]])\n\n"
"Records The audio on the current channel to the specified file.\n\n"
"Valid Options:\n"
" b - Only save audio to the file while the channel is bridged. Note: does\n"
" not include conferences\n"
" a - Append to the file instead of overwriting it.\n"
#include "asterisk.h"
ASTERISK_FILE_VERSION(__FILE__, "$Revision$")
#include "asterisk/file.h"
#include "asterisk/logger.h"
#include "asterisk/channel.h"
#include "asterisk/pbx.h"
#include "asterisk/module.h"
#include "asterisk/lock.h"
#include "asterisk/cli.h"
#include "asterisk/options.h"
#include "asterisk/app.h"
#include "asterisk/linkedlists.h"
#define get_volfactor(x) x ? ((x > 0) ? (1 << x) : ((1 << abs(x)) * -1)) : 0
static const char *tdesc = "Mixed Audio Monitoring Application";
static const char *app = "MixMonitor";
static const char *synopsis = "Record a call and mix the audio during the recording";
static const char *desc = ""
" MixMonitor(<file>.<ext>[|<options>[|<command>]])\n\n"
"Records the audio on the current channel to the specified file.\n"
"If the filename is an absolute path, uses that path, otherwise\n"
"creates the file in the configured monitoring directory from\n"
"asterisk.conf.\n\n"
"Valid options:\n"
" a - Append to the file instead of overwriting it.\n"
" b - Only save audio to the file while the channel is bridged.\n"
" Note: does not include conferences.\n"
" v(<x>) - Adjust the heard volume by a factor of <x> (range -4 to 4)\n"
" V(<x>) - Adjust the spoken volume by a factor of <x> (range -4 to 4)\n"
" W(<x>) - Adjust the both heard and spoken volumes by a factor of <x>\n"
@ -48,14 +66,16 @@ static char *desc = ""
"<command> will be executed when the recording is over\n"
"Any strings matching ^{X} will be unescaped to ${X} and \n"
"all variables will be evaluated at that time.\n"
"The variable MUXMON_FILENAME will contain the filename used to record.\n"
"The variable MIXMONITOR_FILENAME will contain the filename used to record.\n"
"";
STANDARD_LOCAL_USER;
LOCAL_USER_DECL;
struct muxmon {
static const char *mixmonitor_spy_type = "MixMonitor";
struct mixmonitor {
struct ast_channel *chan;
char *filename;
char *post_process;
@ -64,445 +84,341 @@ struct muxmon {
int writevol;
};
typedef enum {
MUXFLAG_RUNNING = (1 << 0),
enum {
MUXFLAG_APPEND = (1 << 1),
MUXFLAG_BRIDGED = (1 << 2),
MUXFLAG_VOLUME = (1 << 3),
MUXFLAG_READVOLUME = (1 << 4),
MUXFLAG_WRITEVOLUME = (1 << 5)
} muxflags;
MUXFLAG_WRITEVOLUME = (1 << 5),
} mixmonitor_flags;
AST_DECLARE_OPTIONS(muxmon_opts,{
['a'] = { MUXFLAG_APPEND },
AST_DECLARE_OPTIONS(mixmonitor_opts,{
['a'] = { MUXFLAG_APPEND },
['b'] = { MUXFLAG_BRIDGED },
['v'] = { MUXFLAG_READVOLUME, 1 },
['V'] = { MUXFLAG_WRITEVOLUME, 2 },
['W'] = { MUXFLAG_VOLUME, 3 },
});
static void stopmon(struct ast_channel *chan, struct ast_channel_spy *spy)
{
struct ast_channel_spy *cptr=NULL, *prev=NULL;
int count = 0;
/* If our status has changed, then the channel we're spying on is gone....
DON'T TOUCH IT!!! RUN AWAY!!! */
if (spy->status != CHANSPY_RUNNING)
return;
if (chan) {
while(ast_mutex_trylock(&chan->lock)) {
if (chan->spiers == spy) {
chan->spiers = NULL;
return;
}
count++;
if (count > 10) {
return;
}
sched_yield();
}
for(cptr=chan->spiers; cptr; cptr=cptr->next) {
if (cptr == spy) {
if (prev) {
prev->next = cptr->next;
cptr->next = NULL;
} else
chan->spiers = NULL;
}
prev = cptr;
}
if (!chan)
return;
ast_mutex_unlock(&chan->lock);
}
ast_mutex_lock(&chan->lock);
ast_channel_spy_remove(chan, spy);
ast_mutex_unlock(&chan->lock);
}
static void startmon(struct ast_channel *chan, struct ast_channel_spy *spy)
static int startmon(struct ast_channel *chan, struct ast_channel_spy *spy)
{
struct ast_channel_spy *cptr=NULL;
struct ast_channel *peer;
int res;
if (chan) {
ast_mutex_lock(&chan->lock);
if (chan->spiers) {
for(cptr=chan->spiers;cptr->next;cptr=cptr->next);
cptr->next = spy;
} else {
chan->spiers = spy;
}
ast_mutex_unlock(&chan->lock);
if (!chan)
return -1;
ast_mutex_lock(&chan->lock);
res = ast_channel_spy_add(chan, spy);
ast_mutex_unlock(&chan->lock);
if (ast_test_flag(chan, AST_FLAG_NBRIDGE) && (peer = ast_bridged_channel(chan))) {
ast_softhangup(peer, AST_SOFTHANGUP_UNBRIDGE);
}
}
}
if (!res && ast_test_flag(chan, AST_FLAG_NBRIDGE) && (peer = ast_bridged_channel(chan)))
ast_softhangup(peer, AST_SOFTHANGUP_UNBRIDGE);
static int spy_queue_translate(struct ast_channel_spy *spy,
struct ast_slinfactory *slinfactory0,
struct ast_slinfactory *slinfactory1)
{
int res = 0;
struct ast_frame *f;
ast_mutex_lock(&spy->lock);
while((f = spy->queue[0])) {
spy->queue[0] = f->next;
ast_slinfactory_feed(slinfactory0, f);
ast_frfree(f);
}
ast_mutex_unlock(&spy->lock);
ast_mutex_lock(&spy->lock);
while((f = spy->queue[1])) {
spy->queue[1] = f->next;
ast_slinfactory_feed(slinfactory1, f);
ast_frfree(f);
}
ast_mutex_unlock(&spy->lock);
return res;
}
static void *muxmon_thread(void *obj)
{
#define SAMPLES_PER_FRAME 160
int len0 = 0, len1 = 0, samp0 = 0, samp1 = 0, framelen, maxsamp = 0, x = 0;
short buf0[1280], buf1[1280], buf[1280];
struct ast_frame frame;
struct muxmon *muxmon = obj;
static void *mixmonitor_thread(void *obj)
{
struct mixmonitor *mixmonitor = obj;
struct ast_channel_spy spy;
struct ast_filestream *fs = NULL;
char *ext, *name;
unsigned int oflags;
struct ast_slinfactory slinfactory[2];
struct ast_frame *f;
char post_process[1024] = "";
name = ast_strdupa(muxmon->chan->name);
STANDARD_INCREMENT_USECOUNT;
name = ast_strdupa(mixmonitor->chan->name);
framelen = 320;
frame.frametype = AST_FRAME_VOICE;
frame.subclass = AST_FORMAT_SLINEAR;
frame.data = buf;
ast_set_flag(muxmon, MUXFLAG_RUNNING);
oflags = O_CREAT|O_WRONLY;
ast_slinfactory_init(&slinfactory[0]);
ast_slinfactory_init(&slinfactory[1]);
/* for efficiency, use a flag to bypass volume logic when it's not needed */
if (muxmon->readvol || muxmon->writevol) {
ast_set_flag(muxmon, MUXFLAG_VOLUME);
}
if ((ext = strchr(muxmon->filename, '.'))) {
oflags |= ast_test_flag(mixmonitor, MUXFLAG_APPEND) ? O_APPEND : O_TRUNC;
if ((ext = strchr(mixmonitor->filename, '.'))) {
*(ext++) = '\0';
} else {
ext = "raw";
}
memset(&spy, 0, sizeof(spy));
spy.status = CHANSPY_RUNNING;
ast_mutex_init(&spy.lock);
startmon(muxmon->chan, &spy);
if (ast_test_flag(muxmon, MUXFLAG_RUNNING)) {
if (option_verbose > 1) {
ast_verbose(VERBOSE_PREFIX_2 "Begin Muxmon Recording %s\n", name);
}
oflags |= ast_test_flag(muxmon, MUXFLAG_APPEND) ? O_APPEND : O_TRUNC;
if (!(fs = ast_writefile(muxmon->filename, ext, NULL, oflags, 0, 0644))) {
ast_log(LOG_ERROR, "Cannot open %s\n", muxmon->filename);
spy.status = CHANSPY_DONE;
} else {
if (ast_test_flag(muxmon, MUXFLAG_APPEND)) {
ast_seekstream(fs, 0, SEEK_END);
}
while (ast_test_flag(muxmon, MUXFLAG_RUNNING)) {
samp0 = samp1 = len0 = len1 = 0;
if (ast_check_hangup(muxmon->chan) || spy.status != CHANSPY_RUNNING) {
ast_clear_flag(muxmon, MUXFLAG_RUNNING);
break;
}
if (ast_test_flag(muxmon, MUXFLAG_BRIDGED) && !ast_bridged_channel(muxmon->chan)) {
usleep(1000);
sched_yield();
continue;
}
spy_queue_translate(&spy, &slinfactory[0], &slinfactory[1]);
if (slinfactory[0].size < framelen || slinfactory[1].size < framelen) {
usleep(1000);
sched_yield();
continue;
}
if ((len0 = ast_slinfactory_read(&slinfactory[0], buf0, framelen))) {
samp0 = len0 / 2;
}
if((len1 = ast_slinfactory_read(&slinfactory[1], buf1, framelen))) {
samp1 = len1 / 2;
}
if (ast_test_flag(muxmon, MUXFLAG_VOLUME)) {
if (samp0 && muxmon->readvol > 0) {
for(x=0; x < samp0 / 2; x++) {
buf0[x] *= muxmon->readvol;
}
} else if (samp0 && muxmon->readvol < 0) {
for(x=0; x < samp0 / 2; x++) {
buf0[x] /= muxmon->readvol;
}
}
if (samp1 && muxmon->writevol > 0) {
for(x=0; x < samp1 / 2; x++) {
buf1[x] *= muxmon->writevol;
}
} else if (muxmon->writevol < 0) {
for(x=0; x < samp1 / 2; x++) {
buf1[x] /= muxmon->writevol;
}
}
}
maxsamp = (samp0 > samp1) ? samp0 : samp1;
if (samp0 && samp1) {
for(x=0; x < maxsamp; x++) {
if (x < samp0 && x < samp1) {
buf[x] = buf0[x] + buf1[x];
} else if (x < samp0) {
buf[x] = buf0[x];
} else if (x < samp1) {
buf[x] = buf1[x];
}
}
} else if(samp0) {
memcpy(buf, buf0, len0);
x = samp0;
} else if(samp1) {
memcpy(buf, buf1, len1);
x = samp1;
}
frame.samples = x;
frame.datalen = x * 2;
ast_writestream(fs, &frame);
usleep(1000);
sched_yield();
}
}
fs = ast_writefile(mixmonitor->filename, ext, NULL, oflags, 0, 0644);
if (!fs) {
ast_log(LOG_ERROR, "Cannot open %s.%s\n", mixmonitor->filename, ext);
goto out;
}
if (muxmon->post_process) {
if (ast_test_flag(mixmonitor, MUXFLAG_APPEND))
ast_seekstream(fs, 0, SEEK_END);
memset(&spy, 0, sizeof(spy));
ast_set_flag(&spy, CHANSPY_FORMAT_AUDIO);
ast_set_flag(&spy, CHANSPY_MIXAUDIO);
spy.type = mixmonitor_spy_type;
spy.status = CHANSPY_RUNNING;
spy.read_queue.format = AST_FORMAT_SLINEAR;
spy.write_queue.format = AST_FORMAT_SLINEAR;
if (mixmonitor->readvol) {
ast_set_flag(&spy, CHANSPY_READ_VOLADJUST);
spy.read_vol_adjustment = mixmonitor->readvol;
}
if (mixmonitor->writevol) {
ast_set_flag(&spy, CHANSPY_WRITE_VOLADJUST);
spy.write_vol_adjustment = mixmonitor->writevol;
}
ast_mutex_init(&spy.lock);
if (startmon(mixmonitor->chan, &spy)) {
ast_log(LOG_WARNING, "Unable to add '%s' spy to channel '%s'\n",
spy.type, mixmonitor->chan->name);
goto out2;
}
if (option_verbose > 1)
ast_verbose(VERBOSE_PREFIX_2 "Begin MixMonitor Recording %s\n", name);
while (1) {
struct ast_frame *next;
int write;
ast_mutex_lock(&spy.lock);
ast_channel_spy_trigger_wait(&spy);
if (ast_check_hangup(mixmonitor->chan) || spy.status != CHANSPY_RUNNING) {
ast_mutex_unlock(&spy.lock);
break;
}
while (1) {
if (!(f = ast_channel_spy_read_frame(&spy, SAMPLES_PER_FRAME)))
break;
write = (!ast_test_flag(mixmonitor, MUXFLAG_BRIDGED) ||
ast_bridged_channel(mixmonitor->chan));
/* it is possible for ast_channel_spy_read_frame() to return a chain
of frames if a queue flush was necessary, so process them
*/
for (; f; f = next) {
next = f->next;
if (write)
ast_writestream(fs, f);
ast_frfree(f);
}
}
ast_mutex_unlock(&spy.lock);
}
if (mixmonitor->post_process) {
char *p;
for(p = muxmon->post_process; *p ; p++) {
for (p = mixmonitor->post_process; *p ; p++) {
if (*p == '^' && *(p+1) == '{') {
*p = '$';
}
}
pbx_substitute_variables_helper(muxmon->chan, muxmon->post_process, post_process, sizeof(post_process) - 1);
free(muxmon->post_process);
muxmon->post_process = NULL;
pbx_substitute_variables_helper(mixmonitor->chan, mixmonitor->post_process, post_process, sizeof(post_process) - 1);
}
stopmon(muxmon->chan, &spy);
if (option_verbose > 1) {
ast_verbose(VERBOSE_PREFIX_2 "Finished Recording %s\n", name);
}
ast_mutex_destroy(&spy.lock);
if(fs) {
ast_closestream(fs);
}
ast_slinfactory_destroy(&slinfactory[0]);
ast_slinfactory_destroy(&slinfactory[1]);
stopmon(mixmonitor->chan, &spy);
if (muxmon) {
if (muxmon->filename) {
free(muxmon->filename);
}
free(muxmon);
}
if (option_verbose > 1)
ast_verbose(VERBOSE_PREFIX_2 "End MixMonitor Recording %s\n", name);
if (!ast_strlen_zero(post_process)) {
if (option_verbose > 2) {
if (option_verbose > 2)
ast_verbose(VERBOSE_PREFIX_2 "Executing [%s]\n", post_process);
}
ast_safe_system(post_process);
}
out2:
ast_mutex_destroy(&spy.lock);
if (fs)
ast_closestream(fs);
out:
free(mixmonitor);
STANDARD_DECREMENT_USECOUNT;
return NULL;
}
static void launch_monitor_thread(struct ast_channel *chan, char *filename, unsigned int flags, int readvol , int writevol, char *post_process)
static void launch_monitor_thread(struct ast_channel *chan, const char *filename, unsigned int flags,
int readvol, int writevol, const char *post_process)
{
pthread_attr_t attr;
int result = 0;
pthread_t thread;
struct muxmon *muxmon;
struct mixmonitor *mixmonitor;
int len;
len = sizeof(*mixmonitor) + strlen(filename) + 1;
if (post_process && !ast_strlen_zero(post_process))
len += strlen(post_process) + 1;
if (!(muxmon = malloc(sizeof(struct muxmon)))) {
if (!(mixmonitor = calloc(1, len))) {
ast_log(LOG_ERROR, "Memory Error!\n");
return;
}
memset(muxmon, 0, sizeof(struct muxmon));
muxmon->chan = chan;
muxmon->filename = strdup(filename);
if(post_process) {
muxmon->post_process = strdup(post_process);
mixmonitor->chan = chan;
mixmonitor->filename = (char *) mixmonitor + sizeof(*mixmonitor);
strcpy(mixmonitor->filename, filename);
if (post_process && !ast_strlen_zero(post_process)) {
mixmonitor->post_process = mixmonitor->filename + strlen(filename) + 1;
strcpy(mixmonitor->post_process, post_process);
}
muxmon->readvol = readvol;
muxmon->writevol = writevol;
muxmon->flags = flags;
mixmonitor->readvol = readvol;
mixmonitor->writevol = writevol;
mixmonitor->flags = flags;
result = pthread_attr_init(&attr);
pthread_attr_setschedpolicy(&attr, SCHED_RR);
pthread_attr_init(&attr);
pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
result = ast_pthread_create(&thread, &attr, muxmon_thread, muxmon);
result = pthread_attr_destroy(&attr);
ast_pthread_create(&thread, &attr, mixmonitor_thread, mixmonitor);
pthread_attr_destroy(&attr);
}
static int muxmon_exec(struct ast_channel *chan, void *data)
static int mixmonitor_exec(struct ast_channel *chan, void *data)
{
int res = 0, x = 0, readvol = 0, writevol = 0;
int x, readvol = 0, writevol = 0;
struct localuser *u;
struct ast_flags flags = {0};
int argc;
char *options = NULL,
*args,
*argv[3],
*filename = NULL,
*post_process = NULL;
char *parse;
AST_DECLARE_APP_ARGS(args,
AST_APP_ARG(filename);
AST_APP_ARG(options);
AST_APP_ARG(post_process);
);
if (ast_strlen_zero(data)) {
ast_log(LOG_WARNING, "muxmon requires an argument\n");
ast_log(LOG_WARNING, "MixMonitor requires an argument (filename)\n");
return -1;
}
LOCAL_USER_ADD(u);
args = ast_strdupa(data);
if (!args) {
if (!(parse = ast_strdupa(data))) {
ast_log(LOG_WARNING, "Memory Error!\n");
LOCAL_USER_REMOVE(u);
return -1;
}
if ((argc = ast_separate_app_args(args, '|', argv, sizeof(argv) / sizeof(argv[0])))) {
filename = argv[0];
if (argc > 1) {
options = argv[1];
}
if (argc > 2) {
post_process = argv[2];
}
}
AST_STANDARD_APP_ARGS(args, parse);
if (ast_strlen_zero(filename)) {
if (ast_strlen_zero(args.filename)) {
ast_log(LOG_WARNING, "Muxmon requires an argument (filename)\n");
LOCAL_USER_REMOVE(u);
return -1;
}
if (options) {
char *opts[3] = {};
ast_parseoptions(muxmon_opts, &flags, opts, options);
if (args.options) {
char *opts[3] = { NULL, };
if (ast_test_flag(&flags, MUXFLAG_READVOLUME) && opts[0]) {
if (sscanf(opts[0], "%d", &x) != 1)
ast_log(LOG_NOTICE, "volume must be a number between -4 and 4\n");
else {
readvol = minmax(x, 4);
x = get_volfactor(readvol);
readvol = minmax(x, 16);
ast_parseoptions(mixmonitor_opts, &flags, opts, args.options);
if (ast_test_flag(&flags, MUXFLAG_READVOLUME)) {
if (!opts[0] || ast_strlen_zero(opts[0])) {
ast_log(LOG_WARNING, "No volume level was provided for the heard volume ('v') option.\n");
} else if ((sscanf(opts[0], "%d", &x) != 1) || (x < -4) || (x > 4)) {
ast_log(LOG_NOTICE, "Heard volume must be a number between -4 and 4, not '%s'\n", opts[0]);
} else {
readvol = get_volfactor(x);
}
}
if (ast_test_flag(&flags, MUXFLAG_WRITEVOLUME) && opts[1]) {
if (sscanf(opts[1], "%d", &x) != 1)
ast_log(LOG_NOTICE, "volume must be a number between -4 and 4\n");
else {
writevol = minmax(x, 4);
x = get_volfactor(writevol);
writevol = minmax(x, 16);
if (ast_test_flag(&flags, MUXFLAG_WRITEVOLUME)) {
if (!opts[1] || ast_strlen_zero(opts[1])) {
ast_log(LOG_WARNING, "No volume level was provided for the spoken volume ('V') option.\n");
} else if ((sscanf(opts[1], "%d", &x) != 1) || (x < -4) || (x > 4)) {
ast_log(LOG_NOTICE, "Spoken volume must be a number between -4 and 4, not '%s'\n", opts[1]);
} else {
writevol = get_volfactor(x);
}
}
if (ast_test_flag(&flags, MUXFLAG_VOLUME) && opts[2]) {
if (sscanf(opts[2], "%d", &x) != 1)
ast_log(LOG_NOTICE, "volume must be a number between -4 and 4\n");
else {
readvol = writevol = minmax(x, 4);
x = get_volfactor(readvol);
readvol = minmax(x, 16);
x = get_volfactor(writevol);
writevol = minmax(x, 16);
if (ast_test_flag(&flags, MUXFLAG_VOLUME)) {
if (!opts[2] || ast_strlen_zero(opts[2])) {
ast_log(LOG_WARNING, "No volume level was provided for the combined volume ('W') option.\n");
} else if ((sscanf(opts[2], "%d", &x) != 1) || (x < -4) || (x > 4)) {
ast_log(LOG_NOTICE, "Combined volume must be a number between -4 and 4, not '%s'\n", opts[2]);
} else {
readvol = writevol = get_volfactor(x);
}
}
}
pbx_builtin_setvar_helper(chan, "MUXMON_FILENAME", filename);
launch_monitor_thread(chan, filename, flags.flags, readvol, writevol, post_process);
/* if not provided an absolute path, use the system-configured monitoring directory */
if (args.filename[0] != '/') {
char *build;
build = alloca(strlen(ast_config_AST_MONITOR_DIR) + strlen(args.filename) + 3);
sprintf(build, "%s/%s", ast_config_AST_MONITOR_DIR, args.filename);
args.filename = build;
}
pbx_builtin_setvar_helper(chan, "MIXMONITOR_FILENAME", args.filename);
launch_monitor_thread(chan, args.filename, flags.flags, readvol, writevol, args.post_process);
LOCAL_USER_REMOVE(u);
return res;
return 0;
}
static int muxmon_cli(int fd, int argc, char **argv)
static int mixmonitor_cli(int fd, int argc, char **argv)
{
char *op, *chan_name = NULL, *args = NULL;
struct ast_channel *chan;
if (argc > 2) {
op = argv[1];
chan_name = argv[2];
if (argc < 3)
return RESULT_SHOWUSAGE;
if (argv[3]) {
args = argv[3];
}
if (!(chan = ast_get_channel_by_name_prefix_locked(chan_name, strlen(chan_name)))) {
ast_cli(fd, "Invalid Channel!\n");
return -1;
}
if (!strcasecmp(op, "start")) {
muxmon_exec(chan, args);
} else if (!strcasecmp(op, "stop")) {
struct ast_channel_spy *cptr=NULL;
for(cptr=chan->spiers; cptr; cptr=cptr->next) {
cptr->status = CHANSPY_DONE;
}
}
ast_mutex_unlock(&chan->lock);
return 0;
if (!(chan = ast_get_channel_by_name_prefix_locked(argv[2], strlen(argv[2])))) {
ast_cli(fd, "No channel matching '%s' found.\n", argv[2]);
return RESULT_SUCCESS;
}
ast_cli(fd, "Usage: muxmon <start|stop> <chan_name> <args>\n");
return -1;
if (!strcasecmp(argv[1], "start"))
mixmonitor_exec(chan, argv[3]);
else if (!strcasecmp(argv[1], "stop"))
ast_channel_spy_stop_by_type(chan, mixmonitor_spy_type);
ast_mutex_unlock(&chan->lock);
return RESULT_SUCCESS;
}
static struct ast_cli_entry cli_muxmon = {
{ "muxmon", NULL, NULL }, muxmon_cli,
"Execute a monitor command", "muxmon <start|stop> <chan_name> <args>"};
static struct ast_cli_entry cli_mixmonitor = {
{ "mixmonitor", NULL, NULL },
mixmonitor_cli,
"Execute a MixMonitor command",
"mixmonitor <start|stop> <chan_name> [<args>]"
};
int unload_module(void)
{
int res;
res = ast_cli_unregister(&cli_muxmon);
res = ast_cli_unregister(&cli_mixmonitor);
res |= ast_unregister_application(app);
STANDARD_HANGUP_LOCALUSERS;
@ -514,21 +430,23 @@ int load_module(void)
{
int res;
res = ast_cli_register(&cli_muxmon);
res |= ast_register_application(app, muxmon_exec, synopsis, desc);
res = ast_cli_register(&cli_mixmonitor);
res |= ast_register_application(app, mixmonitor_exec, synopsis, desc);
return res;
}
char *description(void)
{
return tdesc;
return (char *) tdesc;
}
int usecount(void)
{
int res;
STANDARD_USECOUNT(res);
return res;
}
@ -536,4 +454,3 @@ char *key()
{
return ASTERISK_GPL_KEY;
}

617
apps/app_muxmon.c
View File

@ -1,46 +1,64 @@
/*
* Asterisk -- A telephony toolkit for Linux.
*
* Asterisk -- An open source telephony toolkit.
*
* Copyright (C) 2005, Anthony Minessale II
* Copyright (C) 2005, Digium, Inc.
*
* Mark Spencer <markster@digium.com>
* Kevin P. Fleming <kpfleming@digium.com>
*
* Based on app_muxmon.c provided by
* Anthony Minessale II <anthmct@yahoo.com>
*
* See http://www.asterisk.org for more information about
* the Asterisk project. Please do not directly contact
* any of the maintainers of this project for assistance;
* the project provides a web site, mailing lists and IRC
* channels for your use.
*
* This program is free software, distributed under the terms of
* the GNU General Public License
* the GNU General Public License Version 2. See the LICENSE file
* at the top of the source tree.
*/
/*! \file
* \brief muxmon() - record a call natively
*/
#include <asterisk/file.h>
#include <asterisk/logger.h>
#include <asterisk/channel.h>
#include <asterisk/pbx.h>
#include <asterisk/module.h>
#include <asterisk/lock.h>
#include <asterisk/cli.h>
#include <asterisk/options.h>
#include <asterisk/app.h>
#include <asterisk/translate.h>
#include <asterisk/slinfactory.h>
#include <stdlib.h>
#include <unistd.h>
#include <string.h>
#define get_volfactor(x) x ? ((x > 0) ? (1 << x) : ((1 << abs(x)) * -1)) : 0
#define minmax(x,y) x ? (x > y) ? y : ((x < (y * -1)) ? (y * -1) : x) : 0
static char *tdesc = "Native Channel Monitoring Module";
static char *app = "MuxMon";
static char *synopsis = "Record A Call Natively";
static char *desc = ""
" MuxMon(<file>.<ext>[|<options>[|<command>]])\n\n"
"Records The audio on the current channel to the specified file.\n\n"
"Valid Options:\n"
" b - Only save audio to the file while the channel is bridged. Note: does\n"
" not include conferences\n"
" a - Append to the file instead of overwriting it.\n"
#include "asterisk.h"
ASTERISK_FILE_VERSION(__FILE__, "$Revision$")
#include "asterisk/file.h"
#include "asterisk/logger.h"
#include "asterisk/channel.h"
#include "asterisk/pbx.h"
#include "asterisk/module.h"
#include "asterisk/lock.h"
#include "asterisk/cli.h"
#include "asterisk/options.h"
#include "asterisk/app.h"
#include "asterisk/linkedlists.h"
#define get_volfactor(x) x ? ((x > 0) ? (1 << x) : ((1 << abs(x)) * -1)) : 0
static const char *tdesc = "Mixed Audio Monitoring Application";
static const char *app = "MixMonitor";
static const char *synopsis = "Record a call and mix the audio during the recording";
static const char *desc = ""
" MixMonitor(<file>.<ext>[|<options>[|<command>]])\n\n"
"Records the audio on the current channel to the specified file.\n"
"If the filename is an absolute path, uses that path, otherwise\n"
"creates the file in the configured monitoring directory from\n"
"asterisk.conf.\n\n"
"Valid options:\n"
" a - Append to the file instead of overwriting it.\n"
" b - Only save audio to the file while the channel is bridged.\n"
" Note: does not include conferences.\n"
" v(<x>) - Adjust the heard volume by a factor of <x> (range -4 to 4)\n"
" V(<x>) - Adjust the spoken volume by a factor of <x> (range -4 to 4)\n"
" W(<x>) - Adjust the both heard and spoken volumes by a factor of <x>\n"
@ -48,14 +66,16 @@ static char *desc = ""
"<command> will be executed when the recording is over\n"
"Any strings matching ^{X} will be unescaped to ${X} and \n"
"all variables will be evaluated at that time.\n"
"The variable MUXMON_FILENAME will contain the filename used to record.\n"
"The variable MIXMONITOR_FILENAME will contain the filename used to record.\n"
"";
STANDARD_LOCAL_USER;
LOCAL_USER_DECL;
struct muxmon {
static const char *mixmonitor_spy_type = "MixMonitor";
struct mixmonitor {
struct ast_channel *chan;
char *filename;
char *post_process;
@ -64,445 +84,341 @@ struct muxmon {
int writevol;
};
typedef enum {
MUXFLAG_RUNNING = (1 << 0),
enum {
MUXFLAG_APPEND = (1 << 1),
MUXFLAG_BRIDGED = (1 << 2),
MUXFLAG_VOLUME = (1 << 3),
MUXFLAG_READVOLUME = (1 << 4),
MUXFLAG_WRITEVOLUME = (1 << 5)
} muxflags;
MUXFLAG_WRITEVOLUME = (1 << 5),
} mixmonitor_flags;
AST_DECLARE_OPTIONS(muxmon_opts,{
['a'] = { MUXFLAG_APPEND },
AST_DECLARE_OPTIONS(mixmonitor_opts,{
['a'] = { MUXFLAG_APPEND },
['b'] = { MUXFLAG_BRIDGED },
['v'] = { MUXFLAG_READVOLUME, 1 },
['V'] = { MUXFLAG_WRITEVOLUME, 2 },
['W'] = { MUXFLAG_VOLUME, 3 },
});
static void stopmon(struct ast_channel *chan, struct ast_channel_spy *spy)
{
struct ast_channel_spy *cptr=NULL, *prev=NULL;
int count = 0;
/* If our status has changed, then the channel we're spying on is gone....
DON'T TOUCH IT!!! RUN AWAY!!! */
if (spy->status != CHANSPY_RUNNING)
return;
if (chan) {
while(ast_mutex_trylock(&chan->lock)) {
if (chan->spiers == spy) {
chan->spiers = NULL;
return;
}
count++;
if (count > 10) {
return;
}
sched_yield();
}
for(cptr=chan->spiers; cptr; cptr=cptr->next) {
if (cptr == spy) {
if (prev) {
prev->next = cptr->next;
cptr->next = NULL;
} else
chan->spiers = NULL;
}
prev = cptr;
}
if (!chan)
return;
ast_mutex_unlock(&chan->lock);
}
ast_mutex_lock(&chan->lock);
ast_channel_spy_remove(chan, spy);
ast_mutex_unlock(&chan->lock);
}
static void startmon(struct ast_channel *chan, struct ast_channel_spy *spy)
static int startmon(struct ast_channel *chan, struct ast_channel_spy *spy)
{
struct ast_channel_spy *cptr=NULL;
struct ast_channel *peer;
int res;
if (chan) {
ast_mutex_lock(&chan->lock);
if (chan->spiers) {
for(cptr=chan->spiers;cptr->next;cptr=cptr->next);
cptr->next = spy;
} else {
chan->spiers = spy;
}
ast_mutex_unlock(&chan->lock);
if (!chan)
return -1;
ast_mutex_lock(&chan->lock);
res = ast_channel_spy_add(chan, spy);
ast_mutex_unlock(&chan->lock);
if (ast_test_flag(chan, AST_FLAG_NBRIDGE) && (peer = ast_bridged_channel(chan))) {
ast_softhangup(peer, AST_SOFTHANGUP_UNBRIDGE);
}
}
}
if (!res && ast_test_flag(chan, AST_FLAG_NBRIDGE) && (peer = ast_bridged_channel(chan)))
ast_softhangup(peer, AST_SOFTHANGUP_UNBRIDGE);
static int spy_queue_translate(struct ast_channel_spy *spy,
struct ast_slinfactory *slinfactory0,
struct ast_slinfactory *slinfactory1)
{
int res = 0;
struct ast_frame *f;
ast_mutex_lock(&spy->lock);
while((f = spy->queue[0])) {
spy->queue[0] = f->next;
ast_slinfactory_feed(slinfactory0, f);
ast_frfree(f);
}
ast_mutex_unlock(&spy->lock);
ast_mutex_lock(&spy->lock);
while((f = spy->queue[1])) {
spy->queue[1] = f->next;
ast_slinfactory_feed(slinfactory1, f);
ast_frfree(f);
}
ast_mutex_unlock(&spy->lock);
return res;
}
static void *muxmon_thread(void *obj)
{
#define SAMPLES_PER_FRAME 160
int len0 = 0, len1 = 0, samp0 = 0, samp1 = 0, framelen, maxsamp = 0, x = 0;
short buf0[1280], buf1[1280], buf[1280];
struct ast_frame frame;
struct muxmon *muxmon = obj;
static void *mixmonitor_thread(void *obj)
{
struct mixmonitor *mixmonitor = obj;
struct ast_channel_spy spy;
struct ast_filestream *fs = NULL;
char *ext, *name;
unsigned int oflags;
struct ast_slinfactory slinfactory[2];
struct ast_frame *f;
char post_process[1024] = "";
name = ast_strdupa(muxmon->chan->name);
STANDARD_INCREMENT_USECOUNT;
name = ast_strdupa(mixmonitor->chan->name);
framelen = 320;
frame.frametype = AST_FRAME_VOICE;
frame.subclass = AST_FORMAT_SLINEAR;
frame.data = buf;
ast_set_flag(muxmon, MUXFLAG_RUNNING);
oflags = O_CREAT|O_WRONLY;
ast_slinfactory_init(&slinfactory[0]);
ast_slinfactory_init(&slinfactory[1]);
/* for efficiency, use a flag to bypass volume logic when it's not needed */
if (muxmon->readvol || muxmon->writevol) {
ast_set_flag(muxmon, MUXFLAG_VOLUME);
}
if ((ext = strchr(muxmon->filename, '.'))) {
oflags |= ast_test_flag(mixmonitor, MUXFLAG_APPEND) ? O_APPEND : O_TRUNC;
if ((ext = strchr(mixmonitor->filename, '.'))) {
*(ext++) = '\0';
} else {
ext = "raw";
}
memset(&spy, 0, sizeof(spy));
spy.status = CHANSPY_RUNNING;
ast_mutex_init(&spy.lock);
startmon(muxmon->chan, &spy);
if (ast_test_flag(muxmon, MUXFLAG_RUNNING)) {
if (option_verbose > 1) {
ast_verbose(VERBOSE_PREFIX_2 "Begin Muxmon Recording %s\n", name);
}
oflags |= ast_test_flag(muxmon, MUXFLAG_APPEND) ? O_APPEND : O_TRUNC;
if (!(fs = ast_writefile(muxmon->filename, ext, NULL, oflags, 0, 0644))) {
ast_log(LOG_ERROR, "Cannot open %s\n", muxmon->filename);
spy.status = CHANSPY_DONE;
} else {
if (ast_test_flag(muxmon, MUXFLAG_APPEND)) {
ast_seekstream(fs, 0, SEEK_END);
}
while (ast_test_flag(muxmon, MUXFLAG_RUNNING)) {
samp0 = samp1 = len0 = len1 = 0;
if (ast_check_hangup(muxmon->chan) || spy.status != CHANSPY_RUNNING) {
ast_clear_flag(muxmon, MUXFLAG_RUNNING);
break;
}
if (ast_test_flag(muxmon, MUXFLAG_BRIDGED) && !ast_bridged_channel(muxmon->chan)) {
usleep(1000);
sched_yield();
continue;
}
spy_queue_translate(&spy, &slinfactory[0], &slinfactory[1]);
if (slinfactory[0].size < framelen || slinfactory[1].size < framelen) {
usleep(1000);
sched_yield();
continue;
}
if ((len0 = ast_slinfactory_read(&slinfactory[0], buf0, framelen))) {
samp0 = len0 / 2;
}
if((len1 = ast_slinfactory_read(&slinfactory[1], buf1, framelen))) {
samp1 = len1 / 2;
}
if (ast_test_flag(muxmon, MUXFLAG_VOLUME)) {
if (samp0 && muxmon->readvol > 0) {
for(x=0; x < samp0 / 2; x++) {
buf0[x] *= muxmon->readvol;
}
} else if (samp0 && muxmon->readvol < 0) {
for(x=0; x < samp0 / 2; x++) {
buf0[x] /= muxmon->readvol;
}
}
if (samp1 && muxmon->writevol > 0) {
for(x=0; x < samp1 / 2; x++) {
buf1[x] *= muxmon->writevol;
}
} else if (muxmon->writevol < 0) {
for(x=0; x < samp1 / 2; x++) {
buf1[x] /= muxmon->writevol;
}
}
}
maxsamp = (samp0 > samp1) ? samp0 : samp1;
if (samp0 && samp1) {
for(x=0; x < maxsamp; x++) {
if (x < samp0 && x < samp1) {
buf[x] = buf0[x] + buf1[x];
} else if (x < samp0) {
buf[x] = buf0[x];
} else if (x < samp1) {
buf[x] = buf1[x];
}
}
} else if(samp0) {
memcpy(buf, buf0, len0);
x = samp0;
} else if(samp1) {
memcpy(buf, buf1, len1);
x = samp1;
}
frame.samples = x;
frame.datalen = x * 2;
ast_writestream(fs, &frame);
usleep(1000);
sched_yield();
}
}
fs = ast_writefile(mixmonitor->filename, ext, NULL, oflags, 0, 0644);
if (!fs) {
ast_log(LOG_ERROR, "Cannot open %s.%s\n", mixmonitor->filename, ext);
goto out;
}
if (muxmon->post_process) {
if (ast_test_flag(mixmonitor, MUXFLAG_APPEND))
ast_seekstream(fs, 0, SEEK_END);
memset(&spy, 0, sizeof(spy));
ast_set_flag(&spy, CHANSPY_FORMAT_AUDIO);
ast_set_flag(&spy, CHANSPY_MIXAUDIO);
spy.type = mixmonitor_spy_type;
spy.status = CHANSPY_RUNNING;
spy.read_queue.format = AST_FORMAT_SLINEAR;
spy.write_queue.format = AST_FORMAT_SLINEAR;
if (mixmonitor->readvol) {
ast_set_flag(&spy, CHANSPY_READ_VOLADJUST);
spy.read_vol_adjustment = mixmonitor->readvol;
}
if (mixmonitor->writevol) {
ast_set_flag(&spy, CHANSPY_WRITE_VOLADJUST);
spy.write_vol_adjustment = mixmonitor->writevol;
}
ast_mutex_init(&spy.lock);
if (startmon(mixmonitor->chan, &spy)) {
ast_log(LOG_WARNING, "Unable to add '%s' spy to channel '%s'\n",
spy.type, mixmonitor->chan->name);
goto out2;
}
if (option_verbose > 1)
ast_verbose(VERBOSE_PREFIX_2 "Begin MixMonitor Recording %s\n", name);
while (1) {
struct ast_frame *next;
int write;
ast_mutex_lock(&spy.lock);
ast_channel_spy_trigger_wait(&spy);
if (ast_check_hangup(mixmonitor->chan) || spy.status != CHANSPY_RUNNING) {
ast_mutex_unlock(&spy.lock);
break;
}
while (1) {
if (!(f = ast_channel_spy_read_frame(&spy, SAMPLES_PER_FRAME)))
break;
write = (!ast_test_flag(mixmonitor, MUXFLAG_BRIDGED) ||
ast_bridged_channel(mixmonitor->chan));
/* it is possible for ast_channel_spy_read_frame() to return a chain
of frames if a queue flush was necessary, so process them
*/
for (; f; f = next) {
next = f->next;
if (write)
ast_writestream(fs, f);
ast_frfree(f);
}
}
ast_mutex_unlock(&spy.lock);
}
if (mixmonitor->post_process) {
char *p;
for(p = muxmon->post_process; *p ; p++) {
for (p = mixmonitor->post_process; *p ; p++) {
if (*p == '^' && *(p+1) == '{') {
*p = '$';
}
}
pbx_substitute_variables_helper(muxmon->chan, muxmon->post_process, post_process, sizeof(post_process) - 1);
free(muxmon->post_process);
muxmon->post_process = NULL;
pbx_substitute_variables_helper(mixmonitor->chan, mixmonitor->post_process, post_process, sizeof(post_process) - 1);
}
stopmon(muxmon->chan, &spy);
if (option_verbose > 1) {
ast_verbose(VERBOSE_PREFIX_2 "Finished Recording %s\n", name);
}
ast_mutex_destroy(&spy.lock);
if(fs) {
ast_closestream(fs);
}
ast_slinfactory_destroy(&slinfactory[0]);
ast_slinfactory_destroy(&slinfactory[1]);
stopmon(mixmonitor->chan, &spy);
if (muxmon) {
if (muxmon->filename) {
free(muxmon->filename);
}
free(muxmon);
}
if (option_verbose > 1)
ast_verbose(VERBOSE_PREFIX_2 "End MixMonitor Recording %s\n", name);
if (!ast_strlen_zero(post_process)) {
if (option_verbose > 2) {
if (option_verbose > 2)
ast_verbose(VERBOSE_PREFIX_2 "Executing [%s]\n", post_process);
}
ast_safe_system(post_process);
}
out2:
ast_mutex_destroy(&spy.lock);
if (fs)
ast_closestream(fs);
out:
free(mixmonitor);
STANDARD_DECREMENT_USECOUNT;
return NULL;
}
static void launch_monitor_thread(struct ast_channel *chan, char *filename, unsigned int flags, int readvol , int writevol, char *post_process)
static void launch_monitor_thread(struct ast_channel *chan, const char *filename, unsigned int flags,
int readvol, int writevol, const char *post_process)
{
pthread_attr_t attr;
int result = 0;
pthread_t thread;
struct muxmon *muxmon;
struct mixmonitor *mixmonitor;
int len;
len = sizeof(*mixmonitor) + strlen(filename) + 1;
if (post_process && !ast_strlen_zero(post_process))
len += strlen(post_process) + 1;
if (!(muxmon = malloc(sizeof(struct muxmon)))) {
if (!(mixmonitor = calloc(1, len))) {
ast_log(LOG_ERROR, "Memory Error!\n");
return;
}
memset(muxmon, 0, sizeof(struct muxmon));
muxmon->chan = chan;
muxmon->filename = strdup(filename);
if(post_process) {
muxmon->post_process = strdup(post_process);
mixmonitor->chan = chan;
mixmonitor->filename = (char *) mixmonitor + sizeof(*mixmonitor);
strcpy(mixmonitor->filename, filename);
if (post_process && !ast_strlen_zero(post_process)) {
mixmonitor->post_process = mixmonitor->filename + strlen(filename) + 1;
strcpy(mixmonitor->post_process, post_process);
}
muxmon->readvol = readvol;
muxmon->writevol = writevol;
muxmon->flags = flags;
mixmonitor->readvol = readvol;
mixmonitor->writevol = writevol;
mixmonitor->flags = flags;
result = pthread_attr_init(&attr);
pthread_attr_setschedpolicy(&attr, SCHED_RR);
pthread_attr_init(&attr);
pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
result = ast_pthread_create(&thread, &attr, muxmon_thread, muxmon);
result = pthread_attr_destroy(&attr);
ast_pthread_create(&thread, &attr, mixmonitor_thread, mixmonitor);
pthread_attr_destroy(&attr);
}
static int muxmon_exec(struct ast_channel *chan, void *data)
static int mixmonitor_exec(struct ast_channel *chan, void *data)
{
int res = 0, x = 0, readvol = 0, writevol = 0;
int x, readvol = 0, writevol = 0;
struct localuser *u;
struct ast_flags flags = {0};
int argc;
char *options = NULL,
*args,
*argv[3],
*filename = NULL,
*post_process = NULL;
char *parse;
AST_DECLARE_APP_ARGS(args,
AST_APP_ARG(filename);
AST_APP_ARG(options);
AST_APP_ARG(post_process);
);
if (ast_strlen_zero(data)) {
ast_log(LOG_WARNING, "muxmon requires an argument\n");
ast_log(LOG_WARNING, "MixMonitor requires an argument (filename)\n");
return -1;
}
LOCAL_USER_ADD(u);
args = ast_strdupa(data);
if (!args) {
if (!(parse = ast_strdupa(data))) {
ast_log(LOG_WARNING, "Memory Error!\n");
LOCAL_USER_REMOVE(u);
return -1;
}
if ((argc = ast_separate_app_args(args, '|', argv, sizeof(argv) / sizeof(argv[0])))) {
filename = argv[0];
if (argc > 1) {
options = argv[1];
}
if (argc > 2) {
post_process = argv[2];
}
}
AST_STANDARD_APP_ARGS(args, parse);
if (ast_strlen_zero(filename)) {
if (ast_strlen_zero(args.filename)) {
ast_log(LOG_WARNING, "Muxmon requires an argument (filename)\n");
LOCAL_USER_REMOVE(u);
return -1;
}
if (options) {
char *opts[3] = {};
ast_parseoptions(muxmon_opts, &flags, opts, options);
if (args.options) {
char *opts[3] = { NULL, };
if (ast_test_flag(&flags, MUXFLAG_READVOLUME) && opts[0]) {
if (sscanf(opts[0], "%d", &x) != 1)
ast_log(LOG_NOTICE, "volume must be a number between -4 and 4\n");
else {
readvol = minmax(x, 4);
x = get_volfactor(readvol);
readvol = minmax(x, 16);
ast_parseoptions(mixmonitor_opts, &flags, opts, args.options);
if (ast_test_flag(&flags, MUXFLAG_READVOLUME)) {
if (!opts[0] || ast_strlen_zero(opts[0])) {
ast_log(LOG_WARNING, "No volume level was provided for the heard volume ('v') option.\n");
} else if ((sscanf(opts[0], "%d", &x) != 1) || (x < -4) || (x > 4)) {
ast_log(LOG_NOTICE, "Heard volume must be a number between -4 and 4, not '%s'\n", opts[0]);
} else {
readvol = get_volfactor(x);
}
}
if (ast_test_flag(&flags, MUXFLAG_WRITEVOLUME) && opts[1]) {
if (sscanf(opts[1], "%d", &x) != 1)
ast_log(LOG_NOTICE, "volume must be a number between -4 and 4\n");
else {
writevol = minmax(x, 4);
x = get_volfactor(writevol);
writevol = minmax(x, 16);
if (ast_test_flag(&flags, MUXFLAG_WRITEVOLUME)) {
if (!opts[1] || ast_strlen_zero(opts[1])) {
ast_log(LOG_WARNING, "No volume level was provided for the spoken volume ('V') option.\n");
} else if ((sscanf(opts[1], "%d", &x) != 1) || (x < -4) || (x > 4)) {
ast_log(LOG_NOTICE, "Spoken volume must be a number between -4 and 4, not '%s'\n", opts[1]);
} else {
writevol = get_volfactor(x);
}
}
if (ast_test_flag(&flags, MUXFLAG_VOLUME) && opts[2]) {
if (sscanf(opts[2], "%d", &x) != 1)
ast_log(LOG_NOTICE, "volume must be a number between -4 and 4\n");
else {
readvol = writevol = minmax(x, 4);
x = get_volfactor(readvol);
readvol = minmax(x, 16);
x = get_volfactor(writevol);
writevol = minmax(x, 16);
if (ast_test_flag(&flags, MUXFLAG_VOLUME)) {
if (!opts[2] || ast_strlen_zero(opts[2])) {
ast_log(LOG_WARNING, "No volume level was provided for the combined volume ('W') option.\n");
} else if ((sscanf(opts[2], "%d", &x) != 1) || (x < -4) || (x > 4)) {
ast_log(LOG_NOTICE, "Combined volume must be a number between -4 and 4, not '%s'\n", opts[2]);
} else {
readvol = writevol = get_volfactor(x);
}
}
}
pbx_builtin_setvar_helper(chan, "MUXMON_FILENAME", filename);
launch_monitor_thread(chan, filename, flags.flags, readvol, writevol, post_process);
/* if not provided an absolute path, use the system-configured monitoring directory */
if (args.filename[0] != '/') {
char *build;
build = alloca(strlen(ast_config_AST_MONITOR_DIR) + strlen(args.filename) + 3);
sprintf(build, "%s/%s", ast_config_AST_MONITOR_DIR, args.filename);
args.filename = build;
}
pbx_builtin_setvar_helper(chan, "MIXMONITOR_FILENAME", args.filename);
launch_monitor_thread(chan, args.filename, flags.flags, readvol, writevol, args.post_process);
LOCAL_USER_REMOVE(u);
return res;
return 0;
}
static int muxmon_cli(int fd, int argc, char **argv)
static int mixmonitor_cli(int fd, int argc, char **argv)
{
char *op, *chan_name = NULL, *args = NULL;
struct ast_channel *chan;
if (argc > 2) {
op = argv[1];
chan_name = argv[2];
if (argc < 3)
return RESULT_SHOWUSAGE;
if (argv[3]) {
args = argv[3];
}
if (!(chan = ast_get_channel_by_name_prefix_locked(chan_name, strlen(chan_name)))) {
ast_cli(fd, "Invalid Channel!\n");
return -1;
}
if (!strcasecmp(op, "start")) {
muxmon_exec(chan, args);
} else if (!strcasecmp(op, "stop")) {
struct ast_channel_spy *cptr=NULL;
for(cptr=chan->spiers; cptr; cptr=cptr->next) {
cptr->status = CHANSPY_DONE;
}
}
ast_mutex_unlock(&chan->lock);
return 0;
if (!(chan = ast_get_channel_by_name_prefix_locked(argv[2], strlen(argv[2])))) {
ast_cli(fd, "No channel matching '%s' found.\n", argv[2]);
return RESULT_SUCCESS;
}
ast_cli(fd, "Usage: muxmon <start|stop> <chan_name> <args>\n");
return -1;
if (!strcasecmp(argv[1], "start"))
mixmonitor_exec(chan, argv[3]);
else if (!strcasecmp(argv[1], "stop"))
ast_channel_spy_stop_by_type(chan, mixmonitor_spy_type);
ast_mutex_unlock(&chan->lock);
return RESULT_SUCCESS;
}
static struct ast_cli_entry cli_muxmon = {
{ "muxmon", NULL, NULL }, muxmon_cli,
"Execute a monitor command", "muxmon <start|stop> <chan_name> <args>"};
static struct ast_cli_entry cli_mixmonitor = {
{ "mixmonitor", NULL, NULL },
mixmonitor_cli,
"Execute a MixMonitor command",
"mixmonitor <start|stop> <chan_name> [<args>]"
};
int unload_module(void)
{
int res;
res = ast_cli_unregister(&cli_muxmon);
res = ast_cli_unregister(&cli_mixmonitor);
res |= ast_unregister_application(app);
STANDARD_HANGUP_LOCALUSERS;
@ -514,21 +430,23 @@ int load_module(void)
{
int res;
res = ast_cli_register(&cli_muxmon);
res |= ast_register_application(app, muxmon_exec, synopsis, desc);
res = ast_cli_register(&cli_mixmonitor);
res |= ast_register_application(app, mixmonitor_exec, synopsis, desc);
return res;
}
char *description(void)
{
return tdesc;
return (char *) tdesc;
}
int usecount(void)
{
int res;
STANDARD_USECOUNT(res);
return res;
}
@ -536,4 +454,3 @@ char *key()
{
return ASTERISK_GPL_KEY;
}

477
channel.c
View File

@ -71,6 +71,17 @@ ASTERISK_FILE_VERSION(__FILE__, "$Revision$")
#include "asterisk/transcap.h"
#include "asterisk/devicestate.h"
struct channel_spy_trans {
int last_format;
struct ast_trans_pvt *path;
};
struct ast_channel_spy_list {
struct channel_spy_trans read_translator;
struct channel_spy_trans write_translator;
AST_LIST_HEAD_NOLOCK(, ast_channel_spy) list;
};
/* uncomment if you have problems with 'monitoring' synchronized files */
#if 0
#define MONITOR_CONSTANT_DELAY
@ -931,10 +942,8 @@ void ast_channel_free(struct ast_channel *chan)
/* loop over the variables list, freeing all data and deleting list items */
/* no need to lock the list, as the channel is already locked */
while (!AST_LIST_EMPTY(headp)) { /* List Deletion. */
vardata = AST_LIST_REMOVE_HEAD(headp, entries);
ast_var_delete(vardata);
}
while ((vardata = AST_LIST_REMOVE_HEAD(headp, entries)))
ast_var_delete(vardata);
free(chan);
ast_mutex_unlock(&chlock);
@ -942,19 +951,134 @@ void ast_channel_free(struct ast_channel *chan)
ast_device_state_changed_literal(name);
}
static void ast_spy_detach(struct ast_channel *chan)
int ast_channel_spy_add(struct ast_channel *chan, struct ast_channel_spy *spy)
{
struct ast_channel_spy *chanspy;
/* Marking the spies as done is sufficient. Chanspy or spy users will get the picture. */
for (chanspy = chan->spiers; chanspy; chanspy = chanspy->next) {
if (chanspy->status == CHANSPY_RUNNING) {
chanspy->status = CHANSPY_DONE;
}
if (!ast_test_flag(spy, CHANSPY_FORMAT_AUDIO)) {
ast_log(LOG_WARNING, "Could not add channel spy '%s' to channel '%s', only audio format spies are supported.\n",
spy->type, chan->name);
return -1;
}
chan->spiers = NULL;
return;
if (ast_test_flag(spy, CHANSPY_READ_VOLADJUST) && (spy->read_queue.format != AST_FORMAT_SLINEAR)) {
ast_log(LOG_WARNING, "Cannot provide volume adjustment on '%s' format spies\n",
ast_getformatname(spy->read_queue.format));
return -1;
}
if (ast_test_flag(spy, CHANSPY_WRITE_VOLADJUST) && (spy->write_queue.format != AST_FORMAT_SLINEAR)) {
ast_log(LOG_WARNING, "Cannot provide volume adjustment on '%s' format spies\n",
ast_getformatname(spy->write_queue.format));
return -1;
}
if (ast_test_flag(spy, CHANSPY_MIXAUDIO) &&
((spy->read_queue.format != AST_FORMAT_SLINEAR) ||
(spy->write_queue.format != AST_FORMAT_SLINEAR))) {
ast_log(LOG_WARNING, "Cannot provide audio mixing on '%s'-'%s' format spies\n",
ast_getformatname(spy->read_queue.format), ast_getformatname(spy->write_queue.format));
return -1;
}
if (!chan->spies) {
if (!(chan->spies = calloc(1, sizeof(*chan->spies)))) {
ast_log(LOG_WARNING, "Memory allocation failure\n");
return -1;
}
AST_LIST_HEAD_INIT_NOLOCK(&chan->spies->list);
AST_LIST_INSERT_HEAD(&chan->spies->list, spy, list);
} else {
AST_LIST_INSERT_TAIL(&chan->spies->list, spy, list);
}
if (ast_test_flag(spy, CHANSPY_TRIGGER_MODE) != CHANSPY_TRIGGER_NONE) {
ast_cond_init(&spy->trigger, NULL);
ast_set_flag(spy, CHANSPY_TRIGGER_READ);
ast_clear_flag(spy, CHANSPY_TRIGGER_WRITE);
}
ast_log(LOG_DEBUG, "Spy %s added to channel %s\n",
spy->type, chan->name);
return 0;
}
void ast_channel_spy_stop_by_type(struct ast_channel *chan, const char *type)
{
struct ast_channel_spy *spy;
if (!chan->spies)
return;
AST_LIST_TRAVERSE(&chan->spies->list, spy, list) {
if ((spy->type == type) && (spy->status == CHANSPY_RUNNING))
spy->status = CHANSPY_DONE;
}
}
void ast_channel_spy_trigger_wait(struct ast_channel_spy *spy)
{
ast_cond_wait(&spy->trigger, &spy->lock);
}
void ast_channel_spy_remove(struct ast_channel *chan, struct ast_channel_spy *spy)
{
struct ast_frame *f;
if (!chan->spies)
return;
AST_LIST_REMOVE(&chan->spies->list, spy, list);
ast_mutex_lock(&spy->lock);
for (f = spy->read_queue.head; f; f = spy->read_queue.head) {
spy->read_queue.head = f->next;
ast_frfree(f);
}
for (f = spy->write_queue.head; f; f = spy->write_queue.head) {
spy->write_queue.head = f->next;
ast_frfree(f);
}
if (ast_test_flag(spy, CHANSPY_TRIGGER_MODE) != CHANSPY_TRIGGER_NONE)
ast_cond_destroy(&spy->trigger);
ast_mutex_unlock(&spy->lock);
ast_log(LOG_DEBUG, "Spy %s removed from channel %s\n",
spy->type, chan->name);
if (AST_LIST_EMPTY(&chan->spies->list)) {
if (chan->spies->read_translator.path)
ast_translator_free_path(chan->spies->read_translator.path);
if (chan->spies->write_translator.path)
ast_translator_free_path(chan->spies->write_translator.path);
free(chan->spies);
chan->spies = NULL;
}
}
static void detach_spies(struct ast_channel *chan)
{
struct ast_channel_spy *spy;
if (!chan->spies)
return;
/* Marking the spies as done is sufficient. Chanspy or spy users will get the picture. */
AST_LIST_TRAVERSE(&chan->spies->list, spy, list) {
ast_mutex_lock(&spy->lock);
if (spy->status == CHANSPY_RUNNING)
spy->status = CHANSPY_DONE;
if (ast_test_flag(spy, CHANSPY_TRIGGER_MODE) != CHANSPY_TRIGGER_NONE)
ast_cond_signal(&spy->trigger);
ast_mutex_unlock(&spy->lock);
}
AST_LIST_TRAVERSE_SAFE_BEGIN(&chan->spies->list, spy, list)
ast_channel_spy_remove(chan, spy);
AST_LIST_TRAVERSE_SAFE_END;
}
/*--- ast_softhangup_nolock: Softly hangup a channel, don't lock */
@ -983,40 +1107,136 @@ int ast_softhangup(struct ast_channel *chan, int cause)
return res;
}
static void ast_queue_spy_frame(struct ast_channel_spy *spy, struct ast_frame *f, int pos)
enum spy_direction {
SPY_READ,
SPY_WRITE,
};
#define SPY_QUEUE_SAMPLE_LIMIT 4000 /* half of one second */
static void queue_frame_to_spies(struct ast_channel *chan, struct ast_frame *f, enum spy_direction dir)
{
struct ast_frame *tmpf = NULL;
int count = 0;
struct ast_frame *translated_frame = NULL;
struct ast_channel_spy *spy;
struct ast_channel_spy_queue *queue;
struct ast_channel_spy_queue *other_queue;
struct channel_spy_trans *trans;
struct ast_frame *last;
ast_mutex_lock(&spy->lock);
for (tmpf=spy->queue[pos]; tmpf && tmpf->next; tmpf=tmpf->next) {
count++;
}
if (count > 1000) {
struct ast_frame *freef, *headf;
trans = (dir == SPY_READ) ? &chan->spies->read_translator : &chan->spies->write_translator;
ast_log(LOG_ERROR, "Too many frames queued at once, flushing cache.\n");
headf = spy->queue[pos];
/* deref the queue right away so it looks empty */
spy->queue[pos] = NULL;
tmpf = headf;
/* free the wasted frames */
while (tmpf) {
freef = tmpf;
tmpf = tmpf->next;
ast_frfree(freef);
AST_LIST_TRAVERSE(&chan->spies->list, spy, list) {
ast_mutex_lock(&spy->lock);
queue = (dir == SPY_READ) ? &spy->read_queue : &spy->write_queue;
if ((queue->format == AST_FORMAT_SLINEAR) && (f->subclass != AST_FORMAT_SLINEAR)) {
if (!translated_frame) {
if (trans->path && (trans->last_format != f->subclass)) {
ast_translator_free_path(trans->path);
trans->path = NULL;
}
if (!trans->path) {
ast_log(LOG_DEBUG, "Building translator from %s to SLINEAR for spies on channel %s\n",
ast_getformatname(f->subclass), chan->name);
if ((trans->path = ast_translator_build_path(AST_FORMAT_SLINEAR, f->subclass)) == NULL) {
ast_log(LOG_WARNING, "Cannot build a path from %s to %s\n",
ast_getformatname(f->subclass), ast_getformatname(AST_FORMAT_SLINEAR));
ast_mutex_unlock(&spy->lock);
continue;
} else {
trans->last_format = f->subclass;
}
}
translated_frame = ast_translate(trans->path, f, 0);
}
for (last = queue->head; last && last->next; last = last->next);
if (last)
last->next = ast_frdup(translated_frame);
else
queue->head = ast_frdup(translated_frame);
} else {
if (f->subclass != queue->format) {
ast_log(LOG_WARNING, "Spy '%s' on channel '%s' wants format '%s', but frame is '%s', dropping\n",
spy->type, chan->name,
ast_getformatname(queue->format), ast_getformatname(f->subclass));
ast_mutex_unlock(&spy->lock);
continue;
}
for (last = queue->head; last && last->next; last = last->next);
if (last)
last->next = ast_frdup(f);
else
queue->head = ast_frdup(f);
}
queue->samples += f->samples;
if (queue->samples > SPY_QUEUE_SAMPLE_LIMIT) {
if (ast_test_flag(spy, CHANSPY_TRIGGER_MODE) != CHANSPY_TRIGGER_NONE) {
other_queue = (dir == SPY_WRITE) ? &spy->read_queue : &spy->write_queue;
if (other_queue->samples == 0) {
switch (ast_test_flag(spy, CHANSPY_TRIGGER_MODE)) {
case CHANSPY_TRIGGER_READ:
if (dir == SPY_WRITE) {
ast_set_flag(spy, CHANSPY_TRIGGER_WRITE);
ast_clear_flag(spy, CHANSPY_TRIGGER_READ);
if (option_debug)
ast_log(LOG_DEBUG, "Switching spy '%s' on '%s' to write-trigger mode\n",
spy->type, chan->name);
}
break;
case CHANSPY_TRIGGER_WRITE:
if (dir == SPY_READ) {
ast_set_flag(spy, CHANSPY_TRIGGER_READ);
ast_clear_flag(spy, CHANSPY_TRIGGER_WRITE);
if (option_debug)
ast_log(LOG_DEBUG, "Switching spy '%s' on '%s' to read-trigger mode\n",
spy->type, chan->name);
}
break;
}
if (option_debug)
ast_log(LOG_DEBUG, "Triggering queue flush for spy '%s' on '%s'\n",
spy->type, chan->name);
ast_set_flag(spy, CHANSPY_TRIGGER_FLUSH);
ast_cond_signal(&spy->trigger);
ast_mutex_unlock(&spy->lock);
continue;
}
}
if (option_debug)
ast_log(LOG_DEBUG, "Spy '%s' on channel '%s' %s queue too long, dropping frames\n",
spy->type, chan->name, (dir == SPY_READ) ? "read" : "write");
while (queue->samples > SPY_QUEUE_SAMPLE_LIMIT) {
struct ast_frame *drop = queue->head;
queue->samples -= drop->samples;
queue->head = drop->next;
ast_frfree(drop);
}
} else {
switch (ast_test_flag(spy, CHANSPY_TRIGGER_MODE)) {
case CHANSPY_TRIGGER_READ:
if (dir == SPY_READ)
ast_cond_signal(&spy->trigger);
break;
case CHANSPY_TRIGGER_WRITE:
if (dir == SPY_WRITE)
ast_cond_signal(&spy->trigger);
break;
}
}
ast_mutex_unlock(&spy->lock);
return;
}
if (tmpf) {
tmpf->next = ast_frdup(f);
} else {
spy->queue[pos] = ast_frdup(f);
}
ast_mutex_unlock(&spy->lock);
if (translated_frame)
ast_frfree(translated_frame);
}
static void free_translation(struct ast_channel *clone)
@ -1040,7 +1260,7 @@ int ast_hangup(struct ast_channel *chan)
if someone is going to masquerade as us */
ast_mutex_lock(&chan->lock);
ast_spy_detach(chan); /* get rid of spies */
detach_spies(chan); /* get rid of spies */
if (chan->masq) {
if (ast_do_masquerade(chan))
@ -1174,20 +1394,28 @@ static int generator_force(void *data)
int ast_activate_generator(struct ast_channel *chan, struct ast_generator *gen, void *params)
{
int res = 0;
ast_mutex_lock(&chan->lock);
if (chan->generatordata) {
if (chan->generator && chan->generator->release)
chan->generator->release(chan, chan->generatordata);
chan->generatordata = NULL;
}
ast_prod(chan);
if ((chan->generatordata = gen->alloc(chan, params))) {
if (gen->alloc) {
if (!(chan->generatordata = gen->alloc(chan, params)))
res = -1;
}
if (!res) {
ast_settimeout(chan, 160, generator_force, chan);
chan->generator = gen;
} else {
res = -1;
}
ast_mutex_unlock(&chan->lock);
return res;
}
@ -1661,12 +1889,9 @@ struct ast_frame *ast_read(struct ast_channel *chan)
ast_frfree(f);
f = &null_frame;
} else {
if (chan->spiers) {
struct ast_channel_spy *spying;
for (spying = chan->spiers; spying; spying=spying->next) {
ast_queue_spy_frame(spying, f, 0);
}
}
if (chan->spies)
queue_frame_to_spies(chan, f, SPY_READ);
if (chan->monitor && chan->monitor->read_stream ) {
#ifndef MONITOR_CONSTANT_DELAY
int jump = chan->outsmpl - chan->insmpl - 2 * f->samples;
@ -2007,17 +2232,10 @@ int ast_write(struct ast_channel *chan, struct ast_frame *fr)
break;
default:
if (chan->tech->write) {
if (chan->writetrans)
f = ast_translate(chan->writetrans, fr, 0);
else
f = fr;
f = (chan->writetrans) ? ast_translate(chan->writetrans, fr, 0) : fr;
if (f) {
if (f->frametype == AST_FRAME_VOICE && chan->spiers) {
struct ast_channel_spy *spying;
for (spying = chan->spiers; spying; spying=spying->next) {
ast_queue_spy_frame(spying, f, 1);
}
}
if (f->frametype == AST_FRAME_VOICE && chan->spies)
queue_frame_to_spies(chan, f, SPY_WRITE);
if( chan->monitor && chan->monitor->write_stream &&
f && ( f->frametype == AST_FRAME_VOICE ) ) {
@ -3207,8 +3425,9 @@ enum ast_bridge_result ast_channel_bridge(struct ast_channel *c0, struct ast_cha
if (c0->tech->bridge &&
(config->timelimit == 0) &&
(c0->tech->bridge == c1->tech->bridge) &&
!nativefailed && !c0->monitor && !c1->monitor && !c0->spiers && !c1->spiers) {
/* Looks like they share a bridge method */
!nativefailed && !c0->monitor && !c1->monitor &&
!c0->spies && !c1->spies) {
/* Looks like they share a bridge method and nothing else is in the way */
if (option_verbose > 2)
ast_verbose(VERBOSE_PREFIX_3 "Attempting native bridge of %s and %s\n", c0->name, c1->name);
ast_set_flag(c0, AST_FLAG_NBRIDGE);
@ -3237,6 +3456,7 @@ enum ast_bridge_result ast_channel_bridge(struct ast_channel *c0, struct ast_cha
} else {
ast_clear_flag(c0, AST_FLAG_NBRIDGE);
ast_clear_flag(c1, AST_FLAG_NBRIDGE);
ast_verbose(VERBOSE_PREFIX_3 "Native bridge of %s and %s was unsuccessful\n", c0->name, c1->name);
}
if (res == AST_BRIDGE_RETRY)
continue;
@ -3570,3 +3790,134 @@ void ast_set_variables(struct ast_channel *chan, struct ast_variable *vars)
for (cur = vars; cur; cur = cur->next)
pbx_builtin_setvar_helper(chan, cur->name, cur->value);
}
static void copy_data_from_queue(struct ast_channel_spy_queue *queue, short *buf, unsigned int samples)
{
struct ast_frame *f;
int tocopy;
int bytestocopy;
while (samples) {
f = queue->head;
if (!f) {
ast_log(LOG_ERROR, "Ran out of frames before buffer filled!\n");
break;
}
tocopy = (f->samples > samples) ? samples : f->samples;
bytestocopy = ast_codec_get_len(queue->format, samples);
memcpy(buf, f->data, bytestocopy);
samples -= tocopy;
buf += tocopy;
f->samples -= tocopy;
f->data += bytestocopy;
f->datalen -= bytestocopy;
f->offset += bytestocopy;
queue->samples -= tocopy;
if (!f->samples) {
queue->head = f->next;
ast_frfree(f);
}
}
}
struct ast_frame *ast_channel_spy_read_frame(struct ast_channel_spy *spy, unsigned int samples)
{
struct ast_frame *result;
/* buffers are allocated to hold SLINEAR, which is the largest format */
short read_buf[samples];
short write_buf[samples];
struct ast_frame *read_frame;
struct ast_frame *write_frame;
int need_dup;
struct ast_frame stack_read_frame = { .frametype = AST_FRAME_VOICE,
.subclass = spy->read_queue.format,
.data = read_buf,
.samples = samples,
.datalen = ast_codec_get_len(spy->read_queue.format, samples),
};
struct ast_frame stack_write_frame = { .frametype = AST_FRAME_VOICE,
.subclass = spy->write_queue.format,
.data = write_buf,
.samples = samples,
.datalen = ast_codec_get_len(spy->write_queue.format, samples),
};
/* if a flush has been requested, dump everything in whichever queue is larger */
if (ast_test_flag(spy, CHANSPY_TRIGGER_FLUSH)) {
if (spy->read_queue.samples > spy->write_queue.samples) {
if (ast_test_flag(spy, CHANSPY_READ_VOLADJUST)) {
for (result = spy->read_queue.head; result; result = result->next)
ast_frame_adjust_volume(result, spy->read_vol_adjustment);
}
result = spy->read_queue.head;
spy->read_queue.head = NULL;
spy->read_queue.samples = 0;
ast_clear_flag(spy, CHANSPY_TRIGGER_FLUSH);
return result;
} else {
if (ast_test_flag(spy, CHANSPY_WRITE_VOLADJUST)) {
for (result = spy->write_queue.head; result; result = result->next)
ast_frame_adjust_volume(result, spy->write_vol_adjustment);
}
result = spy->write_queue.head;
spy->write_queue.head = NULL;
spy->write_queue.samples = 0;
ast_clear_flag(spy, CHANSPY_TRIGGER_FLUSH);
return result;
}
}
if ((spy->read_queue.samples < samples) || (spy->write_queue.samples < samples))
return NULL;
/* short-circuit if both head frames have exactly what we want */
if ((spy->read_queue.head->samples == samples) &&
(spy->write_queue.head->samples == samples)) {
read_frame = spy->read_queue.head;
spy->read_queue.head = read_frame->next;
read_frame->next = NULL;
write_frame = spy->write_queue.head;
spy->write_queue.head = write_frame->next;
write_frame->next = NULL;
spy->read_queue.samples -= samples;
spy->write_queue.samples -= samples;
need_dup = 0;
} else {
copy_data_from_queue(&spy->read_queue, read_buf, samples);
copy_data_from_queue(&spy->write_queue, write_buf, samples);
read_frame = &stack_read_frame;
write_frame = &stack_write_frame;
need_dup = 1;
}
if (ast_test_flag(spy, CHANSPY_READ_VOLADJUST))
ast_frame_adjust_volume(read_frame, spy->read_vol_adjustment);
if (ast_test_flag(spy, CHANSPY_WRITE_VOLADJUST))
ast_frame_adjust_volume(write_frame, spy->write_vol_adjustment);
if (ast_test_flag(spy, CHANSPY_MIXAUDIO)) {
ast_frame_slinear_sum(read_frame, write_frame);
if (need_dup)
result = ast_frdup(read_frame);
else
result = read_frame;
} else {
if (need_dup) {
result = ast_frdup(read_frame);
result->next = ast_frdup(write_frame);
} else {
result = read_frame;
result->next = write_frame;
}
}
return result;
}

115
include/asterisk/channel.h
View File

@ -23,12 +23,6 @@
#ifndef _ASTERISK_CHANNEL_H
#define _ASTERISK_CHANNEL_H
#include "asterisk/compat.h"
#include "asterisk/frame.h"
#include "asterisk/sched.h"
#include "asterisk/chanvars.h"
#include "asterisk/config.h"
#include <unistd.h>
#include <setjmp.h>
#ifdef POLLCOMPAT
@ -41,18 +35,23 @@
extern "C" {
#endif
#include "asterisk/lock.h"
/*! Max length of an extension */
#define AST_MAX_EXTENSION 80
#define AST_MAX_CONTEXT 80
#define AST_CHANNEL_NAME 80
#include "asterisk/compat.h"
#include "asterisk/frame.h"
#include "asterisk/sched.h"
#include "asterisk/chanvars.h"
#include "asterisk/config.h"
#include "asterisk/lock.h"
#include "asterisk/cdr.h"
#include "asterisk/monitor.h"
#include "asterisk/utils.h"
#define AST_CHANNEL_NAME 80
#include "asterisk/linkedlists.h"
#define MAX_LANGUAGE 20
@ -170,17 +169,48 @@ struct ast_channel_tech {
};
#define CHANSPY_NEW 0
#define CHANSPY_RUNNING 1
#define CHANSPY_DONE 2
struct ast_channel_spy {
struct ast_frame *queue[2];
ast_mutex_t lock;
char status;
struct ast_channel_spy *next;
enum chanspy_states {
CHANSPY_NEW = 0,
CHANSPY_RUNNING = 1,
CHANSPY_DONE = 2,
};
enum chanspy_flags {
CHANSPY_MIXAUDIO = (1 << 0),
CHANSPY_READ_VOLADJUST = (1 << 1),
CHANSPY_WRITE_VOLADJUST = (1 << 2),
CHANSPY_FORMAT_AUDIO = (1 << 3),
CHANSPY_TRIGGER_MODE = (3 << 4),
CHANSPY_TRIGGER_READ = (1 << 4),
CHANSPY_TRIGGER_WRITE = (2 << 4),
CHANSPY_TRIGGER_NONE = (3 << 4),
CHANSPY_TRIGGER_FLUSH = (1 << 6),
};
struct ast_channel_spy_queue {
struct ast_frame *head;
unsigned int samples;
unsigned int format;
};
struct ast_channel_spy {
ast_mutex_t lock;
ast_cond_t trigger;
struct ast_channel_spy_queue read_queue;
struct ast_channel_spy_queue write_queue;
unsigned int flags;
enum chanspy_states status;
const char *type;
/* The volume adjustment values are very straightforward:
positive values cause the samples to be multiplied by that amount
negative values cause the samples to be divided by the absolute value of that amount
*/
int read_vol_adjustment;
int write_vol_adjustment;
AST_LIST_ENTRY(ast_channel_spy) list;
};
struct ast_channel_spy_list;
/*! Main Channel structure associated with a channel. */
/*!
@ -345,11 +375,10 @@ struct ast_channel {
int rawwriteformat;
/*! Chan Spy stuff */
struct ast_channel_spy *spiers;
struct ast_channel_spy_list *spies;
/*! For easy linking */
struct ast_channel *next;
};
/* Channel tech properties: */
@ -1008,6 +1037,50 @@ void ast_channel_inherit_variables(const struct ast_channel *parent, struct ast_
*/
void ast_set_variables(struct ast_channel *chan, struct ast_variable *vars);
/*!
\brief Adds a spy to a channel, to begin receiving copies of the channel's audio frames.
\param chan The channel to add the spy to.
\param spy A pointer to ast_channel_spy structure describing how the spy is to be used.
\return 0 for success, non-zero for failure
*/
int ast_channel_spy_add(struct ast_channel *chan, struct ast_channel_spy *spy);
/*!
\brief Remove a spy from a channel.
\param chan The channel to remove the spy from
\param spy The spy to be removed
\return nothing
*/
void ast_channel_spy_remove(struct ast_channel *chan, struct ast_channel_spy *spy);
/*!
\brief Find all spies of a particular type on a channel and stop them.
\param chan The channel to operate on
\param type A character string identifying the type of spies to be stopped
\return nothing
*/
void ast_channel_spy_stop_by_type(struct ast_channel *chan, const char *type);
/*!
\brief Read one (or more) frames of audio from a channel being spied upon.
\param spy The spy to operate on
\param samples The number of audio samples to read
\return NULL for failure, one ast_frame pointer, or a chain of ast_frame pointers
This function can return multiple frames if the spy structure needs to be 'flushed'
due to mismatched queue lengths, or if the spy structure is configured to return
unmixed audio (in which case each call to this function will return a frame of audio
from each side of channel).
*/
struct ast_frame *ast_channel_spy_read_frame(struct ast_channel_spy *spy, unsigned int samples);
/*!
\brief Efficiently wait until audio is available for a spy, or an exception occurs.
\param spy The spy to wait on
\return nothing
*/
void ast_channel_spy_trigger_wait(struct ast_channel_spy *spy);
/* Misc. functions below */
/* Helper function for migrating select to poll */