pbx_spool.c File Reference

Full-featured outgoing call spool support. More...

#include "asterisk.h"
#include <sys/stat.h>
#include <time.h>
#include <utime.h>
#include <dirent.h>
#include <sys/inotify.h>
#include "asterisk/paths.h"
#include "asterisk/lock.h"
#include "asterisk/file.h"
#include "asterisk/logger.h"
#include "asterisk/channel.h"
#include "asterisk/callerid.h"
#include "asterisk/pbx.h"
#include "asterisk/module.h"
#include "asterisk/utils.h"
#include "asterisk/options.h"
#include "asterisk/format.h"
#include "asterisk/format_cache.h"

Include dependency graph for pbx_spool.c:

Go to the source code of this file.

Data Structures
struct	createlist
struct	direntry
struct	dirlist
struct	openlist
struct	outgoing

Macros
#define	LINE_BUFFER_SIZE   1024

Enumerations
enum	{ SPOOL_FLAG_ALWAYS_DELETE = (1 << 0), SPOOL_FLAG_ARCHIVE = (1 << 1), SPOOL_FLAG_EARLY_MEDIA = (1 << 2) }

Functions
static void	__reg_module (void)
static void	__unreg_module (void)
static void	append_variable (struct outgoing *o, const char *name, const char *value)
static int	apply_outgoing (struct outgoing *o, FILE *f)
struct ast_module *	AST_MODULE_SELF_SYM (void)
static void *	attempt_thread (void *data)
static void	free_outgoing (struct outgoing *o)
static void	launch_service (struct outgoing *o)
static int	load_module (void)
static struct outgoing *	new_outgoing (const char *fn)
static void	parse_line (char *line, unsigned int lineno, struct outgoing *o)
static void	queue_created_files (void)
static void	queue_file (const char *filename, time_t when)
static void	queue_file_create (const char *filename)
static void	queue_file_open (const char *filename)
static void	queue_file_write (const char *filename)
static int	remove_from_queue (struct outgoing *o, const char *status)
	Remove a call file from the outgoing queue optionally moving it in the archive dir. More...
static void	safe_append (struct outgoing *o, time_t now, char *s)
static int	scan_service (const char *fn, time_t now)
static void *	scan_thread (void *unused)
static int	unload_module (void)

Variables
static struct ast_module_info	__mod_info = { .name = AST_MODULE, .flags = AST_MODFLAG_LOAD_ORDER , .description = "Outgoing Spool Support" , .key = "This paragraph is copyright (c) 2006 by Digium, Inc. \In order for your module to load, it must return this \key via a function called \"key\". Any code which \includes this paragraph must be licensed under the GNU \General Public License version 2 or later (at your \option). In addition to Digium's general reservations \of rights, Digium expressly reserves the right to \allow other parties to license this paragraph under \different terms. Any use of Digium, Inc. trademarks or \logos (including \"Asterisk\" or \"Digium\") without \express written permission of Digium, Inc. is prohibited.\n" , .buildopt_sum = "30ef0c93b36035ec78c9cfd712d36d9b" , .load = load_module, .unload = unload_module, .load_pri = AST_MODPRI_DEFAULT, .support_level = AST_MODULE_SUPPORT_CORE, }
static const struct ast_module_info *	ast_module_info = &__mod_info
static struct createlist	createlist = { .first = NULL, .last = NULL, }
static struct dirlist	dirlist = { .first = NULL, .last = NULL, .lock = { PTHREAD_RECURSIVE_MUTEX_INITIALIZER_NP , NULL, {1, 0} } , }
static struct openlist	openlist = { .first = NULL, .last = NULL, }
static char	qdir [255]
static char	qdonedir [255]

Detailed Description

Full-featured outgoing call spool support.

Definition in file pbx_spool.c.

Macro Definition Documentation

LINE_BUFFER_SIZE

#define LINE_BUFFER_SIZE   1024

Definition at line 294 of file pbx_spool.c.

Referenced by apply_outgoing().

Enumeration Type Documentation

anonymous enum

anonymous enum

Enumerator
SPOOL_FLAG_ALWAYS_DELETE	Always delete the call file after a call succeeds or the maximum number of retries is exceeded, even if the modification time of the call file is in the future.
SPOOL_FLAG_ARCHIVE
SPOOL_FLAG_EARLY_MEDIA

Definition at line 62 of file pbx_spool.c.

 {
 /*! Always delete the call file after a call succeeds or the
 * maximum number of retries is exceeded, even if the
 * modification time of the call file is in the future.
 */
 SPOOL_FLAG_ALWAYS_DELETE = (1 << 0),
 /* Don't unlink the call file after processing, move in qdonedir */
 SPOOL_FLAG_ARCHIVE = (1 << 1),
 /* Connect the channel with the outgoing extension once early media is received */
 SPOOL_FLAG_EARLY_MEDIA = (1 << 2),
};

Function Documentation

__reg_module()

static void __reg_module ( void  )

static

Definition at line 965 of file pbx_spool.c.

__unreg_module()

static void __unreg_module ( void  )

static

Definition at line 965 of file pbx_spool.c.

append_variable()

static void append_variable ( struct outgoing *  o, const char *  name, const char *  value  )

static

Definition at line 164 of file pbx_spool.c.

References ast_variable_list_append, ast_variable_new, outgoing::fn, var, and outgoing::vars.

