app_queue.c

#endif
			if (!f || ((f->frametype == AST_FRAME_CONTROL) && (f->subclass == AST_CONTROL_HANGUP))) {
				/* Got hung up */
				*to=-1;
				if (f)
					ast_frfree(f);
				return NULL;
			}
			if ((f->frametype == AST_FRAME_DTMF) && caller_disconnect && (f->subclass == '*')) {
				if (option_verbose > 3)
					ast_verbose(VERBOSE_PREFIX_3 "User hit %c to disconnect call.\n", f->subclass);
				*to=0;
				ast_frfree(f);
				return NULL;
			}
			if ((f->frametype == AST_FRAME_DTMF) && (f->subclass != '*') && valid_exit(qe, f->subclass)) {
				if (option_verbose > 3)
					ast_verbose(VERBOSE_PREFIX_3 "User pressed digit: %c\n", f->subclass);
				*to=0;
				*digit=f->subclass;
				ast_frfree(f);
				return NULL;
			}
			ast_frfree(f);
		}
		if (!*to) 
			rna(orig, qe, on);
	}

	return peer;
	
}

static int is_our_turn(struct queue_ent *qe)
{
	struct queue_ent *ch;
	struct member *cur;
	int avl = 0;
	int idx = 0;
	int res;

	if (!qe->parent->autofill) {
		/* Atomically read the parent head -- does not need a lock */
		ch = qe->parent->head;
		/* If we are now at the top of the head, break out */
		if (ch == qe) {
			if (option_debug)
				ast_log(LOG_DEBUG, "It's our turn (%s).\n", qe->chan->name);
			res = 1;
		} else {
			if (option_debug)
				ast_log(LOG_DEBUG, "It's not our turn (%s).\n", qe->chan->name);
			res = 0;
		}	

	} else {

		/* This needs a lock. How many members are available to be served? */
	
		ast_mutex_lock(&qe->parent->lock);
			
		ch = qe->parent->head;
		cur = qe->parent->members;
	
		while (cur) {
			if (cur->status == 1) 
				avl++;
			cur = cur->next;
		}

		if (option_debug)
			ast_log(LOG_DEBUG, "There are %d available members.\n", avl);
	
		if (qe->parent->strategy == QUEUE_STRATEGY_RINGALL) {
			if (option_debug)
				ast_log(LOG_DEBUG, "Even though there are %d available members, the strategy is ringall so only the head call is allowed in!\n", avl);
			avl = 1;
		}
	
		while ((idx < avl) && (ch) && (ch != qe)) {
			idx++;
			ch = ch->next;			
		}
	
		/* If the queue entry is within avl [the number of available members] calls from the top ... */
		if (ch && idx < avl) {
			if (option_debug)
				ast_log(LOG_DEBUG, "It's our turn (%s).\n", qe->chan->name);
			res = 1;
		} else {
			if (option_debug)
				ast_log(LOG_DEBUG, "It's not our turn (%s).\n", qe->chan->name);
			res = 0;
		}
		
		ast_mutex_unlock(&qe->parent->lock);
	}

	return res;
}

static int wait_our_turn(struct queue_ent *qe, int ringing, enum queue_result *reason)
{
	int res = 0;

	/* This is the holding pen for callers 2 through maxlen */
	for (;;) {
		enum queue_member_status stat;

		if (is_our_turn(qe))
			break;

		/* If we have timed out, break out */
		if (qe->expire && (time(NULL) > qe->expire)) {
			*reason = QUEUE_TIMEOUT;
			ast_queue_log(qe->parent->name, qe->chan->uniqueid,"NONE", "EXITWITHTIMEOUT", "%d", qe->pos);
			break;
		}

		stat = get_member_status(qe->parent, qe->max_penalty);

		/* leave the queue if no agents, if enabled */
		if (qe->parent->leavewhenempty && (stat == QUEUE_NO_MEMBERS)) {
			*reason = QUEUE_LEAVEEMPTY;
			ast_queue_log(qe->parent->name, qe->chan->uniqueid, "NONE", "EXITEMPTY", "%d|%d|%ld", qe->pos, qe->opos, (long)time(NULL) - qe->start);
			leave_queue(qe);
			break;
		}

		/* leave the queue if no reachable agents, if enabled */
		if ((qe->parent->leavewhenempty == QUEUE_EMPTY_STRICT) && (stat == QUEUE_NO_REACHABLE_MEMBERS)) {
			*reason = QUEUE_LEAVEUNAVAIL;
			ast_queue_log(qe->parent->name, qe->chan->uniqueid, "NONE", "EXITEMPTY", "%d|%d|%ld", qe->pos, qe->opos, (long)time(NULL) - qe->start);
			leave_queue(qe);
			break;
		}

		/* Make a position announcement, if enabled */
		if (qe->parent->announcefrequency && !ringing &&
		    (res = say_position(qe)))
			break;

		/* Make a periodic announcement, if enabled */
		if (qe->parent->periodicannouncefrequency && !ringing &&
		    (res = say_periodic_announcement(qe)))
			break;

		/* Wait a second before checking again */
		if ((res = ast_waitfordigit(qe->chan, RECHECK * 1000)))
			break;
	}
	return res;
}

