From 846b39a9f1e2ed8c636f16a2809a5c1c6c6ce820 Mon Sep 17 00:00:00 2001 From: "Kevin P. Fleming" Date: Fri, 28 Oct 2005 23:01:13 +0000 Subject: 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 --- .cleancount | 2 +- apps/app_chanspy.c | 291 +++++----------------- apps/app_mixmonitor.c | 605 +++++++++++++++++++-------------------------- apps/app_muxmon.c | 605 +++++++++++++++++++-------------------------- channel.c | 475 ++++++++++++++++++++++++++++++----- include/asterisk/channel.h | 109 ++++++-- 6 files changed, 1085 insertions(+), 1002 deletions(-) diff --git a/.cleancount b/.cleancount index 7f8f011eb..45a4fb75d 100755 --- a/.cleancount +++ b/.cleancount @@ -1 +1 @@ -7 +8 diff --git a/apps/app_chanspy.c b/apps/app_chanspy.c index 036c5edef..d9138ed00 100755 --- a/apps/app_chanspy.c +++ b/apps/app_chanspy.c @@ -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([][|])\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) -{ - struct chanspy_translation_helper *csth = data; - - ast_slinfactory_destroy(&csth->slinfactory[0]); - ast_slinfactory_destroy(&csth->slinfactory[1]); - - return; -} - -static void *spy_alloc(struct ast_channel *chan, void *params) -{ - struct chanspy_translation_helper *csth = params; - ast_slinfactory_init(&csth->slinfactory[0]); - ast_slinfactory_init(&csth->slinfactory[1]); - return params; -} - -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) +static void *spy_alloc(struct ast_channel *chan, void *data) { - 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); + /* just store the data pointer in the channel structure */ + return data; } - -#if 0 -static int extract_audio(short *buf, size_t len, struct ast_trans_pvt *trans, struct ast_frame *fr, int *maxsamp) +static void spy_release(struct ast_channel *chan, void *data) { - 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; + /* nothing to do */ } -#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; + if (!chan) + 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_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; } diff --git a/apps/app_mixmonitor.c b/apps/app_mixmonitor.c index ab8045835..c166c71aa 100755 --- a/apps/app_mixmonitor.c +++ b/apps/app_mixmonitor.c @@ -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 + * Kevin P. Fleming * + * Based on app_muxmon.c provided by * Anthony Minessale II * + * 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 -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include #include #include #include + +#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 -#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(.[|[|]])\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" + +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(.[|[|]])\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() - Adjust the heard volume by a factor of (range -4 to 4)\n" " V() - Adjust the spoken volume by a factor of (range -4 to 4)\n" " W() - Adjust the both heard and spoken volumes by a factor of \n" @@ -48,14 +66,16 @@ static char *desc = "" " 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 (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 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_unlock(&chan->lock); - } + if (!chan) + return; + + 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; - 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]); - + name = ast_strdupa(mixmonitor->chan->name); - - /* 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 = O_CREAT|O_WRONLY; + oflags |= ast_test_flag(mixmonitor, MUXFLAG_APPEND) ? O_APPEND : O_TRUNC; + + if ((ext = strchr(mixmonitor->filename, '.'))) { *(ext++) = '\0'; } else { ext = "raw"; } + 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 (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); - 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 (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 (ast_test_flag(muxmon, MUXFLAG_APPEND)) { - ast_seekstream(fs, 0, SEEK_END); - } + 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); - 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); + ast_channel_spy_trigger_wait(&spy); - usleep(1000); - sched_yield(); + 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); } } - } - if (muxmon->post_process) { + 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 (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); + if (args.options) { + char *opts[3] = { NULL, }; + + 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 (argv[3]) { - args = argv[3]; - } + if (argc < 3) + return RESULT_SHOWUSAGE; - 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 \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 "}; +static struct ast_cli_entry cli_mixmonitor = { + { "mixmonitor", NULL, NULL }, + mixmonitor_cli, + "Execute a MixMonitor command", + "mixmonitor []" +}; 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; } - diff --git a/apps/app_muxmon.c b/apps/app_muxmon.c index ab8045835..c166c71aa 100755 --- a/apps/app_muxmon.c +++ b/apps/app_muxmon.c @@ -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 + * Kevin P. Fleming * + * Based on app_muxmon.c provided by * Anthony Minessale II * + * 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 -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include #include #include #include + +#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 -#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(.[|[|]])\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" + +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(.[|[|]])\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() - Adjust the heard volume by a factor of (range -4 to 4)\n" " V() - Adjust the spoken volume by a factor of (range -4 to 4)\n" " W() - Adjust the both heard and spoken volumes by a factor of \n" @@ -48,14 +66,16 @@ static char *desc = "" " 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 (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 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_unlock(&chan->lock); - } + if (!chan) + return; + + 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; - 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]); - + name = ast_strdupa(mixmonitor->chan->name); - - /* 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 = O_CREAT|O_WRONLY; + oflags |= ast_test_flag(mixmonitor, MUXFLAG_APPEND) ? O_APPEND : O_TRUNC; + + if ((ext = strchr(mixmonitor->filename, '.'))) { *(ext++) = '\0'; } else { ext = "raw"; } + 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 (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); - 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 (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 (ast_test_flag(muxmon, MUXFLAG_APPEND)) { - ast_seekstream(fs, 0, SEEK_END); - } + 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); - 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); + ast_channel_spy_trigger_wait(&spy); - usleep(1000); - sched_yield(); + 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); } } - } - if (muxmon->post_process) { + 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 (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); + if (args.options) { + char *opts[3] = { NULL, }; + + 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 (argv[3]) { - args = argv[3]; - } + if (argc < 3) + return RESULT_SHOWUSAGE; - 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 \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 "}; +static struct ast_cli_entry cli_mixmonitor = { + { "mixmonitor", NULL, NULL }, + mixmonitor_cli, + "Execute a MixMonitor command", + "mixmonitor []" +}; 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; } - diff --git a/channel.c b/channel.c index 85692e7d6..c3f3d0384 100755 --- a/channel.c +++ b/channel.c @@ -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; + 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; + } - /* 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_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); } - chan->spiers = NULL; - return; + 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; - - 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); + trans = (dir == SPY_READ) ? &chan->spies->read_translator : &chan->spies->write_translator; + + 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); } - ast_mutex_unlock(&spy->lock); - return; - } - if (tmpf) { - tmpf->next = ast_frdup(f); - } else { - spy->queue[pos] = 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); } - 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; +} diff --git a/include/asterisk/channel.h b/include/asterisk/channel.h index 03e8da4fa..8c67b1631 100755 --- a/include/asterisk/channel.h +++ b/include/asterisk/channel.h @@ -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 #include #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 +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 { - struct ast_frame *queue[2]; ast_mutex_t lock; - char status; - struct ast_channel_spy *next; + 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 */ -- cgit v1.2.3