Referenced by apply_outgoing(), and parse_line().

{
 struct ast_variable *var = ast_variable_new(name, value, o->fn);

 if (!var) {
 return;
 }

 /* Always insert at the end, because some people want to treat the spool
 * file as a script */
 ast_variable_list_append(&o->vars, var);
}

apply_outgoing()

static int apply_outgoing ( struct outgoing *  o, FILE *  f  )

static

Definition at line 296 of file pbx_spool.c.

References outgoing::app, append_variable(), ast_log, ast_strlen_zero, buf, outgoing::dest, outgoing::exten, outgoing::fn, len(), LINE_BUFFER_SIZE, ast_variable::lineno, LOG_WARNING, parse_line(), outgoing::retries, and outgoing::tech.

Referenced by scan_service().

{
 char buf[LINE_BUFFER_SIZE];
 unsigned int lineno = 0;

 while (fgets(buf, sizeof(buf), f)) {
 size_t len = strlen(buf);

 lineno++;

 if (buf[len - 1] == '\n' || feof(f)) {
 /* We have a line, parse it */
 parse_line(buf, lineno, o);
 continue;
 }

 /* Crazy long line, skip it */
 ast_log(LOG_WARNING, "Skipping extremely long line at line %d of %s\n", lineno, o->fn);

 /* Consume the rest of the problematic line */
 while (fgets(buf, sizeof(buf), f)) {
 len = strlen(buf);
 if (buf[len - 1] == '\n' || feof(f)) {
 break;
 }
 }
 }

 if (ast_strlen_zero(o->tech)
 || ast_strlen_zero(o->dest)
 || (ast_strlen_zero(o->app) && ast_strlen_zero(o->exten))) {
 ast_log(LOG_WARNING, "At least one of app or extension must be specified, "
 "along with tech and dest in file %s\n", o->fn);
 return -1;
 }

 if (snprintf(buf, sizeof(buf), "%d", o->retries + 1) < sizeof(buf)) {
 append_variable(o, "AST_OUTGOING_ATTEMPT", buf);
 }

 return 0;
}

AST_MODULE_SELF_SYM()

struct ast_module* AST_MODULE_SELF_SYM

(

void

)

Definition at line 965 of file pbx_spool.c.

attempt_thread()

static void* attempt_thread ( void *  data )

static

Definition at line 431 of file pbx_spool.c.

References outgoing::account, outgoing::app, ast_channel_reason2str(), ast_log, AST_OUTGOING_WAIT_COMPLETE, ast_pbx_outgoing_app(), ast_pbx_outgoing_exten(), ast_strlen_zero, ast_test_flag, ast_verb, outgoing::capabilities, outgoing::cid_name, outgoing::cid_num, outgoing::context, outgoing::data, outgoing::dest, outgoing::exten, outgoing::fn, free_outgoing(), LOG_NOTICE, outgoing::maxretries, NULL, outgoing::options, outgoing::priority, queue_file(), remove_from_queue(), outgoing::retries, outgoing::retrytime, safe_append(), SPOOL_FLAG_EARLY_MEDIA, outgoing::tech, outgoing::vars, and outgoing::waittime.

Referenced by launch_service().

{
 struct outgoing *o = data;
 int res, reason;
 if (!ast_strlen_zero(o->app)) {
 ast_verb(3, "Attempting call on %s/%s for application %s(%s) (Retry %d)\n", o->tech, o->dest, o->app, o->data, o->retries);
 res = ast_pbx_outgoing_app(o->tech, o->capabilities, o->dest,
 o->waittime * 1000, o->app, o->data, &reason,
 AST_OUTGOING_WAIT_COMPLETE, o->cid_num, o->cid_name,
 o->vars, o->account, NULL, NULL);
 } else {
 ast_verb(3, "Attempting call on %s/%s for %s@%s:%d (Retry %d)\n", o->tech, o->dest, o->exten, o->context,o->priority, o->retries);
 res = ast_pbx_outgoing_exten(o->tech, o->capabilities, o->dest,
 o->waittime * 1000, o->context, o->exten, o->priority, &reason,
 AST_OUTGOING_WAIT_COMPLETE, o->cid_num, o->cid_name,
 o->vars, o->account, NULL, ast_test_flag(&o->options, SPOOL_FLAG_EARLY_MEDIA),
 NULL);
 }
 if (res) {
 ast_log(LOG_NOTICE, "Call failed to go through, reason (%d) %s\n", reason, ast_channel_reason2str(reason));
 if (o->retries >= o->maxretries + 1) {
 /* Max retries exceeded */
 ast_log(LOG_NOTICE, "Queued call to %s/%s expired without completion after %d attempt%s\n", o->tech, o->dest, o->retries - 1, ((o->retries - 1) != 1) ? "s" : "");
 remove_from_queue(o, "Expired");
 } else {
 /* Notate that the call is still active */
 safe_append(o, time(NULL), "EndRetry");
#if defined(HAVE_INOTIFY) || defined(HAVE_KQUEUE)
 queue_file(o->fn, time(NULL) + o->retrytime);
#endif
 }
 } else {
 ast_log(LOG_NOTICE, "Call completed to %s/%s\n", o->tech, o->dest);
 remove_from_queue(o, "Completed");
 }
 free_outgoing(o);
 return NULL;
}

free_outgoing()