static int update_queue(struct ast_call_queue *q, struct member *member)
{
	struct member *cur;

	/* Since a reload could have taken place, we have to traverse the list to
		be sure it's still valid */
	ast_mutex_lock(&q->lock);
	cur = q->members;
	while(cur) {
		if (member == cur) {
			time(&cur->lastcall);
			cur->calls++;
			break;
		}
		cur = cur->next;
	}
	q->callscompleted++;
	ast_mutex_unlock(&q->lock);
	return 0;
}

static int calc_metric(struct ast_call_queue *q, struct member *mem, int pos, struct queue_ent *qe, struct callattempt *tmp)
{
	if (mem->penalty > qe->max_penalty)
		return -1;

	switch (q->strategy) {
	case QUEUE_STRATEGY_RINGALL:
		/* Everyone equal, except for penalty */
		tmp->metric = mem->penalty * 1000000;
		break;
	case QUEUE_STRATEGY_ROUNDROBIN:
		if (!pos) {
			if (!q->wrapped) {
				/* No more channels, start over */
				q->rrpos = 0;
			} else {
				/* Prioritize next entry */
				q->rrpos++;
			}
			q->wrapped = 0;
		}
		/* Fall through */
	case QUEUE_STRATEGY_RRMEMORY:
		if (pos < q->rrpos) {
			tmp->metric = 1000 + pos;
		} else {
			if (pos > q->rrpos)
				/* Indicate there is another priority */
				q->wrapped = 1;
			tmp->metric = pos;
		}
		tmp->metric += mem->penalty * 1000000;
		break;
	case QUEUE_STRATEGY_RANDOM:
		tmp->metric = ast_random() % 1000;
		tmp->metric += mem->penalty * 1000000;
		break;
	case QUEUE_STRATEGY_FEWESTCALLS:
		tmp->metric = mem->calls;
		tmp->metric += mem->penalty * 1000000;
		break;
	case QUEUE_STRATEGY_LEASTRECENT:
		if (!mem->lastcall)
			tmp->metric = 0;
		else
			tmp->metric = 1000000 - (time(NULL) - mem->lastcall);
		tmp->metric += mem->penalty * 1000000;
		break;
	default:
		ast_log(LOG_WARNING, "Can't calculate metric for unknown strategy %d\n", q->strategy);
		break;
	}
	return 0;
}

