Asterisk — фи, это же моветон
Здравствуйте уважаемые читатели этого замечательного ресурса. По уже сложившейся традиции — являюсь давним читателем habr'а, но только сейчас решил сделать пост. Что, собственно, побудило к написанию? Честно сказать, и сам не знаю. То ли притянутые статьи о производительности FreeSWITCH/Yate/3CX/etc в сравнении с Asterisk, то ли действительные, реальные проблемы архитектуры последнего, а, возможно, желание сделать что-нибудь уникальное.
И что удивительно, в первом случае, как правило, сравнивают мягкое и теплое, так сказать, FreeSWITCH/Yate/etc и FreePBX. Да-да, именно FreePBX. Это не опечатка. Причем интересно, что во всех сравнениях зачастую один Asterisk в дефолтной конфигурации. Ну, вы знаете, эта конфигурация — загруженные все имеющиеся модули, кривой диалплан (FreePBX как бы способствует) и куча остальной необъективщины. Что до родовых болячек Asterisk'а — да, объективно их вагон и маленькая тележка.
Что со всем этим делать? Разрушать стереотипы и исправлять родовые травмы. Этим и займемся.
Скрещиваем ежа с ужом
Многие из новичков испытывают дискомфорт, глядя на синтаксис описания диалплана в Asterisk'е, а некоторые на полном серьезе обосновывают выбор другого сервера телефонии именно необходимостью писать диалплан в том виде, в котором он есть по дефолту. Типа, перелопачивать многострочный XML — это верх комфорта. Да, есть возможность юзать LUA/AEL, и это хорошо. Но лично я отнес бы эту возможность в минусы и в частности то, что касается pbx_lua.
Как уже было сказано, иметь возможность описывать диалплан полноценным языком программирования — это хорошо. Проблема в том, что время жизни скрипта и его окружения равно времени жизни канала. Для каждого нового канала запускается свой экземпляр скрипта, следовательно, прощай, разделяемые между каналами переменные, единичная загрузка сторонних модулей, один разделяемый коннект к базе и т.д., и т.п. Строго говоря, от встраиваемого языка описания сценария этого и не нужно, но уж сильно хочется. А если хочется – значит, надо.
Итак, от классического Asterisk'а возьмем принципы pbx_lua, от Yate возьмем модель маршрутизации, а от FreeSWITCH ничего брать не будем, ибо "overhead" не нужен. ОК, с тем, что нам нужно родить, определились. Что же будем использовать для генетических экспериментов:
- Asterisk, причем хотелось бы без привязок к версии. Тот же ARI был анонсирован, если мне не изменяет память, с 12-й версии. Если учесть, что до сих пор где-то юзаются 1.8/1.6, а возможно и 1.4, то зависимость от версионных плюшек нам не нужна.
- Lua — замечательный, гибкий и крайне функциональный скриптовый язык. Сам бог велел, так сказать, без комментариев.
- Lunapark — интересный проект на github'е, своего рода сервер voip-приложений.
Про Lunapark стоит рассказать подробнее. Это сервер, реализующий потенциал AMI-протокола в связке с классическим FastAGI, что немаловажно в едином пространстве выполнения. То есть, получаем аналог ARI посредством тесной кооперации AGI и AMI в одном флаконе.
Предвижу логичный вопрос: для чего это все? Есть же Asterisk REST Interface, чей функционал ты тут пытаешься переизобрести! Ответ на этот вопрос неоднозначен. Согласен, ARI декларирует ряд преимуществ: да, он асинхронен, да, позволяет работать с "сырыми" примитивами, WebSockets и да, стильный, модный, молодежный XML/JSON — куда ж без него. Но, черт возьми, часть этих так называемых преимуществ крайне сомнительна и добавляет один, а то и более уровней абстракции. Другая же часть — вообще не преимущества. Преимущества — это когда что-то свойственно только тебе, ниже мы это увидим на примере той же асинхронности.
Как это работает? Стандартными средствами заворачиваем канал в FastAGI-приложение, внутри которого получаем возможность управлять звонком, как будто юзаем pbx_lua с незначительным изменением синтаксиса. Вишенкой на торте является возможность управлять состоянием самого Asterisk’а и окружением канала, для этого в распоряжении текущего FastAGI-приложения есть глобальный AMI-объект. Кстати, можно не заворачивать канал в FastAGI-приложение, а создать глобальный обработчик события, допустим, для NewChannel. А это уже преимущество по сравнению с ARI, там как известно, вне stasis'а ARI слеп.
Реализован Lunapark в лучших традициях кооперативной многозадачности, а именно всеми любимая асинхронность на сопрограммах. И как следствие отсутствие проблем с "shared data". То есть плюсы присутствуют, но и проблемы появляются. Одна из них — это необходимость описывать логику с оглядкой на асинхронность, но я думаю, это мы как-нибудь переживем.
Что дальше?
Синтаксис описания контекстов — а что же с ним не так? Да все с ним нормально, более того, практику написания диалплана нужно прописывать как профилактику для формирования структурированного мышления у новичков. Но, вместе с тем нужно понижать порог вхождения. Поэтому будем упрощать и в то же время добавлять функционала.
Простой пример:
[test]
exten => _XXX/102,1,Hangup()
exten => _XXX,1,Dial(SIP/${EXTEN})В этом примере идет дозвон до трехзнака, кроме абонента 102. Вроде бы все логично и лаконично за исключением того, что шаблоны соответствия экстеншена ограничены небольшим набором правил, а так называемая extended маршрутизация возможна только по CallerID звонящего. А хотелось бы, к примеру, по CallerIDName или по текущему состоянию звонящего канала, а возможно по имени самого канала, а если реализовать полноценный regexp, так вообще красота. И да, я знаю, все эти хотелки можно реализовать, расписав контекст в таком виде:
[test]
exten => _XXX/102,1,Hangup()
; по CallerIDName
exten => _XXX,1,ExecIf($[ "${CALLERID(name)}" == "Vasya" ]?Hangup())
; По состоянию канала
exten => _XXX,n,ExecIf($[ "${CHANNEL(state)}" != "Ring" ]?Hangup())
; По имени канала
exten => _XXX,n,ExecIf($[ "${CUT(CUT(CHANNEL,-,1),/,2)}" == "333" ]?Hangup())
exten => _XXX,n,Dial(SIP/${EXTEN})Но мой внутренний перфекционист начинает бунтовать при виде такого, а если представить аналогичную выборку по всем пользователям, да еще и действия нужны разные и посложнее Hangup'а, то extensions.conf превращается в длииинную портянку вызовов Goto, GoSub, Macro и, не дай бог, с каналами типа Local.
Выход один — прикручивать свои правила маршрутизации с подкидным и дамами с низкой социальной ответственностью.
В качестве примера:
${Exten}:match('%d%d%d')
and
(
${CallerIDNum}:match('201') or
${CallerIDName}:match('Vasya') or
${State}:lower() ~= 'ring' or
${Channel}:match('^[^/]+/([^%-]+)') == '333'
) => Hangup();
${Exten}:match('%d%d%d') => Dial {callee = ('SIP/%s'):format(${Exten})};Хм, вырвиглазненько получилось, но на удивление читается и понимается с первого взгляда. А самое главное, что у нас появился аналог regexp'ов и группировка правил на действие, что, несомненно, упростит составление маршрутов в будущем.
Что тут думать, прыгать надо.
В итоге имеем Lunapark как замену pbx_lua. Его средствами нам и нужно создать логику обработки нашей модели маршрутизации. Для начала нужно распарсить набор правил и заменить все вхождения ${...} на соответствующие им значения, то есть привести к виду ('...'). Значения будут браться из окружения текущего канала.
Затем приводим каждое правило к виду условного оператора, чтобы получить нечто похожее:
-- Exten = 123
-- Sate = Ring
-- CallerIDNum = 100
-- CallerIDName = Test
-- Channel = SIP/100-00000012c
if ('123'):match('%d%d%d') and
(
('100'):match('201') or
('Test'):match('Vasya') or
('Ring'):lower() ~= 'ring' or
('SIP/100-00000012c'):match('^[^/]+/([^%-]+)') == '333'
) then
Hangup()
end
if ('123'):match('%d%d%d') then
Dial {callee = ('SIP/%s'):format(('123'))}
endДелать это будут две функции fmt и syntax соответственно:
local fmt = function(str, tab)
return (str:gsub('(%${[^}{]+})', function(w)
local mark = w:sub(3, -2)
return (mark:gsub('(.+)',function(v)
local out = tab[v] or v
return ("('%s')"):format(out)
end))
end))
end
local syntax = function(str)
return (str:gsub('([^;]+)=>([^;]+)',function(p,r)
return ([[
if %s then
%s
end
]]):format(p,r)
end))
endВ принципе ничего сложного, все просто и понятно. Идем дальше — считывать наши правила будем из файла в переменную при старте сервиса, а парсить их уже при звонке с актуальным окружением. Считыванием правил займется функция routes.
local routes = function(...)
local conf, content = ...
local f, err = io.open(conf, "r")
if io.type(f) ~= 'file' then
log.warn(err) -- Глобальный LOG объект доступный благодаря Lunapark'у
return ""
else
content = f:read('*all')
end
f:close() return content
endОсталось сделать две вещи: завернуть звонки в Lunapark и соответственно их обработать с учетом наших маршрутов. Тут стоит немного пояснить такой момент — в Lunapark вся логика описывается в handler'е. Это текстовый файл, в котором мы будем определять наши FastAGI-приложения и работать с AMI и нашими маршрутами.
Как уже было сказано, объект AMI — глобальный и, помимо роли AMI-клиента, может устанавливать своего рода слушатели для конкретных AMI событий. Этим мы и воспользуемся, но для начала сделаем некоторые приготовления в extensions.conf.
[default]
exten => _[hit],1,NoOp()
exten => _.,n,Wait(5)
exten => _.,1,AGI(agi://127.0.0.1/${EXTEN}${IF($[ "X${PRMS}" != "X" ]?"?${PRMS}")})Wait(5) в примере выше позволит нам не обрывать канал при завершении FastAGI-приложения, так как в маршрутах может быть описано несколько приложений, а выполнение их осуществляется по средствам Redirect на контекст default по ${EXTEN}.
Таким образом, беря во внимание все выше описанное и помня о кооперативной природе Lunapark'а, попробуем закодить логику обработки маршрутов через FastAGI-приложения.
-- Считываем наши правила в переменную rules
local rules = routes('routes.conf')
-- Очищаем все обработчики, таким образом очистятся только не именные обработчики
-- Это даст возможность не затирать цепочку выполнения при сигналах HUP/QUIT
ami.removeEvents('*')
-- Обработчик события создания нового канала
ami.addEvents {
['newchannel'] = function(e)
-- Условия, только каналы с набором и каналы с контекстом users
if (e['Context'] and e['Context']:match('users')) and e['Exten'] then
-- Переменная, указывающая на выполнении, какого FastAGI приложения мы находимся
local step
-- Будущий порядковый номер FatsAGI приложения в цепочке выполнения
local count = 0
-- Парсим маршруты для текущего окружения канала
local code, err = loadstring(syntax(fmt(rules,e)))
-- В описании маршрутов нет ошибок, двигаемся дальше
if type(code) == 'function' then
-- Проксируем будущие FastAGI приложения
setfenv(code,setmetatable({indexes = {}},{__index = function(t,k)
-- Вот они последствия кооперативности
return coroutine.wrap(
function(...)
local prms = {} -- Будущие параметры FastAGI приложения
local owner = t -- Копия окружения
local event = e -- Копия таблицы event
local thread = coroutine.running() -- ID текущей сопрограммы
-- Парсим параметры и приводим к виду URI
for p,v in pairs({...}) do
if type(v) == 'table' then
for key, val in pairs(v) do
table.insert(prms,("%s=%s"):format(key,val))
end
else
table.insert(prms,("%s=%s"):format(p,v))
end
end
-- Если это не первое FastAGI приложение в цепочке
if step then
-- Запоминаем предыдущее перед этим
local last = ("%s"):format(step)
-- Добавляем ИМЕННЫЕ обработчики события UserEvent по доп. условиям
-- И записываем в таблицу indexes(в окружении) их порядковые номера
-- Именные обработчики требуют последующего удаления самостоятельно
table.insert(owner['indexes'],ami.addEvent('UserEvent',function(evt)
-- Ловим событие AGIStatus указывающее на завершение приложения
-- Если это предыдущее перед нами, пробуждаем сопрограмму
if (evt['Channel'] and evt['Channel'] == event['Channel'])
and
(evt['UserEvent'] and evt['UserEvent']:match('AGIStatus'))
and
(evt['Script'] and evt['Script'] == last)
then
-- Соответствие порядкового номера нашей сопрограмме
-- В цепочке может быть вызов одного приложения несколько раз
-- Это позволит выполнять сопрограммы в порядке их определения
if owner['indexes'][count] == thread then
if coroutine.status(thread) ~= 'dead' then
coroutine.resume(thread)
end
end
end
end,thread))
-- Устанавливаем маркер текущего FastAGI приложения
step = k
-- Приостанавливаем сопрограмму
coroutine.yield()
else -- Здесь обрабатывается первое FastAGI приложение в цепочке
local index -- Индекс для обработчика Hangup события
-- Устанавливаем маркер текущего FastAGI приложения
step = k
-- Добавляем ИМЕННОЙ обработчик события Hangup для канала
-- В этом месте подчищаем за собой
index = ami.addEvent('Hangup',function(evt)
if evt['Channel'] and evt['Channel'] == event['Channel'] then
-- Удаляем обработчик событие Hangup по ранее запомненному индексу
ami.removeEvent('Hangup',index)
-- Удаляем все обработчики цепочек выполнения по индексу
for _,v in pairs(owner['indexes']) do
ami.removeEvent('UserEvent',v)
end
-- Делаем приятно сборщику мусора
owner = nil
end
end,thread)
end
-- По средствам AMI выставляем переменную для канала и вызова в цепочке
ami.setvar{
Value = table.concat(prms,'&'),
Channel = event['Channel'],
Variable = 'PRMS'
}
-- Перенаправляем канал на AGI-приложение через контекст default
ami.redirect{
Exten = k,
Priority = 1,
Channel = event['Channel'],
Context = 'default'
}
-- Выставляем индекс приложения
count = count + 1
end)
end}))()
else
-- Если что-то пошло не так
log.warn(err)
end
end
end
}Для тех, кто не смотрел или не понял код, поясню простым языком, что же мы там накодили. Для каждого действия в маршруте создается сопрограмма с копированием всего, чего нужно через замыкания и в приостановленном состоянии. Кроме сопрограммы первого действия, она выполняется как есть. Приостанавливаем мы их для эмуляции последовательного выполнения, то есть строго одна за другой, вторая после первой, третья после второй и т.д.
Стоп, так они и так друг за другом выполняются. На самом деле нет — если не эмулировать синхронность, то асинхронный redirect пробежится по каждому действию и займет это доли секунды. В нашем же коде мы выполняем каждое действие, по наступлению определенного события, а именно завершении предыдущего FastAGI-приложения. Lunapark заботливо генерирует специальный UserEvent по окончании выполнения каждого FastAGI-приложения с соответствующими параметрами — вот на это событие и ориентируемся. Сами же сопрограммы просто редиректят текущий канал в контекст default с экстеншном, равным текущему действию, предварительно установив переменную канала PRMS.
Самое интересное, что звонок после redirect'а придет опять в handler, но уже в контексте выполнения AGI и на соответствующее приложение. В нашем случае это Hangup() и Dial(). Давайте же напишем их для полноты повествования.
function Hangup(...)
local app, channel = ... -- В этом отличие от pbx_lua
app.verbose(('The Channel %s does not match by routing rules'):format(channel.get('CHANNEL')))
app.hangup()
end
function Dial(...)
local app, channel = ...
local leg = app.agi.params['callee'] or ''
app.verbose(('Trying to make a call from %s to %s'):format(
channel.get('CALLERID(num)'),
leg:match('^[^/]+/([^%-]+)'))
)
app.dial(leg)
endНу, вот и все — допрыгались
Итак, давайте подытожим. Что же мы получили в результате этих генетических экспериментов?
- гибкий, функциональный подход описания маршрутов;
- возможность создания полнофункциональных VoIP-приложений. Теперь нам не нужно приложение Queue, мы можем его сами написать, не заморачиваясь с созданием своего собственного модуля для asterisk'а;
- вынесли логику формирования, управления звонками на сторону сервера VoIP-приложений, тем самым сделав из asterisk'а Mediahub, что позволило повысить производительность VoIP-системы в целом;
- возможность использовать достаточно простой, расширяемый и очень гибкий скриптовый язык для создания VoIP-приложений;
- расширили возможности интеграции с внешними системами из VoIP-приложений.
Кому как, а мне пока все нравится.
local fmt = function(str, tab)
return (str:gsub('(%${[^}{]+})', function(w)
local mark = w:sub(3, -2)
return (mark:gsub('(.+)',function(v)
local out = tab[v] or v
return ("('%s')"):format(out)
end))
end))
end
local syntax = function(str)
return (str:gsub('([^;]+)=>([^;]+)',function(p,r)
return ([[
if %s then
%s
end
]]):format(p,r)
end))
end
local routes = function(...)
local conf, content = ...
local f, err = io.open(conf, "r")
if io.type(f) ~= 'file' then
log.warn(err) -- Глобальный LOG объект доступный благодаря Lunapark'у
return ""
else
content = f:read('*all')
end
f:close() return content
end
-- Считываем наши правила в переменную rules
local rules = routes('routes.conf')
-- Очищаем все обработчики, причем таким образом очистятся только неименные обработчики событий
-- Это даст возможность не затирать цепочку выполнения при сигналах HUP/QUIT
ami.removeEvents('*')
-- Обработчик события создания нового канала
ami.addEvents {
['newchannel'] = function(e)
-- Условия, только каналы с набором и каналы с контекстом users
if (e['Context'] and e['Context']:match('users')) and e['Exten'] then
local step -- Переменная, указывающая на выполнении, какого FastAGI приложения мы находимся
local count = 0 -- Будущий порядковый номер FatsAGI приложения в цепочке выполнения
-- Парсим маршруты для текущего окружения канала
local code, err = loadstring(syntax(fmt(rules,e)))
-- В описании маршрутов нет ошибок, двигаемся дальше
if type(code) == 'function' then
-- Проксируем будущие FastAGI приложения
setfenv(code,setmetatable({indexes = {}},{__index = function(t,k)
-- Вот они последствия кооперативности
return coroutine.wrap(
function(...)
local prms = {} -- Будущие параметры FastAGI приложения
local owner = t -- Копия окружения
local event = e -- Копия таблицы event
local thread = coroutine.running() -- ID текущей сопрограммы
-- Парсим параметры и приводим к виду URI
for p,v in pairs({...}) do
if type(v) == 'table' then
for key, val in pairs(v) do
table.insert(prms,("%s=%s"):format(key,val))
end
else
table.insert(prms,("%s=%s"):format(p,v))
end
end
-- Если это не первое FastAGI приложение в цепочке
if step then
-- Запоминаем предыдущее перед этим
local last = ("%s"):format(step)
-- Добавляем ИМЕННЫЕ обработчики события UserEvent по доп. условиям
-- И записываем в таблицу indexes(в окружении) их порядковые номера
-- Именные обработчики требуют последующего удаления самостоятельно
table.insert(owner['indexes'],ami.addEvent('UserEvent',function(evt)
-- Ловим событие AGIStatus указывающее на завершение приложения
-- Если это предыдущее перед нами, пробуждаем сопрограмму
if (evt['Channel'] and evt['Channel'] == event['Channel'])
and
(evt['UserEvent'] and evt['UserEvent']:match('AGIStatus'))
and
(evt['Script'] and evt['Script'] == last)
then
-- Соответствие порядкового номера нашей сопрограмме
-- В цепочке может быть вызов одного приложения несколько раз
-- Это позволит выполнять сопрограммы в порядке их определения
if owner['indexes'][count] == thread then
if coroutine.status(thread) ~= 'dead' then
coroutine.resume(thread)
end
end
end
end,thread))
-- Устанавливаем маркер текущего FastAGI приложения
step = k
-- Приостанавливаем сопрограмму
coroutine.yield()
else -- Здесь обрабатывается первое FastAGI приложение в цепочке
local index -- Индекс для обработчика Hangup события
-- Устанавливаем маркер текущего FastAGI приложения
step = k
-- Добавляем ИМЕННОЙ обработчик события Hangup для канала
-- В этом месте подчищаем за собой
index = ami.addEvent('Hangup',function(evt)
if evt['Channel'] and evt['Channel'] == event['Channel'] then
-- Удаляем обработчик событие Hangup по ранее запомненному индексу
ami.removeEvent('Hangup',index)
-- Удаляем все обработчики цепочек выполнения по индексу
for _,v in pairs(owner['indexes']) do
ami.removeEvent('UserEvent',v)
end
-- Делаем приятно сборщику мусора
owner = nil
end
end,thread)
end
-- По средствам AMI выставляем переменную для канала и вызова в цепочке
ami.setvar{
Value = table.concat(prms,'&'),
Channel = event['Channel'],
Variable = 'PRMS'
}
-- Перенаправляем канал на AGI-приложение через контекст default
ami.redirect{
Exten = k,
Priority = 1,
Channel = event['Channel'],
Context = 'default'
}
-- Выставляем индекс приложения
count = count + 1
end)
end}))()
else
-- Если что-то пошло не так
log.warn(err)
end
end
end
}
function Hangup(...)
local app, channel = ... -- В этом отличие от pbx_lua
app.verbose(('The Channel %s does not match by routing rules'):format(channel.get('CHANNEL')))
app.hangup()
end
function Dial(...)
local app, channel = ...
local leg = app.agi.params['callee'] or ''
app.verbose(('Trying to make a call from %s to %s'):format(
channel.get('CALLERID(num)'),
leg:match('^[^/]+/([^%-]+)'))
)
app.dial(leg)
end