static void free_outgoing ( struct outgoing *  o )

static

Definition at line 114 of file pbx_spool.c.

References ao2_cleanup, ast_free, ast_string_field_free_memory, ast_variables_destroy(), outgoing::capabilities, and outgoing::vars.

Referenced by attempt_thread(), launch_service(), new_outgoing(), and scan_service().

{
 if (o->vars) {
 ast_variables_destroy(o->vars);
 }
 ao2_cleanup(o->capabilities);
 ast_string_field_free_memory(o);
 ast_free(o);
}

launch_service()

static void launch_service ( struct outgoing *  o )

static

Definition at line 470 of file pbx_spool.c.

References ast_log, ast_pthread_create_detached, attempt_thread(), free_outgoing(), LOG_WARNING, and NULL.

Referenced by scan_service().

{
 pthread_t t;
 int ret;

 if ((ret = ast_pthread_create_detached(&t, NULL, attempt_thread, o))) {
 ast_log(LOG_WARNING, "Unable to create thread :( (returned error: %d)\n", ret);
 free_outgoing(o);
 }
}

load_module()

static int load_module ( void  )

static

Definition at line 946 of file pbx_spool.c.

References ast_config_AST_SPOOL_DIR, ast_log, ast_mkdir(), AST_MODULE_LOAD_DECLINE, AST_MODULE_LOAD_FAILURE, AST_MODULE_LOAD_SUCCESS, ast_pthread_create_detached_background, LOG_WARNING, NULL, scan_thread(), and thread.

{
 pthread_t thread;
 int ret;
 snprintf(qdir, sizeof(qdir), "%s/%s", ast_config_AST_SPOOL_DIR, "outgoing");
 if (ast_mkdir(qdir, 0777)) {
 ast_log(LOG_WARNING, "Unable to create queue directory %s -- outgoing spool disabled\n", qdir);
 return AST_MODULE_LOAD_DECLINE;
 }
 snprintf(qdonedir, sizeof(qdir), "%s/%s", ast_config_AST_SPOOL_DIR, "outgoing_done");

 if ((ret = ast_pthread_create_detached_background(&thread, NULL, scan_thread, NULL))) {
 ast_log(LOG_WARNING, "Unable to create thread :( (returned error: %d)\n", ret);
 return AST_MODULE_LOAD_FAILURE;
 }

 return AST_MODULE_LOAD_SUCCESS;
}

new_outgoing()

static struct outgoing* new_outgoing ( const char *  fn )

static

Definition at line 124 of file pbx_spool.c.

References ast_calloc, ast_format_cap_alloc, ast_format_cap_append, AST_FORMAT_CAP_FLAG_DEFAULT, ast_format_slin, ast_free, ast_set_flag, ast_string_field_init, ast_string_field_set, ast_strlen_zero, outgoing::capabilities, outgoing::fn, free_outgoing(), NULL, outgoing::options, outgoing::priority, outgoing::retrytime, SPOOL_FLAG_ALWAYS_DELETE, and outgoing::waittime.

Referenced by scan_service().

{
 struct outgoing *o;

 o = ast_calloc(1, sizeof(*o));
 if (!o) {
 return NULL;
 }

 /* Initialize the new object. */
 o->priority = 1;
 o->retrytime = 300;
 o->waittime = 45;
 ast_set_flag(&o->options, SPOOL_FLAG_ALWAYS_DELETE);
 if (ast_string_field_init(o, 128)) {
 /*
 * No need to call free_outgoing here since the failure was to
 * allocate string fields and no variables have been allocated
 * yet.
 */
 ast_free(o);
 return NULL;
 }
 ast_string_field_set(o, fn, fn);
 if (ast_strlen_zero(o->fn)) {
 /* String field set failed. Since this string is important we must fail. */
 free_outgoing(o);
 return NULL;
 }

 o->capabilities = ast_format_cap_alloc(AST_FORMAT_CAP_FLAG_DEFAULT);
 if (!o->capabilities) {
 free_outgoing(o);
 return NULL;
 }
 ast_format_cap_append(o->capabilities, ast_format_slin, 0);

 return o;
}

parse_line()

static void parse_line ( char *  line, unsigned int  lineno, struct outgoing *  o  )

static

Definition at line 177 of file pbx_spool.c.

References outgoing::account, outgoing::app, append_variable(), ast_callerid_split(), ast_format_cap_update_by_allow_disallow(), ast_log, ast_set2_flag, ast_skip_blanks(), ast_string_field_set, ast_strlen_zero, ast_trim_blanks(), ast_true(), c, outgoing::callingpid, outgoing::capabilities, outgoing::cid_name, outgoing::cid_num, outgoing::context, outgoing::data, outgoing::dest, outgoing::exten, outgoing::fn, LOG_NOTICE, LOG_WARNING, outgoing::maxretries, outgoing::options, outgoing::priority, outgoing::retries, outgoing::retrytime, SPOOL_FLAG_ALWAYS_DELETE, SPOOL_FLAG_ARCHIVE, SPOOL_FLAG_EARLY_MEDIA, strsep(), outgoing::tech, and outgoing::waittime.

Referenced by apply_outgoing().