static int try_calling(struct queue_ent *qe, const char *options, char *announceoverride, const char *url, int *go_on, const char *agi)
{
	struct member *cur;
	struct callattempt *outgoing=NULL; /* the queue we are building */
	int to;
	char oldexten[AST_MAX_EXTENSION]="";
	char oldcontext[AST_MAX_CONTEXT]="";
	char queuename[256]="";
	struct ast_channel *peer;
	struct ast_channel *which;
	struct callattempt *lpeer;
	struct member *member;
	struct ast_app *app;
	int res = 0, bridge = 0;
	int numbusies = 0;
	int x=0;
	char *announce = NULL;
	char digit = 0;
	time_t callstart;
	time_t now = time(NULL);
	struct ast_bridge_config bridge_config;
	char nondataquality = 1;
	char *agiexec = NULL;
	int ret = 0;
	const char *monitorfilename;
	const char *monitor_exec;
	const char *monitor_options;
	char tmpid[256], tmpid2[256];
	char meid[1024], meid2[1024];
	char mixmonargs[1512];
	struct ast_app *mixmonapp = NULL;
	char *p;


	memset(&bridge_config, 0, sizeof(bridge_config));
	time(&now);
		
	for (; options && *options; options++)
		switch (*options) {
		case 't':
			ast_set_flag(&(bridge_config.features_callee), AST_FEATURE_REDIRECT);
			break;
		case 'T':
			ast_set_flag(&(bridge_config.features_caller), AST_FEATURE_REDIRECT);
			break;
		case 'w':
			ast_set_flag(&(bridge_config.features_callee), AST_FEATURE_AUTOMON);
			break;
		case 'W':
			ast_set_flag(&(bridge_config.features_caller), AST_FEATURE_AUTOMON);
			break;
		case 'd':
			nondataquality = 0;
			break;
		case 'h':
			ast_set_flag(&(bridge_config.features_callee), AST_FEATURE_DISCONNECT);
			break;
		case 'H':
			ast_set_flag(&(bridge_config.features_caller), AST_FEATURE_DISCONNECT);
			break;
		case 'n':
			if ((now - qe->start >= qe->parent->timeout))
				*go_on = 1;
			break;
		}

	/* Hold the lock while we setup the outgoing calls */
	if (use_weight) 
		AST_LIST_LOCK(&queues);
	ast_mutex_lock(&qe->parent->lock);
	if (option_debug)
		ast_log(LOG_DEBUG, "%s is trying to call a queue member.\n", 
							qe->chan->name);
	ast_copy_string(queuename, qe->parent->name, sizeof(queuename));
	cur = qe->parent->members;
	if (!ast_strlen_zero(qe->announce))
		announce = qe->announce;
	if (!ast_strlen_zero(announceoverride))
		announce = announceoverride;

	for (;cur; cur = cur->next) {
		struct callattempt *tmp = ast_calloc(1, sizeof(*tmp));
		if (!tmp) {
			ast_mutex_unlock(&qe->parent->lock);
			if (use_weight) 
				AST_LIST_UNLOCK(&queues);
			goto out;
		}
		tmp->stillgoing = -1;
		if (option_debug) {
			if (url)
				ast_log(LOG_DEBUG, "Queue with URL=%s_\n", url);
			else 
				ast_log(LOG_DEBUG, "Simple queue (no URL)\n");
		}

		tmp->member = cur;		/* Never directly dereference!  Could change on reload */
		tmp->oldstatus = cur->status;
		tmp->lastcall = cur->lastcall;
		ast_copy_string(tmp->interface, cur->interface, sizeof(tmp->interface));
		/* Special case: If we ring everyone, go ahead and ring them, otherwise
		   just calculate their metric for the appropriate strategy */
		if (!calc_metric(qe->parent, cur, x++, qe, tmp)) {
			/* Put them in the list of outgoing thingies...  We're ready now. 
			   XXX If we're forcibly removed, these outgoing calls won't get
			   hung up XXX */
			tmp->q_next = outgoing;
			outgoing = tmp;		
			/* If this line is up, don't try anybody else */
			if (outgoing->chan && (outgoing->chan->_state == AST_STATE_UP))
				break;
		} else {
			free(tmp);
		}
	}
	if (qe->expire && (!qe->parent->timeout || (qe->expire - now) <= qe->parent->timeout))
		to = (qe->expire - now) * 1000;
	else
		to = (qe->parent->timeout) ? qe->parent->timeout * 1000 : -1;
	ring_one(qe, outgoing, &numbusies);
	ast_mutex_unlock(&qe->parent->lock);
	if (use_weight) 
		AST_LIST_UNLOCK(&queues);
	lpeer = wait_for_answer(qe, outgoing, &to, &digit, numbusies, ast_test_flag(&(bridge_config.features_caller), AST_FEATURE_DISCONNECT));
	ast_mutex_lock(&qe->parent->lock);
	if (qe->parent->strategy == QUEUE_STRATEGY_RRMEMORY) {
		store_next(qe, outgoing);
	}
	ast_mutex_unlock(&qe->parent->lock);
	peer = lpeer ? lpeer->chan : NULL;
	if (!peer) {
		if (to) {
			/* Must gotten hung up */
			res = -1;
		} else {
			res = digit;
		}
		if (option_debug)
			ast_log(LOG_DEBUG, "%s: Nobody answered.\n", qe->chan->name);
	} else { /* peer is valid */
		/* Ah ha!  Someone answered within the desired timeframe.  Of course after this
		   we will always return with -1 so that it is hung up properly after the 
		   conversation.  */
		qe->handled++;
		if (!strcmp(qe->chan->tech->type, "Zap"))
			ast_channel_setoption(qe->chan, AST_OPTION_TONE_VERIFY, &nondataquality, sizeof(nondataquality), 0);
		if (!strcmp(peer->tech->type, "Zap"))
			ast_channel_setoption(peer, AST_OPTION_TONE_VERIFY, &nondataquality, sizeof(nondataquality), 0);
		/* Update parameters for the queue */
		recalc_holdtime(qe);
		member = lpeer->member;
		hangupcalls(outgoing, peer);
		outgoing = NULL;
		if (announce || qe->parent->reportholdtime || qe->parent->memberdelay) {
			int res2;
			res2 = ast_autoservice_start(qe->chan);
			if (!res2) {
				if (qe->parent->memberdelay) {
					ast_log(LOG_NOTICE, "Delaying member connect for %d seconds\n", qe->parent->memberdelay);
					res2 |= ast_safe_sleep(peer, qe->parent->memberdelay * 1000);
				}
				if (!res2 && announce) {
					if (play_file(peer, announce))
						ast_log(LOG_WARNING, "Announcement file '%s' is unavailable, continuing anyway...\n", announce);
				}
				if (!res2 && qe->parent->reportholdtime) {
					if (!play_file(peer, qe->parent->sound_reporthold)) {
						int holdtime;

						time(&now);
						holdtime = abs((now - qe->start) / 60);
						if (holdtime < 2) {
							play_file(peer, qe->parent->sound_lessthan);
							ast_say_number(peer, 2, AST_DIGIT_ANY, peer->language, NULL);
						} else 
							ast_say_number(peer, holdtime, AST_DIGIT_ANY, peer->language, NULL);
						play_file(peer, qe->parent->sound_minutes);
					}
				}
			}
			res2 |= ast_autoservice_stop(qe->chan);
			if (peer->_softhangup) {
				/* Agent must have hung up */
				ast_log(LOG_WARNING, "Agent on %s hungup on the customer.  They're going to be pissed.\n", peer->name);
				ast_queue_log(queuename, qe->chan->uniqueid, peer->name, "AGENTDUMP", "%s", "");
				record_abandoned(qe);
				if (qe->parent->eventwhencalled) {
					manager_event(EVENT_FLAG_AGENT, "AgentDump",
						      "Queue: %s\r\n"
						      "Uniqueid: %s\r\n"
						      "Channel: %s\r\n"
						      "Member: %s\r\n",
						      queuename, qe->chan->uniqueid, peer->name, member->interface);
				}
				ast_hangup(peer);
				goto out;
			} else if (res2) {
				/* Caller must have hung up just before being connected*/
				ast_log(LOG_NOTICE, "Caller was about to talk to agent on %s but the caller hungup.\n", peer->name);
				ast_queue_log(queuename, qe->chan->uniqueid, peer->name, "ABANDON", "%d|%d|%ld", qe->pos, qe->opos, (long)time(NULL) - qe->start);
				record_abandoned(qe);
				ast_hangup(peer);
				return -1;
			}
		}
		/* Stop music on hold */
		ast_moh_stop(qe->chan);
		/* If appropriate, log that we have a destination channel */
		if (qe->chan->cdr)
			ast_cdr_setdestchan(qe->chan->cdr, peer->name);
		/* Make sure channels are compatible */
		res = ast_channel_make_compatible(qe->chan, peer);
		if (res < 0) {
			ast_queue_log(queuename, qe->chan->uniqueid, peer->name, "SYSCOMPAT", "%s", "");
			ast_log(LOG_WARNING, "Had to drop call because I couldn't make %s compatible with %s\n", qe->chan->name, peer->name);
			record_abandoned(qe);
			ast_hangup(peer);
			return -1;
		}
		/* Begin Monitoring */
		if (qe->parent->monfmt && *qe->parent->monfmt) {
			if (!qe->parent->montype) {
				if (option_debug)
					ast_log(LOG_DEBUG, "Starting Monitor as requested.\n");
				monitorfilename = pbx_builtin_getvar_helper(qe->chan, "MONITOR_FILENAME");
				if (pbx_builtin_getvar_helper(qe->chan, "MONITOR_EXEC") || pbx_builtin_getvar_helper(qe->chan, "MONITOR_EXEC_ARGS"))
					which = qe->chan;
				else
					which = peer;
				if (monitorfilename)
					ast_monitor_start(which, qe->parent->monfmt, monitorfilename, 1 );
				else if (qe->chan->cdr) 
					ast_monitor_start(which, qe->parent->monfmt, qe->chan->cdr->uniqueid, 1 );
				else {
					/* Last ditch effort -- no CDR, make up something */
					snprintf(tmpid, sizeof(tmpid), "chan-%lx", ast_random());
					ast_monitor_start(which, qe->parent->monfmt, tmpid, 1 );
				}
				if (qe->parent->monjoin)
					ast_monitor_setjoinfiles(which, 1);
			} else {
				if (option_debug)
					ast_log(LOG_DEBUG, "Starting MixMonitor as requested.\n");
				monitorfilename = pbx_builtin_getvar_helper(qe->chan, "MONITOR_FILENAME");
				if (!monitorfilename) {
					if (qe->chan->cdr)
						ast_copy_string(tmpid, qe->chan->cdr->uniqueid, sizeof(tmpid)-1);
					else 
						snprintf(tmpid, sizeof(tmpid), "chan-%lx", ast_random());
				} else {
					ast_copy_string(tmpid2, monitorfilename, sizeof(tmpid2)-1);
					for (p = tmpid2; *p ; p++) {
						if (*p == '^' && *(p+1) == '{') {
							*p = '$';
						}
					}

					memset(tmpid, 0, sizeof(tmpid));
					pbx_substitute_variables_helper(qe->chan, tmpid2, tmpid, sizeof(tmpid) - 1);
				}

				monitor_exec = pbx_builtin_getvar_helper(qe->chan, "MONITOR_EXEC");
				monitor_options = pbx_builtin_getvar_helper(qe->chan, "MONITOR_OPTIONS");

				if (monitor_exec) {
					ast_copy_string(meid2, monitor_exec, sizeof(meid2)-1);
					for (p = meid2; *p ; p++) {
						if (*p == '^' && *(p+1) == '{') {
							*p = '$';
						}
					}

					memset(meid, 0, sizeof(meid));
					pbx_substitute_variables_helper(qe->chan, meid2, meid, sizeof(meid) - 1);
				} 
	
				snprintf(tmpid2, sizeof(tmpid2)-1, "%s.%s", tmpid, qe->parent->monfmt);

				mixmonapp = pbx_findapp("MixMonitor");

				if (strchr(tmpid2, '|')) {
					ast_log(LOG_WARNING, "monitor-format (in queues.conf) and MONITOR_FILENAME cannot contain a '|'! Not recording.\n");
					mixmonapp = NULL;
				}

				if (!monitor_options)
					monitor_options = ast_strdupa("");
				
				if (strchr(monitor_options, '|')) {
					ast_log(LOG_WARNING, "MONITOR_OPTIONS cannot contain a '|'! Not recording.\n");
					mixmonapp = NULL;
				}


				if (mixmonapp) {
					if (!ast_strlen_zero(monitor_exec) && !ast_strlen_zero(monitor_options)) 
						snprintf(mixmonargs, sizeof(mixmonargs)-1, "%s|b%s|%s", tmpid2, monitor_options, monitor_exec);
					else 
						snprintf(mixmonargs, sizeof(mixmonargs)-1, "%s|b%s", tmpid2, monitor_options);
						
					if (option_debug)
						ast_log(LOG_DEBUG, "Arguments being passed to MixMonitor: %s\n", mixmonargs);

					ret = pbx_exec(qe->chan, mixmonapp, mixmonargs);

				} else
					ast_log(LOG_WARNING, "Asked to run MixMonitor on this call, but cannot find the MixMonitor app!\n");

			}
		}
		/* Drop out of the queue at this point, to prepare for next caller */
		leave_queue(qe);			
 		if (!ast_strlen_zero(url) && ast_channel_supports_html(peer)) {
			if (option_debug)
	 			ast_log(LOG_DEBUG, "app_queue: sendurl=%s.\n", url);
 			ast_channel_sendurl(peer, url);
 		}
		if (qe->parent->setinterfacevar)
				pbx_builtin_setvar_helper(qe->chan, "MEMBERINTERFACE", member->interface);
		if (!ast_strlen_zero(agi)) {
			if (option_debug)
				ast_log(LOG_DEBUG, "app_queue: agi=%s.\n", agi);
			app = pbx_findapp("agi");
			if (app) {
				agiexec = ast_strdupa(agi);
				ret = pbx_exec(qe->chan, app, agiexec);
			} else 
				ast_log(LOG_WARNING, "Asked to execute an AGI on this channel, but could not find application (agi)!\n");
		}
		ast_queue_log(queuename, qe->chan->uniqueid, peer->name, "CONNECT", "%ld|%s", (long)time(NULL) - qe->start, peer->uniqueid);
		if (qe->parent->eventwhencalled)
			manager_event(EVENT_FLAG_AGENT, "AgentConnect",
				      "Queue: %s\r\n"
				      "Uniqueid: %s\r\n"
				      "Channel: %s\r\n"
				      "Member: %s\r\n"
				      "Holdtime: %ld\r\n"
				      "BridgedChannel: %s\r\n",
				      queuename, qe->chan->uniqueid, peer->name, member->interface,
				      (long)time(NULL) - qe->start,peer->uniqueid);
		ast_copy_string(oldcontext, qe->chan->context, sizeof(oldcontext));
		ast_copy_string(oldexten, qe->chan->exten, sizeof(oldexten));
		time(&callstart);

		bridge = ast_bridge_call(qe->chan,peer, &bridge_config);

		if (strcasecmp(oldcontext, qe->chan->context) || strcasecmp(oldexten, qe->chan->exten)) {
			ast_queue_log(queuename, qe->chan->uniqueid, peer->name, "TRANSFER", "%s|%s|%ld|%ld", 
					  qe->chan->exten, qe->chan->context, (long)(callstart - qe->start), (long)(time(NULL) - callstart));
		} else if (qe->chan->_softhangup) {
			ast_queue_log(queuename, qe->chan->uniqueid, peer->name, "COMPLETECALLER", "%ld|%ld",
				      (long)(callstart - qe->start), (long)(time(NULL) - callstart));
			if (qe->parent->eventwhencalled)
				manager_event(EVENT_FLAG_AGENT, "AgentComplete",
					      "Queue: %s\r\n"
					      "Uniqueid: %s\r\n"
					      "Channel: %s\r\n"
					      "Member: %s\r\n"
					      "HoldTime: %ld\r\n"
					      "TalkTime: %ld\r\n"
					      "Reason: caller\r\n",
					      queuename, qe->chan->uniqueid, peer->name, member->interface,
					      (long)(callstart - qe->start), (long)(time(NULL) - callstart));
		} else {
			ast_queue_log(queuename, qe->chan->uniqueid, peer->name, "COMPLETEAGENT", "%ld|%ld", (long)(callstart - qe->start), (long)(time(NULL) - callstart));
			if (qe->parent->eventwhencalled)
				manager_event(EVENT_FLAG_AGENT, "AgentComplete",
					      "Queue: %s\r\n"
					      "Uniqueid: %s\r\n"
					      "Channel: %s\r\n"
					      "HoldTime: %ld\r\n"
					      "TalkTime: %ld\r\n"
					      "Reason: agent\r\n",
					      queuename, qe->chan->uniqueid, peer->name, (long)(callstart - qe->start),
					      (long)(time(NULL) - callstart));
		}

		if (bridge != AST_PBX_NO_HANGUP_PEER)
			ast_hangup(peer);
		update_queue(qe->parent, member);
		res = bridge ? bridge : 1;
	}
out:
	hangupcalls(outgoing, NULL);
	return res;
}

