ben.carrasco/freeswitch-modules
1
0
Fork 0
mirror of https://github.com/jambonz/freeswitch-modules.git synced 2026-01-25 02:08:27 +00:00
Code Issues Packages Projects Releases Wiki Activity

Fix/audio pipe (#4)

Browse Source 
* use explicit namespaces for mod_audio_fork

* fix crash in reload scenarios

Signed-off-by: Dave Horton <daveh@beachdognet.com>

---------

Signed-off-by: Dave Horton <daveh@beachdognet.com>
This commit is contained in:
Dave Horton
2023-12-27 11:10:57 -05:00
committed by GitHub
parent 3f50aaf0d7
commit 9fc8b1af97
17 changed files with 239 additions and 202 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
mod_audio_fork/audio_pipe.cpp
View File

@@ -7,6 +7,7 @@
#define MAX_RECV_BUF_SIZE (65 * 1024 * 10)
#define RECV_BUF_REALLOC_SIZE (8 * 1024)
using namespace drachtio;
namespace {
static const char* basicAuthUser = std::getenv("MOD_AUDIO_FORK_HTTP_AUTH_USER");
@@ -267,6 +268,7 @@ static const lws_retry_bo_t retry = {
};
struct lws_context *AudioPipe::context = nullptr;
std::thread AudioPipe::serviceThread;
std::string AudioPipe::protocolName;
std::mutex AudioPipe::mutex_connects;
std::mutex AudioPipe::mutex_disconnects;
@@ -436,19 +438,21 @@ bool AudioPipe::lws_service_thread() {
void AudioPipe::initialize(const char* protocol, int loglevel, log_emit_function logger) {
protocolName = protocol;
lws_set_log_level(loglevel, logger);
//lws_set_log_level(loglevel, logger);
lwsl_notice("AudioPipe::initialize starting\n");
std::lock_guard<std::mutex> lock(mapMutex);
std::thread t(&AudioPipe::lws_service_thread);
stopFlag = false;
t.detach();
serviceThread = std::thread(&AudioPipe::lws_service_thread);
}
bool AudioPipe::deinitialize() {
lwsl_notice("AudioPipe::deinitialize\n");
std::lock_guard<std::mutex> lock(mapMutex);
stopFlag = true;
if (serviceThread.joinable()) {
serviceThread.join();
}
return true;
}

259
mod_audio_fork/audio_pipe.hpp
View File

@@ -10,135 +10,140 @@
#include <libwebsockets.h>
class AudioPipe {
public:
enum LwsState_t {
LWS_CLIENT_IDLE,
LWS_CLIENT_CONNECTING,
LWS_CLIENT_CONNECTED,
LWS_CLIENT_FAILED,
LWS_CLIENT_DISCONNECTING,
LWS_CLIENT_DISCONNECTED
};
enum NotifyEvent_t {
CONNECT_SUCCESS,
CONNECT_FAIL,
CONNECTION_DROPPED,
CONNECTION_CLOSED_GRACEFULLY,
MESSAGE
};
typedef void (*log_emit_function)(int level, const char *line);
typedef void (*notifyHandler_t)(const char *sessionId, const char* bugname, NotifyEvent_t event, const char* message);
namespace drachtio {
struct lws_per_vhost_data {
struct lws_context *context;
struct lws_vhost *vhost;
const struct lws_protocols *protocol;
class AudioPipe {
public:
enum LwsState_t {
LWS_CLIENT_IDLE,
LWS_CLIENT_CONNECTING,
LWS_CLIENT_CONNECTED,
LWS_CLIENT_FAILED,
LWS_CLIENT_DISCONNECTING,
LWS_CLIENT_DISCONNECTED
};
enum NotifyEvent_t {
CONNECT_SUCCESS,
CONNECT_FAIL,
CONNECTION_DROPPED,
CONNECTION_CLOSED_GRACEFULLY,
MESSAGE
};
typedef void (*log_emit_function)(int level, const char *line);
typedef void (*notifyHandler_t)(const char *sessionId, const char* bugname, NotifyEvent_t event, const char* message);
struct lws_per_vhost_data {
struct lws_context *context;
struct lws_vhost *vhost;
const struct lws_protocols *protocol;
};
static void initialize(const char* protocolName, int loglevel, log_emit_function logger);
static bool deinitialize();
static bool lws_service_thread();
// constructor
AudioPipe(const char* uuid, const char* host, unsigned int port, const char* path, int sslFlags,
size_t bufLen, size_t minFreespace, const char* username, const char* password, char* bugname, notifyHandler_t callback);
~AudioPipe();
LwsState_t getLwsState(void) { return m_state; }
void connect(void);
void bufferForSending(const char* text);
size_t binarySpaceAvailable(void) {
return m_audio_buffer_max_len - m_audio_buffer_write_offset;
}
size_t binaryMinSpace(void) {
return m_audio_buffer_min_freespace;
}
char * binaryWritePtr(void) {
return (char *) m_audio_buffer + m_audio_buffer_write_offset;
}
void binaryWritePtrAdd(size_t len) {
m_audio_buffer_write_offset += len;
}
void binaryWritePtrResetToZero(void) {
m_audio_buffer_write_offset = 0;
}
void lockAudioBuffer(void) {
m_audio_mutex.lock();
}
void unlockAudioBuffer(void) ;
bool hasBasicAuth(void) {
return !m_username.empty() && !m_password.empty();
}
void getBasicAuth(std::string& username, std::string& password) {
username = m_username;
password = m_password;
}
void do_graceful_shutdown();
bool isGracefulShutdown(void) {
return m_gracefulShutdown;
}
void close() ;
// no default constructor or copying
AudioPipe() = delete;
AudioPipe(const AudioPipe&) = delete;
void operator=(const AudioPipe&) = delete;
private:
static std::thread serviceThread;
static int lws_callback(struct lws *wsi, enum lws_callback_reasons reason, void *user, void *in, size_t len);
static struct lws_context *context;
static std::string protocolName;
static std::mutex mutex_connects;
static std::mutex mutex_disconnects;
static std::mutex mutex_writes;
static std::list<AudioPipe*> pendingConnects;
static std::list<AudioPipe*> pendingDisconnects;
static std::list<AudioPipe*> pendingWrites;
static log_emit_function logger;
static std::mutex mapMutex;
static bool stopFlag;
static AudioPipe* findAndRemovePendingConnect(struct lws *wsi);
static AudioPipe* findPendingConnect(struct lws *wsi);
static void addPendingConnect(AudioPipe* ap);
static void addPendingDisconnect(AudioPipe* ap);
static void addPendingWrite(AudioPipe* ap);
static void processPendingConnects(lws_per_vhost_data *vhd);
static void processPendingDisconnects(lws_per_vhost_data *vhd);
static void processPendingWrites(void);
bool connect_client(struct lws_per_vhost_data *vhd);
LwsState_t m_state;
std::string m_uuid;
std::string m_host;
std::string m_bugname;
unsigned int m_port;
std::string m_path;
std::string m_metadata;
std::mutex m_text_mutex;
std::mutex m_audio_mutex;
int m_sslFlags;
struct lws *m_wsi;
uint8_t *m_audio_buffer;
size_t m_audio_buffer_max_len;
size_t m_audio_buffer_write_offset;
size_t m_audio_buffer_min_freespace;
uint8_t* m_recv_buf;
uint8_t* m_recv_buf_ptr;
size_t m_recv_buf_len;
struct lws_per_vhost_data* m_vhd;
notifyHandler_t m_callback;
log_emit_function m_logger;
std::string m_username;
std::string m_password;
bool m_gracefulShutdown;
};
static void initialize(const char* protocolName, int loglevel, log_emit_function logger);
static bool deinitialize();
static bool lws_service_thread();
// constructor
AudioPipe(const char* uuid, const char* host, unsigned int port, const char* path, int sslFlags,
size_t bufLen, size_t minFreespace, const char* username, const char* password, char* bugname, notifyHandler_t callback);
~AudioPipe();
LwsState_t getLwsState(void) { return m_state; }
void connect(void);
void bufferForSending(const char* text);
size_t binarySpaceAvailable(void) {
return m_audio_buffer_max_len - m_audio_buffer_write_offset;
}
size_t binaryMinSpace(void) {
return m_audio_buffer_min_freespace;
}
char * binaryWritePtr(void) {
return (char *) m_audio_buffer + m_audio_buffer_write_offset;
}
void binaryWritePtrAdd(size_t len) {
m_audio_buffer_write_offset += len;
}
void binaryWritePtrResetToZero(void) {
m_audio_buffer_write_offset = 0;
}
void lockAudioBuffer(void) {
m_audio_mutex.lock();
}
void unlockAudioBuffer(void) ;
bool hasBasicAuth(void) {
return !m_username.empty() && !m_password.empty();
}
void getBasicAuth(std::string& username, std::string& password) {
username = m_username;
password = m_password;
}
void do_graceful_shutdown();
bool isGracefulShutdown(void) {
return m_gracefulShutdown;
}
void close() ;
// no default constructor or copying
AudioPipe() = delete;
AudioPipe(const AudioPipe&) = delete;
void operator=(const AudioPipe&) = delete;
private:
static int lws_callback(struct lws *wsi, enum lws_callback_reasons reason, void *user, void *in, size_t len);
static struct lws_context *context;
static std::string protocolName;
static std::mutex mutex_connects;
static std::mutex mutex_disconnects;
static std::mutex mutex_writes;
static std::list<AudioPipe*> pendingConnects;
static std::list<AudioPipe*> pendingDisconnects;
static std::list<AudioPipe*> pendingWrites;
static log_emit_function logger;
static std::mutex mapMutex;
static bool stopFlag;
static AudioPipe* findAndRemovePendingConnect(struct lws *wsi);
static AudioPipe* findPendingConnect(struct lws *wsi);
static void addPendingConnect(AudioPipe* ap);
static void addPendingDisconnect(AudioPipe* ap);
static void addPendingWrite(AudioPipe* ap);
static void processPendingConnects(lws_per_vhost_data *vhd);
static void processPendingDisconnects(lws_per_vhost_data *vhd);
static void processPendingWrites(void);
bool connect_client(struct lws_per_vhost_data *vhd);
LwsState_t m_state;
std::string m_uuid;
std::string m_host;
std::string m_bugname;
unsigned int m_port;
std::string m_path;
std::string m_metadata;
std::mutex m_text_mutex;
std::mutex m_audio_mutex;
int m_sslFlags;
struct lws *m_wsi;
uint8_t *m_audio_buffer;
size_t m_audio_buffer_max_len;
size_t m_audio_buffer_write_offset;
size_t m_audio_buffer_min_freespace;
uint8_t* m_recv_buf;
uint8_t* m_recv_buf_ptr;
size_t m_recv_buf_len;
struct lws_per_vhost_data* m_vhd;
notifyHandler_t m_callback;
log_emit_function m_logger;
std::string m_username;
std::string m_password;
bool m_gracefulShutdown;
};
} // namespace drachtio
#endif

38
mod_audio_fork/lws_glue.cpp
View File

@@ -156,7 +156,7 @@ namespace {
}
}
static void eventCallback(const char* sessionId, const char* bugname, AudioPipe::NotifyEvent_t event, const char* message) {
static void eventCallback(const char* sessionId, const char* bugname, drachtio::AudioPipe::NotifyEvent_t event, const char* message) {
switch_core_session_t* session = switch_core_session_locate(sessionId);
if (session) {
switch_channel_t *channel = switch_core_session_get_channel(session);
@@ -165,16 +165,16 @@ namespace {
private_t* tech_pvt = (private_t*) switch_core_media_bug_get_user_data(bug);
if (tech_pvt) {
switch (event) {
case AudioPipe::CONNECT_SUCCESS:
case drachtio::AudioPipe::CONNECT_SUCCESS:
switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_INFO, "connection successful\n");
tech_pvt->responseHandler(session, EVENT_CONNECT_SUCCESS, NULL);
if (strlen(tech_pvt->initialMetadata) > 0) {
switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "sending initial metadata %s\n", tech_pvt->initialMetadata);
AudioPipe *pAudioPipe = static_cast<AudioPipe *>(tech_pvt->pAudioPipe);
drachtio::AudioPipe *pAudioPipe = static_cast<drachtio::AudioPipe *>(tech_pvt->pAudioPipe);
pAudioPipe->bufferForSending(tech_pvt->initialMetadata);
}
break;
case AudioPipe::CONNECT_FAIL:
case drachtio::AudioPipe::CONNECT_FAIL:
{
// first thing: we can no longer access the AudioPipe
std::stringstream json;
@@ -184,18 +184,18 @@ namespace {
switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_NOTICE, "connection failed: %s\n", message);
}
break;
case AudioPipe::CONNECTION_DROPPED:
case drachtio::AudioPipe::CONNECTION_DROPPED:
// first thing: we can no longer access the AudioPipe
tech_pvt->pAudioPipe = nullptr;
tech_pvt->responseHandler(session, EVENT_DISCONNECT, NULL);
switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_NOTICE, "connection dropped from far end\n");
break;
case AudioPipe::CONNECTION_CLOSED_GRACEFULLY:
case drachtio::AudioPipe::CONNECTION_CLOSED_GRACEFULLY:
// first thing: we can no longer access the AudioPipe
tech_pvt->pAudioPipe = nullptr;
switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "connection closed gracefully\n");
break;
case AudioPipe::MESSAGE:
case drachtio::AudioPipe::MESSAGE:
processIncomingMessage(tech_pvt, session, message);
break;
}
@@ -239,7 +239,7 @@ namespace {
size_t buflen = LWS_PRE + (FRAME_SIZE_8000 * desiredSampling / 8000 * channels * 1000 / RTP_PACKETIZATION_PERIOD * nAudioBufferSecs);
AudioPipe* ap = new AudioPipe(tech_pvt->sessionId, host, port, path, sslFlags,
drachtio::AudioPipe* ap = new drachtio::AudioPipe(tech_pvt->sessionId, host, port, path, sslFlags,
buflen, read_impl.decoded_bytes_per_packet, username, password, bugname, eventCallback);
if (!ap) {
switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_ERROR, "Error allocating AudioPipe\n");
@@ -374,13 +374,17 @@ extern "C" {
int logs = LLL_ERR | LLL_WARN | LLL_NOTICE ;
//LLL_INFO | LLL_PARSER | LLL_HEADER | LLL_EXT | LLL_CLIENT | LLL_LATENCY | LLL_DEBUG ;
AudioPipe::initialize(mySubProtocolName, logs, lws_logger);
return SWITCH_STATUS_SUCCESS;
drachtio::AudioPipe::initialize(mySubProtocolName, logs, lws_logger);
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_NOTICE, "mod_audio_fork successfully initialized\n");
return SWITCH_STATUS_SUCCESS;
}
switch_status_t fork_cleanup() {
bool cleanup = false;
cleanup = AudioPipe::deinitialize();
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_NOTICE, "mod_audio_fork unloading..\n");
cleanup = drachtio::AudioPipe::deinitialize();
switch_log_printf(SWITCH_CHANNEL_LOG, SWITCH_LOG_NOTICE, "mod_audio_fork unloaded status %d\n", cleanup);
if (cleanup == true) {
return SWITCH_STATUS_SUCCESS;
}
@@ -420,7 +424,7 @@ extern "C" {
switch_status_t fork_session_connect(void **ppUserData) {
private_t *tech_pvt = static_cast<private_t *>(*ppUserData);
AudioPipe *pAudioPipe = static_cast<AudioPipe*>(tech_pvt->pAudioPipe);
drachtio::AudioPipe *pAudioPipe = static_cast<drachtio::AudioPipe*>(tech_pvt->pAudioPipe);
pAudioPipe->connect();
return SWITCH_STATUS_SUCCESS;
}
@@ -438,7 +442,7 @@ extern "C" {
switch_log_printf(SWITCH_CHANNEL_SESSION_LOG(session), SWITCH_LOG_DEBUG, "(%u) fork_session_cleanup\n", id);
if (!tech_pvt) return SWITCH_STATUS_FALSE;
AudioPipe *pAudioPipe = static_cast<AudioPipe *>(tech_pvt->pAudioPipe);
drachtio::AudioPipe *pAudioPipe = static_cast<drachtio::AudioPipe *>(tech_pvt->pAudioPipe);
switch_mutex_lock(tech_pvt->mutex);
@@ -481,7 +485,7 @@ extern "C" {
private_t* tech_pvt = (private_t*) switch_core_media_bug_get_user_data(bug);
if (!tech_pvt) return SWITCH_STATUS_FALSE;
AudioPipe *pAudioPipe = static_cast<AudioPipe *>(tech_pvt->pAudioPipe);
drachtio::AudioPipe *pAudioPipe = static_cast<drachtio::AudioPipe *>(tech_pvt->pAudioPipe);
if (pAudioPipe && text) pAudioPipe->bufferForSending(text);
return SWITCH_STATUS_SUCCESS;
@@ -516,7 +520,7 @@ extern "C" {
tech_pvt->graceful_shutdown = 1;
AudioPipe *pAudioPipe = static_cast<AudioPipe *>(tech_pvt->pAudioPipe);
drachtio::AudioPipe *pAudioPipe = static_cast<drachtio::AudioPipe *>(tech_pvt->pAudioPipe);
if (pAudioPipe) pAudioPipe->do_graceful_shutdown();
return SWITCH_STATUS_SUCCESS;
@@ -535,8 +539,8 @@ extern "C" {
switch_mutex_unlock(tech_pvt->mutex);
return SWITCH_TRUE;
}
AudioPipe *pAudioPipe = static_cast<AudioPipe *>(tech_pvt->pAudioPipe);
if (pAudioPipe->getLwsState() != AudioPipe::LWS_CLIENT_CONNECTED) {
drachtio::AudioPipe *pAudioPipe = static_cast<drachtio::AudioPipe *>(tech_pvt->pAudioPipe);
if (pAudioPipe->getLwsState() != drachtio::AudioPipe::LWS_CLIENT_CONNECTED) {
switch_mutex_unlock(tech_pvt->mutex);
return SWITCH_TRUE;
}