{
 char *c;

 /* Trim comments */
 c = line;
 while ((c = strchr(c, '#'))) {
 if ((c == line) || (*(c-1) == ' ') || (*(c-1) == '\t')) {
 *c = '\0';
 break;
 }
 c++;
 }

 c = line;
 while ((c = strchr(c, ';'))) {
 if ((c > line) && (c[-1] == '\\')) {
 memmove(c - 1, c, strlen(c) + 1);
 } else {
 *c = '\0';
 break;
 }
 }

 /* Trim trailing white space */
 ast_trim_blanks(line);
 if (ast_strlen_zero(line)) {
 return;
 }
 c = strchr(line, ':');
 if (!c) {
 ast_log(LOG_NOTICE, "Syntax error at line %d of %s\n", lineno, o->fn);
 return;
 }
 *c = '\0';
 c = ast_skip_blanks(c + 1);
#if 0
 printf("'%s' is '%s' at line %d\n", line, c, lineno);
#endif
 if (!strcasecmp(line, "channel")) {
 char *c2;
 if ((c2 = strchr(c, '/'))) {
 *c2 = '\0';
 c2++;
 ast_string_field_set(o, tech, c);
 ast_string_field_set(o, dest, c2);
 } else {
 ast_log(LOG_NOTICE, "Channel should be in form Tech/Dest at line %d of %s\n", lineno, o->fn);
 }
 } else if (!strcasecmp(line, "callerid")) {
 char cid_name[80] = {0}, cid_num[80] = {0};
 ast_callerid_split(c, cid_name, sizeof(cid_name), cid_num, sizeof(cid_num));
 ast_string_field_set(o, cid_num, cid_num);
 ast_string_field_set(o, cid_name, cid_name);
 } else if (!strcasecmp(line, "application")) {
 ast_string_field_set(o, app, c);
 } else if (!strcasecmp(line, "data")) {
 ast_string_field_set(o, data, c);
 } else if (!strcasecmp(line, "maxretries")) {
 if (sscanf(c, "%30d", &o->maxretries) != 1) {
 ast_log(LOG_WARNING, "Invalid max retries at line %d of %s\n", lineno, o->fn);
 o->maxretries = 0;
 }
 } else if (!strcasecmp(line, "codecs")) {
 ast_format_cap_update_by_allow_disallow(o->capabilities, c, 1);
 } else if (!strcasecmp(line, "context")) {
 ast_string_field_set(o, context, c);
 } else if (!strcasecmp(line, "extension")) {
 ast_string_field_set(o, exten, c);
 } else if (!strcasecmp(line, "priority")) {
 if ((sscanf(c, "%30d", &o->priority) != 1) || (o->priority < 1)) {
 ast_log(LOG_WARNING, "Invalid priority at line %d of %s\n", lineno, o->fn);
 o->priority = 1;
 }
 } else if (!strcasecmp(line, "retrytime")) {
 if ((sscanf(c, "%30d", &o->retrytime) != 1) || (o->retrytime < 1)) {
 ast_log(LOG_WARNING, "Invalid retrytime at line %d of %s\n", lineno, o->fn);
 o->retrytime = 300;
 }
 } else if (!strcasecmp(line, "waittime")) {
 if ((sscanf(c, "%30d", &o->waittime) != 1) || (o->waittime < 1)) {
 ast_log(LOG_WARNING, "Invalid waittime at line %d of %s\n", lineno, o->fn);
 o->waittime = 45;
 }
 } else if (!strcasecmp(line, "retry")) {
 o->retries++;
 } else if (!strcasecmp(line, "startretry")) {
 if (sscanf(c, "%30ld", &o->callingpid) != 1) {
 ast_log(LOG_WARNING, "Unable to retrieve calling PID!\n");
 o->callingpid = 0;
 }
 } else if (!strcasecmp(line, "endretry") || !strcasecmp(line, "abortretry")) {
 o->callingpid = 0;
 o->retries++;
 } else if (!strcasecmp(line, "delayedretry")) {
 } else if (!strcasecmp(line, "setvar") || !strcasecmp(line, "set")) {
 char *c2 = c;

 strsep(&c2, "=");
 if (c2) {
 append_variable(o, c, c2);
 } else {
 ast_log(LOG_WARNING, "Malformed \"%s\" argument. Should be \"%s: variable=value\"\n", line, line);
 }
 } else if (!strcasecmp(line, "account")) {
 ast_string_field_set(o, account, c);
 } else if (!strcasecmp(line, "alwaysdelete")) {
 ast_set2_flag(&o->options, ast_true(c), SPOOL_FLAG_ALWAYS_DELETE);
 } else if (!strcasecmp(line, "archive")) {
 ast_set2_flag(&o->options, ast_true(c), SPOOL_FLAG_ARCHIVE);
 } else if (!strcasecmp(line, "early_media")) {
 ast_set2_flag(&o->options, ast_true(c), SPOOL_FLAG_EARLY_MEDIA);
 } else {
 ast_log(LOG_WARNING, "Unknown keyword '%s' at line %d of %s\n", line, lineno, o->fn);
 }
}

queue_created_files()

static void queue_created_files ( void  )

static

Definition at line 657 of file pbx_spool.c.

References ast_free, AST_LIST_REMOVE_CURRENT, AST_LIST_TRAVERSE_SAFE_BEGIN, AST_LIST_TRAVERSE_SAFE_END, direntry::list, direntry::mtime, direntry::name, NULL, and queue_file().

Referenced by scan_thread().