static int wait_a_bit(struct queue_ent *qe)
{
	/* Don't need to hold the lock while we setup the outgoing calls */
	int retrywait = qe->parent->retry * 1000;

	return ast_waitfordigit(qe->chan, retrywait);
}

static struct member * interface_exists(struct ast_call_queue *q, char *interface)
{
	struct member *mem;

	if (q)
		for (mem = q->members; mem; mem = mem->next)
			if (!strcasecmp(interface, mem->interface))
				return mem;

	return NULL;
}


/* Dump all members in a specific queue to the database
 *
 * <pm_family>/<queuename> = <interface>;<penalty>;<paused>[|...]
 *
 */
static void dump_queue_members(struct ast_call_queue *pm_queue)
{
	struct member *cur_member;
	char value[PM_MAX_LEN];
	int value_len = 0;
	int res;

	memset(value, 0, sizeof(value));

	if (!pm_queue)
		return;

	for (cur_member = pm_queue->members; cur_member; cur_member = cur_member->next) {
		if (!cur_member->dynamic)
			continue;

		res = snprintf(value + value_len, sizeof(value) - value_len, "%s;%d;%d%s",
			       cur_member->interface, cur_member->penalty, cur_member->paused,
			       cur_member->next ? "|" : "");
		if (res != strlen(value + value_len)) {
			ast_log(LOG_WARNING, "Could not create persistent member string, out of space\n");
			break;
		}
		value_len += res;
	}
	
	if (value_len && !cur_member) {
		if (ast_db_put(pm_family, pm_queue->name, value))
			ast_log(LOG_WARNING, "failed to create persistent dynamic entry!\n");
	} else
		/* Delete the entry if the queue is empty or there is an error */
		ast_db_del(pm_family, pm_queue->name);
}