{
 struct direntry *cur;
 time_t now = time(NULL);

 AST_LIST_TRAVERSE_SAFE_BEGIN(&createlist, cur, list) {
 if (cur->mtime > now) {
 break;
 }

 AST_LIST_REMOVE_CURRENT(list);
 queue_file(cur->name, 0);
 ast_free(cur);
 }
 AST_LIST_TRAVERSE_SAFE_END
}

queue_file()

static void queue_file ( const char *  filename, time_t  when  )

static

Definition at line 560 of file pbx_spool.c.

References ast_alloca, ast_calloc, AST_LIST_EMPTY, AST_LIST_INSERT_BEFORE_CURRENT, AST_LIST_INSERT_HEAD, AST_LIST_INSERT_TAIL, AST_LIST_LOCK, AST_LIST_TRAVERSE, AST_LIST_TRAVERSE_SAFE_BEGIN, AST_LIST_TRAVERSE_SAFE_END, AST_LIST_UNLOCK, ast_log, errno, outgoing::fn, direntry::list, LOG_WARNING, direntry::mtime, direntry::name, NULL, and scan_service().

Referenced by attempt_thread(), queue_created_files(), queue_file_write(), and scan_thread().

{
 struct stat st;
 struct direntry *cur, *new;
 int res;
 time_t now = time(NULL);

 if (!strchr(filename, '/')) {
 char *fn = ast_alloca(strlen(qdir) + strlen(filename) + 2);
 sprintf(fn, "%s/%s", qdir, filename); /* SAFE */
 filename = fn;
 }

 if (when == 0) {
 if (stat(filename, &st)) {
 ast_log(LOG_WARNING, "Unable to stat %s: %s\n", filename, strerror(errno));
 return;
 }

 if (!S_ISREG(st.st_mode)) {
 return;
 }

 when = st.st_mtime;
 }

 /* Need to check the existing list in order to avoid duplicates. */
 AST_LIST_LOCK(&dirlist);
 AST_LIST_TRAVERSE(&dirlist, cur, list) {
 if (cur->mtime == when && !strcmp(filename, cur->name)) {
 AST_LIST_UNLOCK(&dirlist);
 return;
 }
 }

 if ((res = when) > now || (res = scan_service(filename, now)) > 0) {
 if (!(new = ast_calloc(1, sizeof(*new) + strlen(filename) + 1))) {
 AST_LIST_UNLOCK(&dirlist);
 return;
 }
 new->mtime = res;
 strcpy(new->name, filename);
 /* List is ordered by mtime */
 if (AST_LIST_EMPTY(&dirlist)) {
 AST_LIST_INSERT_HEAD(&dirlist, new, list);
 } else {
 int found = 0;
 AST_LIST_TRAVERSE_SAFE_BEGIN(&dirlist, cur, list) {
 if (cur->mtime > new->mtime) {
 AST_LIST_INSERT_BEFORE_CURRENT(new, list);
 found = 1;
 break;
 }
 }
 AST_LIST_TRAVERSE_SAFE_END
 if (!found) {
 AST_LIST_INSERT_TAIL(&dirlist, new, list);
 }
 }
 }
 AST_LIST_UNLOCK(&dirlist);
}

queue_file_create()

static void queue_file_create ( const char *  filename )

static

Definition at line 624 of file pbx_spool.c.

References ast_calloc, AST_LIST_INSERT_TAIL, AST_LIST_TRAVERSE, direntry::list, direntry::mtime, direntry::name, and NULL.

Referenced by scan_thread().

{
 struct direntry *cur;

 AST_LIST_TRAVERSE(&createlist, cur, list) {
 if (!strcmp(cur->name, filename)) {
 return;
 }
 }

 if (!(cur = ast_calloc(1, sizeof(*cur) + strlen(filename) + 1))) {
 return;
 }
 strcpy(cur->name, filename);
 /* We'll handle this file unless an IN_OPEN event occurs within 2 seconds */
 cur->mtime = time(NULL) + 2;
 AST_LIST_INSERT_TAIL(&createlist, cur, list);
}

queue_file_open()

static void queue_file_open ( const char *  filename )

static

Definition at line 643 of file pbx_spool.c.

References AST_LIST_INSERT_TAIL, AST_LIST_REMOVE_CURRENT, AST_LIST_TRAVERSE_SAFE_BEGIN, AST_LIST_TRAVERSE_SAFE_END, direntry::list, and direntry::name.

Referenced by scan_thread().

{
 struct direntry *cur;

 AST_LIST_TRAVERSE_SAFE_BEGIN(&createlist, cur, list) {
 if (!strcmp(cur->name, filename)) {
 AST_LIST_REMOVE_CURRENT(list);
 AST_LIST_INSERT_TAIL(&openlist, cur, list);
 break;
 }
 }
 AST_LIST_TRAVERSE_SAFE_END
}

queue_file_write()

static void queue_file_write ( const char *  filename )

static

Definition at line 674 of file pbx_spool.c.

References ast_free, AST_LIST_REMOVE_CURRENT, AST_LIST_TRAVERSE_SAFE_BEGIN, AST_LIST_TRAVERSE_SAFE_END, direntry::list, direntry::name, and queue_file().

Referenced by scan_thread().

{
 struct direntry *cur;
 /* Only queue entries where an IN_CREATE preceded the IN_CLOSE_WRITE */
 AST_LIST_TRAVERSE_SAFE_BEGIN(&openlist, cur, list) {
 if (!strcmp(cur->name, filename)) {
 AST_LIST_REMOVE_CURRENT(list);
 ast_free(cur);
 queue_file(filename, 0);
 break;
 }
 }
 AST_LIST_TRAVERSE_SAFE_END
}

remove_from_queue()

static int remove_from_queue ( struct outgoing *  o, const char *  status  )

static

Remove a call file from the outgoing queue optionally moving it in the archive dir.

Parameters

o	the pointer to outgoing struct
status	the exit status of the call. Can be "Completed", "Failed" or "Expired"

Definition at line 363 of file pbx_spool.c.

References ast_free, AST_LIST_LOCK, AST_LIST_REMOVE_CURRENT, AST_LIST_TRAVERSE_SAFE_BEGIN, AST_LIST_TRAVERSE_SAFE_END, AST_LIST_UNLOCK, ast_log, ast_mkdir(), ast_test_flag, outgoing::fn, LOG_WARNING, direntry::name, NULL, outgoing::options, SPOOL_FLAG_ALWAYS_DELETE, and SPOOL_FLAG_ARCHIVE.

Referenced by attempt_thread(), and scan_service().

{
 FILE *f;
 char newfn[256];
 const char *bname;

#if defined(HAVE_INOTIFY) || defined(HAVE_KQUEUE)
 struct direntry *cur;
#endif

 if (!ast_test_flag(&o->options, SPOOL_FLAG_ALWAYS_DELETE)) {
 struct stat current_file_status;

 if (!stat(o->fn, &current_file_status)) {
 if (time(NULL) < current_file_status.st_mtime) {
 return 0;
 }
 }
 }

#if defined(HAVE_INOTIFY) || defined(HAVE_KQUEUE)
 AST_LIST_LOCK(&dirlist);
 AST_LIST_TRAVERSE_SAFE_BEGIN(&dirlist, cur, list) {
 if (!strcmp(cur->name, o->fn)) {
 AST_LIST_REMOVE_CURRENT(list);
 ast_free(cur);
 break;
 }
 }
 AST_LIST_TRAVERSE_SAFE_END;
 AST_LIST_UNLOCK(&dirlist);
#endif

 if (!ast_test_flag(&o->options, SPOOL_FLAG_ARCHIVE)) {
 unlink(o->fn);
 return 0;
 }

 if (ast_mkdir(qdonedir, 0777)) {
 ast_log(LOG_WARNING, "Unable to create queue directory %s -- outgoing spool archiving disabled\n", qdonedir);
 unlink(o->fn);
 return -1;
 }

 if (!(bname = strrchr(o->fn, '/'))) {
 bname = o->fn;
 } else {
 bname++;
 }

 snprintf(newfn, sizeof(newfn), "%s/%s", qdonedir, bname);
 /* If there is already a call file with the name in the archive dir, it will be overwritten. */
 unlink(newfn);
 if (rename(o->fn, newfn) != 0) {
 unlink(o->fn);
 return -1;
 }

 /* Only append to the file AFTER we move it out of the watched directory,
 * otherwise the fclose() causes another event for inotify(7) */
 if ((f = fopen(newfn, "a"))) {
 fprintf(f, "Status: %s\n", status);
 fclose(f);
 }

 return 0;
}

safe_append()

static void safe_append ( struct outgoing *  o, time_t  now, char *  s  )

static

Definition at line 339 of file pbx_spool.c.

References ast_debug, ast_log, ast_mainpid, outgoing::dest, errno, outgoing::fn, LOG_WARNING, outgoing::retries, outgoing::retrytime, and outgoing::tech.

Referenced by attempt_thread(), and scan_service().

{
 FILE *f;
 struct utimbuf tbuf = { .actime = now, .modtime = now + o->retrytime };

 ast_debug(1, "Outgoing %s/%s: %s\n", o->tech, o->dest, s);

 if ((f = fopen(o->fn, "a"))) {
 fprintf(f, "\n%s: %ld %d (%ld)\n", s, (long)ast_mainpid, o->retries, (long) now);
 fclose(f);
 }

 /* Update the file time */
 if (utime(o->fn, &tbuf)) {
 ast_log(LOG_WARNING, "Unable to set utime on %s: %s\n", o->fn, strerror(errno));
 }
}

scan_service()

static int scan_service ( const char *  fn, time_t  now  )

static

Todo:

XXX There is some odd delayed duplicate servicing of call files going on. We need to suppress the error message if the file does not exist as a result.

Definition at line 482 of file pbx_spool.c.

References apply_outgoing(), ast_debug, ast_log, ast_mainpid, outgoing::callingpid, outgoing::dest, errno, outgoing::fn, free_outgoing(), launch_service(), LOG_NOTICE, LOG_WARNING, outgoing::maxretries, new_outgoing(), NULL, remove_from_queue(), outgoing::retries, outgoing::retrytime, safe_append(), and outgoing::tech.

Referenced by queue_file(), and scan_thread().