static int remove_from_queue(char *queuename, char *interface)
{
	struct ast_call_queue *q;
	struct member *last_member, *look;
	int res = RES_NOSUCHQUEUE;

	AST_LIST_LOCK(&queues);
	AST_LIST_TRAVERSE(&queues, q, list) {
		ast_mutex_lock(&q->lock);
		if (!strcmp(q->name, queuename)) {
			if ((last_member = interface_exists(q, interface))) {
				if ((look = q->members) == last_member) {
					q->members = last_member->next;
				} else {
					while (look != NULL) {
						if (look->next == last_member) {
							look->next = last_member->next;
							break;
						} else {
							 look = look->next;
						}
					}
				}
				manager_event(EVENT_FLAG_AGENT, "QueueMemberRemoved",
						"Queue: %s\r\n"
						"Location: %s\r\n",
					q->name, last_member->interface);
				free(last_member);

				if (queue_persistent_members)
					dump_queue_members(q);

				res = RES_OKAY;
			} else {
				res = RES_EXISTS;
			}
			ast_mutex_unlock(&q->lock);
			break;
		}
		ast_mutex_unlock(&q->lock);
	}
	if (res == RES_OKAY) {
		remove_from_interfaces(interface);
	}
	AST_LIST_UNLOCK(&queues);
	return res;
}