{
 struct outgoing *o;
 FILE *f;
 int res;

 o = new_outgoing(fn);
 if (!o) {
 return -1;
 }

 /* Attempt to open the file */
 f = fopen(o->fn, "r");
 if (!f) {
#if defined(HAVE_INOTIFY) || defined(HAVE_KQUEUE)
 /*!
 * \todo XXX There is some odd delayed duplicate servicing of
 * call files going on. We need to suppress the error message
 * if the file does not exist as a result.
 */
 if (errno != ENOENT)
#endif
 {
 ast_log(LOG_WARNING, "Unable to open %s: '%s'(%d), deleting\n",
 o->fn, strerror(errno), (int) errno);
 }
 remove_from_queue(o, "Failed");
 free_outgoing(o);
 return -1;
 }

 /* Read in and verify the contents */
 res = apply_outgoing(o, f);
 fclose(f);
 if (res) {
 ast_log(LOG_WARNING, "Invalid file contents in %s, deleting\n", o->fn);
 remove_from_queue(o, "Failed");
 free_outgoing(o);
 return -1;
 }

 ast_debug(1, "Filename: %s, Retries: %d, max: %d\n", o->fn, o->retries, o->maxretries);
 if (o->retries <= o->maxretries) {
 now += o->retrytime;
 if (o->callingpid && (o->callingpid == ast_mainpid)) {
 safe_append(o, time(NULL), "DelayedRetry");
 ast_debug(1, "Delaying retry since we're currently running '%s'\n", o->fn);
 free_outgoing(o);
 } else {
 /* Increment retries */
 o->retries++;
 /* If someone else was calling, they're presumably gone now
 so abort their retry and continue as we were... */
 if (o->callingpid)
 safe_append(o, time(NULL), "AbortRetry");

 safe_append(o, now, "StartRetry");
 launch_service(o);
 }
 return now;
 }

 ast_log(LOG_NOTICE, "Queued call to %s/%s expired without completion after %d attempt%s\n",
 o->tech, o->dest, o->retries - 1, ((o->retries - 1) != 1) ? "s" : "");
 remove_from_queue(o, "Expired");
 free_outgoing(o);
 return 0;
}

scan_thread()

static void* scan_thread ( void *  unused )

static

Definition at line 690 of file pbx_spool.c.

References ast_debug, ast_free, ast_fully_booted, AST_LIST_EMPTY, AST_LIST_FIRST, AST_LIST_LOCK, AST_LIST_REMOVE_HEAD, AST_LIST_UNLOCK, ast_log, buf, errno, outgoing::fn, HAVE_INOTIFY, inotify_fd, last, len(), LOG_ERROR, LOG_WARNING, direntry::mtime, direntry::name, direntry::next, NULL, queue_created_files(), queue_file(), queue_file_create(), queue_file_open(), queue_file_write(), and scan_service().

Referenced by load_module().

{
 DIR *dir;
 struct dirent *de;
 time_t now;
 struct timespec ts = { .tv_sec = 1 };
#ifdef HAVE_INOTIFY
 ssize_t res;
 int inotify_fd = inotify_init();
 struct inotify_event *iev;
 char buf[8192] __attribute__((aligned (sizeof(int))));
 struct pollfd pfd = { .fd = inotify_fd, .events = POLLIN };
#else
 struct timespec nowait = { .tv_sec = 0, .tv_nsec = 1 };
 int inotify_fd = kqueue();
 struct kevent kev;
 struct kevent event;
#endif
 struct direntry *cur;

 while (!ast_fully_booted) {
 nanosleep(&ts, NULL);
 }

 if (inotify_fd < 0) {
 ast_log(LOG_ERROR, "Unable to initialize "
#ifdef HAVE_INOTIFY
 "inotify(7)"
#else
 "kqueue(2)"
#endif
 "\n");
 return NULL;
 }

#ifdef HAVE_INOTIFY
 inotify_add_watch(inotify_fd, qdir, IN_CREATE | IN_OPEN | IN_CLOSE_WRITE | IN_MOVED_TO);
#endif

 /* First, run through the directory and clear existing entries */
 if (!(dir = opendir(qdir))) {
 ast_log(LOG_ERROR, "Unable to open directory %s: %s\n", qdir, strerror(errno));
 return NULL;
 }

#ifndef HAVE_INOTIFY
 EV_SET(&kev, dirfd(dir), EVFILT_VNODE, EV_ADD | EV_ENABLE | EV_CLEAR, NOTE_WRITE, 0, NULL);
 if (kevent(inotify_fd, &kev, 1, &event, 1, &nowait) < 0 && errno != 0) {
 ast_log(LOG_ERROR, "Unable to watch directory %s: %s\n", qdir, strerror(errno));
 }
#endif
 now = time(NULL);
 while ((de = readdir(dir))) {
 queue_file(de->d_name, 0);
 }

#ifdef HAVE_INOTIFY
 /* Directory needs to remain open for kqueue(2) */
 closedir(dir);
#endif

 /* Wait for either a) next timestamp to occur, or b) a change to happen */
 for (;/* ever */;) {
 time_t next = AST_LIST_EMPTY(&dirlist) ? INT_MAX : AST_LIST_FIRST(&dirlist)->mtime;

 time(&now);
 if (next > now) {
#ifdef HAVE_INOTIFY
 int stage = 0;
 /* Convert from seconds to milliseconds, unless there's nothing
 * in the queue already, in which case, we wait forever. */
 int waittime = next == INT_MAX ? -1 : (next - now) * 1000;
 if (!AST_LIST_EMPTY(&createlist)) {
 waittime = 1000;
 }
 /* When a file arrives, add it to the queue, in mtime order. */
 if ((res = poll(&pfd, 1, waittime)) > 0 && (stage = 1) &&
 (res = read(inotify_fd, &buf, sizeof(buf))) >= sizeof(*iev)) {
 ssize_t len = 0;
 /* File(s) added to directory, add them to my list */
 for (iev = (void *) buf; res >= sizeof(*iev); iev = (struct inotify_event *) (((char *) iev) + len)) {
 /* For an IN_MOVED_TO event, simply process the file. However, if
 * we get an IN_CREATE event it *might* be an open(O_CREAT) or it
 * might be a hardlink (like smsq does, since rename() might
 * overwrite an existing file). So we have to see if we get a
 * subsequent IN_OPEN event on the same file. If we do, keep it
 * on the openlist and wait for the corresponding IN_CLOSE_WRITE.
 * If we *don't* see an IN_OPEN event, then it was a hard link so
 * it can be processed immediately.
 *
 * Unfortunately, although open(O_CREAT) is an atomic file system
 * operation, the inotify subsystem doesn't give it to us in a
 * single event with both IN_CREATE|IN_OPEN set. It's two separate
 * events, and the kernel doesn't even give them to us at the same
 * time. We can read() from inotify_fd after the IN_CREATE event,
 * and get *nothing* from it. The IN_OPEN arrives only later! So
 * we have a very short timeout of 2 seconds. */
 if (iev->mask & IN_CREATE) {
 queue_file_create(iev->name);
 } else if (iev->mask & IN_OPEN) {
 queue_file_open(iev->name);
 } else if (iev->mask & IN_CLOSE_WRITE) {
 queue_file_write(iev->name);
 } else if (iev->mask & IN_MOVED_TO) {
 queue_file(iev->name, 0);
 } else {
 ast_log(LOG_ERROR, "Unexpected event %d for file '%s'\n", (int) iev->mask, iev->name);
 }

 len = sizeof(*iev) + iev->len;
 res -= len;
 }
 } else if (res < 0 && errno != EINTR && errno != EAGAIN) {
 ast_debug(1, "Got an error back from %s(2): %s\n", stage ? "read" : "poll", strerror(errno));
 }
 time(&now);
 }
 queue_created_files();
#else
 int num_events;
 /* If queue empty then wait forever */
 if (next == INT_MAX) {
 num_events = kevent(inotify_fd, &kev, 1, &event, 1, NULL);
 } else {
 struct timespec ts2 = { .tv_sec = (unsigned long int)(next - now), .tv_nsec = 0 };
 num_events = kevent(inotify_fd, &kev, 1, &event, 1, &ts2);
 }
 if ((num_events < 0) || (event.flags == EV_ERROR)) {
 ast_debug(10, "KEvent error %s\n", strerror(errno));
 continue;
 } else if (num_events == 0) {
 /* Interrupt or timeout, restart calculations */
 continue;
 } else {
 /* Directory changed, rescan */
 rewinddir(dir);
 while ((de = readdir(dir))) {
 queue_file(de->d_name, 0);
 }
 }
 time(&now);
 }
#endif

 /* Empty the list of all entries ready to be processed */
 AST_LIST_LOCK(&dirlist);
 while (!AST_LIST_EMPTY(&dirlist) && AST_LIST_FIRST(&dirlist)->mtime <= now) {
 cur = AST_LIST_REMOVE_HEAD(&dirlist, list);
 queue_file(cur->name, cur->mtime);
 ast_free(cur);
 }
 AST_LIST_UNLOCK(&dirlist);
 }
 return NULL;
}

unload_module()

static int unload_module ( void  )

static

Definition at line 941 of file pbx_spool.c.

{
 return -1;
}

Variable Documentation

__mod_info

struct ast_module_info __mod_info = { .name = AST_MODULE, .flags = AST_MODFLAG_LOAD_ORDER , .description = "Outgoing Spool Support" , .key = "This paragraph is copyright (c) 2006 by Digium, Inc. \In order for your module to load, it must return this \key via a function called \"key\". Any code which \includes this paragraph must be licensed under the GNU \General Public License version 2 or later (at your \option). In addition to Digium's general reservations \of rights, Digium expressly reserves the right to \allow other parties to license this paragraph under \different terms. Any use of Digium, Inc. trademarks or \logos (including \"Asterisk\" or \"Digium\") without \express written permission of Digium, Inc. is prohibited.\n" , .buildopt_sum = "30ef0c93b36035ec78c9cfd712d36d9b" , .load = load_module, .unload = unload_module, .load_pri = AST_MODPRI_DEFAULT, .support_level = AST_MODULE_SUPPORT_CORE, }

static

Definition at line 965 of file pbx_spool.c.

ast_module_info

const struct ast_module_info* ast_module_info = &__mod_info

static

Definition at line 965 of file pbx_spool.c.

createlist

struct createlist createlist = { .first = NULL, .last = NULL, }

static

dirlist

struct dirlist dirlist = { .first = NULL, .last = NULL, .lock = { PTHREAD_RECURSIVE_MUTEX_INITIALIZER_NP , NULL, {1, 0} } , }

static

openlist

struct openlist openlist = { .first = NULL, .last = NULL, }

static

qdir

char qdir[255]

static

Definition at line 74 of file pbx_spool.c.

qdonedir

char qdonedir[255]

static

Definition at line 75 of file pbx_spool.c.