static int add_to_queue(char *queuename, char *interface, int penalty, int paused, int dump)
{
	struct ast_call_queue *q;
	struct member *new_member;
	int res = RES_NOSUCHQUEUE;

	/* \note Ensure the appropriate realtime queue is loaded.  Note that this
	 * short-circuits if the queue is already in memory. */
	q = load_realtime_queue(queuename);

	AST_LIST_LOCK(&queues);

	if (q) {
		ast_mutex_lock(&q->lock);
		if (interface_exists(q, interface) == NULL) {

			add_to_interfaces(interface);

			new_member = create_queue_member(interface, penalty, paused);

			if (new_member != NULL) {
				new_member->dynamic = 1;
				new_member->next = q->members;
				q->members = new_member;
				manager_event(EVENT_FLAG_AGENT, "QueueMemberAdded",
						"Queue: %s\r\n"
						"Location: %s\r\n"
						"Membership: %s\r\n"
						"Penalty: %d\r\n"
						"CallsTaken: %d\r\n"
						"LastCall: %d\r\n"
						"Status: %d\r\n"
						"Paused: %d\r\n",
						q->name, new_member->interface, new_member->dynamic ? "dynamic" : "static",
						new_member->penalty, new_member->calls, (int)new_member->lastcall, new_member->status, new_member->paused);

				if (dump)
					dump_queue_members(q);

				res = RES_OKAY;
			} else {
				res = RES_OUTOFMEMORY;
			}
		} else {
			res = RES_EXISTS;
		}
		ast_mutex_unlock(&q->lock);
	}
	AST_LIST_UNLOCK(&queues);
	return res;
}

static int set_member_paused(char *queuename, char *interface, int paused)
{
	int found = 0;
	struct ast_call_queue *q;
	struct member *mem;

	/* Special event for when all queues are paused - individual events still generated */

	if (ast_strlen_zero(queuename))
		ast_queue_log("NONE", "NONE", interface, (paused ? "PAUSEALL" : "UNPAUSEALL"), "%s", "");

	AST_LIST_LOCK(&queues);
	AST_LIST_TRAVERSE(&queues, q, list) {
		ast_mutex_lock(&q->lock);
		if (ast_strlen_zero(queuename) || !strcasecmp(q->name, queuename)) {
			if ((mem = interface_exists(q, interface))) {
				found++;
				if (mem->paused == paused)
					ast_log(LOG_DEBUG, "%spausing already-%spaused queue member %s:%s\n", (paused ? "" : "un"), (paused ? "" : "un"), q->name, interface);
				mem->paused = paused;

				if (queue_persistent_members)
					dump_queue_members(q);

				ast_queue_log(q->name, "NONE", interface, (paused ? "PAUSE" : "UNPAUSE"), "%s", "");

				manager_event(EVENT_FLAG_AGENT, "QueueMemberPaused",
					"Queue: %s\r\n"
					"Location: %s\r\n"
					"Paused: %d\r\n",
						q->name, mem->interface, paused);
			}
		}
		ast_mutex_unlock(&q->lock);
	}
	AST_LIST_UNLOCK(&queues);

	if (found)
		return RESULT_SUCCESS;
	else
		return RESULT_FAILURE;
}

/* Reload dynamic queue members persisted into the astdb */
static void reload_queue_members(void)
{
	char *cur_ptr;	
	char *queue_name;
	char *member;
	char *interface;
	char *penalty_tok;
	int penalty = 0;
	char *paused_tok;
	int paused = 0;
	struct ast_db_entry *db_tree;
	struct ast_db_entry *entry;
	struct ast_call_queue *cur_queue;
	char queue_data[PM_MAX_LEN];

	AST_LIST_LOCK(&queues);

	/* Each key in 'pm_family' is the name of a queue */
	db_tree = ast_db_gettree(pm_family, NULL);
	for (entry = db_tree; entry; entry = entry->next) {

		queue_name = entry->key + strlen(pm_family) + 2;

		AST_LIST_TRAVERSE(&queues, cur_queue, list) {
			ast_mutex_lock(&cur_queue->lock);
			if (!strcmp(queue_name, cur_queue->name))
				break;
			ast_mutex_unlock(&cur_queue->lock);
		}

		if (!cur_queue) {
			/* If the queue no longer exists, remove it from the
			 * database */
			ast_db_del(pm_family, queue_name);
			continue;
		} else
			ast_mutex_unlock(&cur_queue->lock);

		if (ast_db_get(pm_family, queue_name, queue_data, PM_MAX_LEN))
			continue;

		cur_ptr = queue_data;
		while ((member = strsep(&cur_ptr, "|"))) {
			if (ast_strlen_zero(member))
				continue;

			interface = strsep(&member, ";");
			penalty_tok = strsep(&member, ";");
			paused_tok = strsep(&member, ";");

			if (!penalty_tok) {
				ast_log(LOG_WARNING, "Error parsing persistent member string for '%s' (penalty)\n", queue_name);
				break;
			}
			penalty = strtol(penalty_tok, NULL, 10);
			if (errno == ERANGE) {
				ast_log(LOG_WARNING, "Error converting penalty: %s: Out of range.\n", penalty_tok);
				break;
			}
			
			if (!paused_tok) {
				ast_log(LOG_WARNING, "Error parsing persistent member string for '%s' (paused)\n", queue_name);
				break;
			}
			paused = strtol(paused_tok, NULL, 10);
			if ((errno == ERANGE) || paused < 0 || paused > 1) {
				ast_log(LOG_WARNING, "Error converting paused: %s: Expected 0 or 1.\n", paused_tok);
				break;
			}

			if (option_debug)
				ast_log(LOG_DEBUG, "Reload Members: Queue: %s  Member: %s  Penalty: %d  Paused: %d\n", queue_name, interface, penalty, paused);
			
			if (add_to_queue(queue_name, interface, penalty, paused, 0) == RES_OUTOFMEMORY) {
				ast_log(LOG_ERROR, "Out of Memory when reloading persistent queue member\n");
				break;
			}
		}
	}

	AST_LIST_UNLOCK(&queues);
	if (db_tree) {
		ast_log(LOG_NOTICE, "Queue members successfully reloaded from database.\n");
		ast_db_freetree(db_tree);
	}
}

static int pqm_exec(struct ast_channel *chan, void *data)
{
	struct localuser *lu;
	char *parse;
	int priority_jump = 0;
	AST_DECLARE_APP_ARGS(args,
		AST_APP_ARG(queuename);
		AST_APP_ARG(interface);
		AST_APP_ARG(options);
	);

	if (ast_strlen_zero(data)) {
		ast_log(LOG_WARNING, "PauseQueueMember requires an argument ([queuename]|interface[|options])\n");
		return -1;
	}

	parse = ast_strdupa(data);

	AST_STANDARD_APP_ARGS(args, parse);

	LOCAL_USER_ADD(lu);

	if (args.options) {
		if (strchr(args.options, 'j'))
			priority_jump = 1;
	}

	if (ast_strlen_zero(args.interface)) {
		ast_log(LOG_WARNING, "Missing interface argument to PauseQueueMember ([queuename]|interface[|options])\n");
		LOCAL_USER_REMOVE(lu);
		return -1;
	}

	if (set_member_paused(args.queuename, args.interface, 1)) {
		ast_log(LOG_WARNING, "Attempt to pause interface %s, not found\n", args.interface);
		if (priority_jump || ast_opt_priority_jumping) {
			if (ast_goto_if_exists(chan, chan->context, chan->exten, chan->priority + 101)) {
				pbx_builtin_setvar_helper(chan, "PQMSTATUS", "NOTFOUND");
				LOCAL_USER_REMOVE(lu);
				return 0;
			}
		}
		LOCAL_USER_REMOVE(lu);
		pbx_builtin_setvar_helper(chan, "PQMSTATUS", "NOTFOUND");
		return -1;
	}

	LOCAL_USER_REMOVE(lu);
	pbx_builtin_setvar_helper(chan, "PQMSTATUS", "PAUSED");
	return 0;
}

static int upqm_exec(struct ast_channel *chan, void *data)
{
	struct localuser *lu;
	char *parse;
	int priority_jump = 0;
	AST_DECLARE_APP_ARGS(args,
		AST_APP_ARG(queuename);
		AST_APP_ARG(interface);
		AST_APP_ARG(options);
	);

	if (ast_strlen_zero(data)) {
		ast_log(LOG_WARNING, "UnpauseQueueMember requires an argument ([queuename]|interface[|options])\n");
		return -1;
	}

	parse = ast_strdupa(data);

	AST_STANDARD_APP_ARGS(args, parse);

	LOCAL_USER_ADD(lu);

	if (args.options) {
		if (strchr(args.options, 'j'))
			priority_jump = 1;
	}

	if (ast_strlen_zero(args.interface)) {
		ast_log(LOG_WARNING, "Missing interface argument to PauseQueueMember ([queuename]|interface[|options])\n");
		LOCAL_USER_REMOVE(lu);
		return -1;
	}

	if (set_member_paused(args.queuename, args.interface, 0)) {
		ast_log(LOG_WARNING, "Attempt to unpause interface %s, not found\n", args.interface);
		if (priority_jump || ast_opt_priority_jumping) {
			if (ast_goto_if_exists(chan, chan->context, chan->exten, chan->priority + 101)) {
				pbx_builtin_setvar_helper(chan, "UPQMSTATUS", "NOTFOUND");
				LOCAL_USER_REMOVE(lu);