gfwleak/stellar-stellar
Archived
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feature: session manager support subscribe session free message

Browse Source
This commit is contained in:
luwenpeng
2024-10-23 14:19:06 +08:00
parent 9987189957
commit 3d4e6a2cd5
3 changed files with 113 additions and 71 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
include/stellar/session_manager.h
View File

@@ -18,12 +18,14 @@ struct session_manager *stellar_module_get_session_manager(struct stellar_module
int session_manager_new_session_exdata_index(struct session_manager *sess_mgr, const char *name, exdata_free *func, void *arg); int session_manager_new_session_exdata_index(struct session_manager *sess_mgr, const char *name, exdata_free *func, void *arg);
typedef void on_session_callback(struct session *sess, struct packet *pkt, void *args); typedef void on_session_free_callback(struct session *sess, void *args);
typedef void on_session_packet_callback(struct session *sess, struct packet *pkt, void *args);
typedef void on_tcp_stream_callback(struct session *sess, const char *tcp_payload, uint32_t tcp_payload_len, void *args); typedef void on_tcp_stream_callback(struct session *sess, const char *tcp_payload, uint32_t tcp_payload_len, void *args);
int session_manager_subscribe_tcp(struct session_manager *sess_mgr, on_session_callback *cb, void *args); int session_manager_subscribe_free(struct session_manager *sess_mgr, on_session_free_callback *cb, void *args);
int session_manager_subscribe_udp(struct session_manager *sess_mgr, on_session_callback *cb, void *args); int session_manager_subscribe_tcp(struct session_manager *sess_mgr, on_session_packet_callback *cb, void *args);
int session_manager_subscribe_control_packet(struct session_manager *sess_mgr, on_session_callback *cb, void *args); int session_manager_subscribe_udp(struct session_manager *sess_mgr, on_session_packet_callback *cb, void *args);
int session_manager_subscribe_control_packet(struct session_manager *sess_mgr, on_session_packet_callback *cb, void *args);
int session_manager_subscribe_tcp_stream(struct session_manager *sess_mgr, on_tcp_stream_callback *cb, void *args); int session_manager_subscribe_tcp_stream(struct session_manager *sess_mgr, on_tcp_stream_callback *cb, void *args);
#ifdef __cplusplus #ifdef __cplusplus

111
infra/session_manager/session_manager.c
View File

@@ -10,6 +10,7 @@
#include "session_internal.h" #include "session_internal.h"
#include "session_manager_runtime.h" #include "session_manager_runtime.h"
#define MAX_CLEANED_SESS 1024
#define SESSION_MANAGER_MODULE_NAME "session_manager_module" #define SESSION_MANAGER_MODULE_NAME "session_manager_module"
#pragma GCC diagnostic ignored "-Wunused-parameter" #pragma GCC diagnostic ignored "-Wunused-parameter"
@@ -22,6 +23,7 @@ struct session_manager_schema
int pkt_exdata_idx; int pkt_exdata_idx;
int topic_id_free;
int topic_id_tcp; int topic_id_tcp;
int topic_id_udp; int topic_id_udp;
int topic_id_ctrl_pkt; int topic_id_ctrl_pkt;
@@ -40,12 +42,18 @@ struct session_manager
* callback * callback
******************************************************************************/ ******************************************************************************/
static void on_session_dispatch(int topic_id, void *msg, on_msg_cb_func *cb, void *cb_args, void *dispatch_args) static void on_session_free_dispatch(int topic_id, void *msg, on_msg_cb_func *cb, void *cb_args, void *dispatch_args)
{
struct session *sess = (struct session *)msg;
((on_session_free_callback *)(void *)cb)(sess, cb_args);
}
static void on_session_packet_dispatch(int topic_id, void *msg, on_msg_cb_func *cb, void *cb_args, void *dispatch_args)
{ {
struct session *sess = (struct session *)msg; struct session *sess = (struct session *)msg;
struct packet *pkt = (struct packet *)session_get0_current_packet(sess); struct packet *pkt = (struct packet *)session_get0_current_packet(sess);
((on_session_callback *)(void *)cb)(sess, pkt, cb_args); ((on_session_packet_callback *)(void *)cb)(sess, pkt, cb_args);
} }
static void on_tcp_stream_dispatch(int topic_id, void *msg, on_msg_cb_func *cb, void *cb_args, void *dispatch_args) static void on_tcp_stream_dispatch(int topic_id, void *msg, on_msg_cb_func *cb, void *cb_args, void *dispatch_args)
@@ -63,6 +71,23 @@ static void on_tcp_stream_free(void *msg, void *args)
session_free_tcp_segment(sess, seg); session_free_tcp_segment(sess, seg);
} }
static void on_session_free(void *msg, void *args)
{
struct session *sess = (struct session *)msg;
struct session_manager *sess_mgr = (struct session_manager *)args;
struct stellar_module_manager *mod_mgr = sess_mgr->mod_mgr;
int thread_id = stellar_module_manager_get_thread_id(mod_mgr);
struct session_manager_runtime *sess_mgr_rt = sess_mgr->runtime[thread_id];
char buffer[4096] = {0};
session_to_str(sess, 0, buffer, sizeof(buffer));
SESSION_MANAGER_LOG_INFO("session free: %s", buffer);
struct exdata_runtime *exdata_rt = (struct exdata_runtime *)session_get_user_data(sess);
exdata_runtime_free(exdata_rt);
session_manager_runtime_free_session(sess_mgr_rt, sess);
}
static void on_packet_forward(enum packet_stage stage, struct packet *pkt, void *args) static void on_packet_forward(enum packet_stage stage, struct packet *pkt, void *args)
{ {
struct session_manager *sess_mgr = (struct session_manager *)args; struct session_manager *sess_mgr = (struct session_manager *)args;
@@ -166,39 +191,20 @@ static void on_packet_output(enum packet_stage stage, struct packet *pkt, void *
} }
} }
static void clean_session(struct session_manager_runtime *sess_mgr_rt, uint64_t now_ms)
{
#define MAX_CLEANED_SESS 1024
char buffer[4096] = {0};
struct session *sess = NULL;
struct session *cleaned_sess[MAX_CLEANED_SESS] = {NULL};
uint64_t used = session_manager_runtime_clean_session(sess_mgr_rt, now_ms, cleaned_sess, MAX_CLEANED_SESS);
for (uint64_t j = 0; j < used; j++)
{
sess = cleaned_sess[j];
session_to_str(sess, 0, buffer, sizeof(buffer));
SESSION_MANAGER_LOG_INFO("session free: %s", buffer);
// TODO publish session free msg
// TODO mq_runtime_dispatch_immediate()
struct exdata_runtime *exdata_rt = (struct exdata_runtime *)session_get_user_data(sess);
exdata_runtime_free(exdata_rt);
session_manager_runtime_free_session(sess_mgr_rt, sess);
}
}
static void on_polling(struct stellar_module_manager *mod_mgr, void *args) static void on_polling(struct stellar_module_manager *mod_mgr, void *args)
{ {
uint64_t now_ms = clock_get_real_time_ms();
struct session *cleaned[MAX_CLEANED_SESS] = {NULL};
struct session_manager *sess_mgr = (struct session_manager *)args; struct session_manager *sess_mgr = (struct session_manager *)args;
int thread_id = stellar_module_manager_get_thread_id(mod_mgr); int thread_id = stellar_module_manager_get_thread_id(mod_mgr);
struct mq_runtime *mq_rt = stellar_module_manager_get_mq_runtime(mod_mgr);
struct session_manager_runtime *sess_mgr_rt = sess_mgr->runtime[thread_id]; struct session_manager_runtime *sess_mgr_rt = sess_mgr->runtime[thread_id];
uint64_t now_ms = clock_get_real_time_ms();
clean_session(sess_mgr_rt, now_ms); uint64_t used = session_manager_runtime_clean_session(sess_mgr_rt, now_ms, cleaned, MAX_CLEANED_SESS);
for (uint64_t i = 0; i < used; i++)
{
mq_runtime_publish_message(mq_rt, sess_mgr->schema->topic_id_free, cleaned[i]);
}
// TODO // TODO
// ouput stat to fs4 // ouput stat to fs4
@@ -215,6 +221,7 @@ void session_manager_schema_free(struct session_manager_schema *sess_mgr_schema)
{ {
if (sess_mgr_schema->mq) if (sess_mgr_schema->mq)
{ {
mq_schema_destroy_topic(sess_mgr_schema->mq, sess_mgr_schema->topic_id_free);
mq_schema_destroy_topic(sess_mgr_schema->mq, sess_mgr_schema->topic_id_tcp); mq_schema_destroy_topic(sess_mgr_schema->mq, sess_mgr_schema->topic_id_tcp);
mq_schema_destroy_topic(sess_mgr_schema->mq, sess_mgr_schema->topic_id_udp); mq_schema_destroy_topic(sess_mgr_schema->mq, sess_mgr_schema->topic_id_udp);
mq_schema_destroy_topic(sess_mgr_schema->mq, sess_mgr_schema->topic_id_ctrl_pkt); mq_schema_destroy_topic(sess_mgr_schema->mq, sess_mgr_schema->topic_id_ctrl_pkt);
@@ -262,28 +269,34 @@ struct session_manager_schema *session_manager_schema_new(struct packet_manager
goto error_out; goto error_out;
} }
sess_mgr_schema->topic_id_tcp = mq_schema_create_topic(sess_mgr_schema->mq, "TCP", &on_session_dispatch, NULL, NULL, NULL); sess_mgr_schema->topic_id_free = mq_schema_create_topic(sess_mgr_schema->mq, "SESSIOM_MANAGER_TOPIC_FREE", &on_session_free_dispatch, NULL, &on_session_free, subscribe_args);
if (sess_mgr_schema->topic_id_free == -1)
{
SESSION_MANAGER_LOG_ERROR("failed to create topic SESSIOM_MANAGER_TOPIC_FREE");
goto error_out;
}
sess_mgr_schema->topic_id_tcp = mq_schema_create_topic(sess_mgr_schema->mq, "SESSIOM_MANAGER_TOPIC_TCP", &on_session_packet_dispatch, NULL, NULL, NULL);
if (sess_mgr_schema->topic_id_tcp == -1) if (sess_mgr_schema->topic_id_tcp == -1)
{ {
SESSION_MANAGER_LOG_ERROR("failed to create topic TCP"); SESSION_MANAGER_LOG_ERROR("failed to create topic SESSIOM_MANAGER_TOPIC_FREE");
goto error_out; goto error_out;
} }
sess_mgr_schema->topic_id_udp = mq_schema_create_topic(sess_mgr_schema->mq, "UDP", &on_session_dispatch, NULL, NULL, NULL); sess_mgr_schema->topic_id_udp = mq_schema_create_topic(sess_mgr_schema->mq, "SESSIOM_MANAGER_TOPIC_UDP", &on_session_packet_dispatch, NULL, NULL, NULL);
if (sess_mgr_schema->topic_id_udp == -1) if (sess_mgr_schema->topic_id_udp == -1)
{ {
SESSION_MANAGER_LOG_ERROR("failed to create topic UDP"); SESSION_MANAGER_LOG_ERROR("failed to create topic SESSIOM_MANAGER_TOPIC_UDP");
goto error_out; goto error_out;
} }
sess_mgr_schema->topic_id_ctrl_pkt = mq_schema_create_topic(sess_mgr_schema->mq, "CTRL_PKT", &on_session_dispatch, NULL, NULL, NULL); sess_mgr_schema->topic_id_ctrl_pkt = mq_schema_create_topic(sess_mgr_schema->mq, "SESSIOM_MANAGER_TOPIC_CTRL_PKT", &on_session_packet_dispatch, NULL, NULL, NULL);
if (sess_mgr_schema->topic_id_ctrl_pkt == -1) if (sess_mgr_schema->topic_id_ctrl_pkt == -1)
{ {
SESSION_MANAGER_LOG_ERROR("failed to create topic CTRL_PKT"); SESSION_MANAGER_LOG_ERROR("failed to create topic SESSIOM_MANAGER_TOPIC_CTRL_PKT");
goto error_out; goto error_out;
} }
sess_mgr_schema->topic_id_tcp_stream = mq_schema_create_topic(sess_mgr_schema->mq, "TCP_STREAM", &on_tcp_stream_dispatch, NULL, &on_tcp_stream_free, NULL); sess_mgr_schema->topic_id_tcp_stream = mq_schema_create_topic(sess_mgr_schema->mq, "SESSIOM_MANAGER_TOPIC_TCP_STREAM", &on_tcp_stream_dispatch, NULL, &on_tcp_stream_free, NULL);
if (sess_mgr_schema->topic_id_tcp_stream == -1) if (sess_mgr_schema->topic_id_tcp_stream == -1)
{ {
SESSION_MANAGER_LOG_ERROR("failed to create topic TCP_STREAM"); SESSION_MANAGER_LOG_ERROR("failed to create topic SESSIOM_MANAGER_TOPIC_TCP_STREAM");
goto error_out; goto error_out;
} }
@@ -348,7 +361,15 @@ int session_manager_new_session_exdata_index(struct session_manager *sess_mgr, c
return exdata_schema_new_index(sess_mgr->schema->exdata, name, func, arg); return exdata_schema_new_index(sess_mgr->schema->exdata, name, func, arg);
} }
int session_manager_subscribe_tcp(struct session_manager *sess_mgr, on_session_callback *cb, void *args) int session_manager_subscribe_free(struct session_manager *sess_mgr, on_session_free_callback *cb, void *args)
{
assert(sess_mgr);
assert(cb);
return mq_schema_subscribe(sess_mgr->schema->mq, sess_mgr->schema->topic_id_free, (on_msg_cb_func *)(void *)cb, args);
}
int session_manager_subscribe_tcp(struct session_manager *sess_mgr, on_session_packet_callback *cb, void *args)
{ {
assert(sess_mgr); assert(sess_mgr);
assert(cb); assert(cb);
@@ -356,7 +377,7 @@ int session_manager_subscribe_tcp(struct session_manager *sess_mgr, on_session_c
return mq_schema_subscribe(sess_mgr->schema->mq, sess_mgr->schema->topic_id_tcp, (on_msg_cb_func *)(void *)cb, args); return mq_schema_subscribe(sess_mgr->schema->mq, sess_mgr->schema->topic_id_tcp, (on_msg_cb_func *)(void *)cb, args);
} }
int session_manager_subscribe_udp(struct session_manager *sess_mgr, on_session_callback *cb, void *args) int session_manager_subscribe_udp(struct session_manager *sess_mgr, on_session_packet_callback *cb, void *args)
{ {
assert(sess_mgr); assert(sess_mgr);
assert(cb); assert(cb);
@@ -364,7 +385,7 @@ int session_manager_subscribe_udp(struct session_manager *sess_mgr, on_session_c
return mq_schema_subscribe(sess_mgr->schema->mq, sess_mgr->schema->topic_id_udp, (on_msg_cb_func *)(void *)cb, args); return mq_schema_subscribe(sess_mgr->schema->mq, sess_mgr->schema->topic_id_udp, (on_msg_cb_func *)(void *)cb, args);
} }
int session_manager_subscribe_control_packet(struct session_manager *sess_mgr, on_session_callback *cb, void *args) int session_manager_subscribe_control_packet(struct session_manager *sess_mgr, on_session_packet_callback *cb, void *args)
{ {
assert(sess_mgr); assert(sess_mgr);
assert(cb); assert(cb);
@@ -405,6 +426,8 @@ void session_manager_clean(struct session_manager *sess_mgr, uint16_t thread_id)
assert(sess_mgr); assert(sess_mgr);
assert(thread_id < sess_mgr->cfg->thread_num); assert(thread_id < sess_mgr->cfg->thread_num);
struct stellar_module_manager *mod_mgr = sess_mgr->mod_mgr;
struct mq_runtime *mq_rt = stellar_module_manager_get_mq_runtime(mod_mgr);
struct session_manager_runtime *sess_mgr_rt = sess_mgr->runtime[thread_id]; struct session_manager_runtime *sess_mgr_rt = sess_mgr->runtime[thread_id];
if (sess_mgr_rt == NULL) if (sess_mgr_rt == NULL)
{ {
@@ -414,7 +437,13 @@ void session_manager_clean(struct session_manager *sess_mgr, uint16_t thread_id)
struct session_manager_stat *stat = session_manager_runtime_get_stat(sess_mgr_rt); struct session_manager_stat *stat = session_manager_runtime_get_stat(sess_mgr_rt);
while (stat->tcp_sess_used || stat->udp_sess_used) while (stat->tcp_sess_used || stat->udp_sess_used)
{ {
clean_session(sess_mgr_rt, UINT64_MAX); struct session *cleaned[MAX_CLEANED_SESS] = {NULL};
uint64_t used = session_manager_runtime_clean_session(sess_mgr_rt, UINT64_MAX, cleaned, MAX_CLEANED_SESS);
for (uint64_t i = 0; i < used; i++)
{
mq_runtime_publish_message(mq_rt, sess_mgr->schema->topic_id_free, cleaned[i]);
mq_runtime_dispatch(mq_rt);
}
} }
SESSION_MANAGER_LOG_INFO("runtime: %p, idx: %d, will be cleaned", sess_mgr_rt, thread_id); SESSION_MANAGER_LOG_INFO("runtime: %p, idx: %d, will be cleaned", sess_mgr_rt, thread_id);

63
test/session_debugger/session_debugger.c
View File

@@ -56,7 +56,7 @@ static void session_debugger_log(int fd, const char *fmt, ...)
char buf[PATH_MAX * 2] = {0}; char buf[PATH_MAX * 2] = {0};
char *p = buf; char *p = buf;
char *end = buf + sizeof(buf); char *end = buf + sizeof(buf);
va_list args; va_list arg;
struct tm local; struct tm local;
time_t t; time_t t;
@@ -67,9 +67,9 @@ static void session_debugger_log(int fd, const char *fmt, ...)
weekday_str[local.tm_wday], month_str[local.tm_mon], local.tm_mday, local.tm_hour, local.tm_min, local.tm_sec, local.tm_year + 1900); weekday_str[local.tm_wday], month_str[local.tm_mon], local.tm_mday, local.tm_hour, local.tm_min, local.tm_sec, local.tm_year + 1900);
p += snprintf(p, end - p, "%lu ", pthread_self()); p += snprintf(p, end - p, "%lu ", pthread_self());
p += snprintf(p, end - p, "(%s), ", module); p += snprintf(p, end - p, "(%s), ", module);
va_start(args, fmt); va_start(arg, fmt);
p += vsnprintf(p, end - p, fmt, args); p += vsnprintf(p, end - p, fmt, arg);
va_end(args); va_end(arg);
p += snprintf(p, end - p, "\n"); p += snprintf(p, end - p, "\n");
do do
@@ -107,20 +107,6 @@ static void session_debugger_exdata_free(struct session_debugger_exdata *exdata)
{ {
if (exdata) if (exdata)
{ {
char buff[PATH_MAX] = {0};
session_to_str(exdata->sess, 0, buff, sizeof(buff) - 1);
session_debugger_log(exdata->dbg->fd, "sess free: %s", buff);
session_debugger_log(exdata->dbg->fd, "session %lu %s stat:\n"
"C2S rx packets: %6lu, C2S rx bytes: %6lu\n"
"S2C rx packets: %6lu, S2C rx bytes: %6lu\n"
"C2S rx TCP segments: %6lu, C2S rx TCP bytes: %6lu\n"
"S2C rx TCP segments: %6lu, S2C rx TCP bytes: %6lu\n",
session_get_id(exdata->sess), session_get0_readable_addr(exdata->sess),
exdata->c2s_rx_pkts, exdata->c2s_rx_bytes,
exdata->s2c_rx_pkts, exdata->s2c_rx_bytes,
exdata->c2s_rx_tcp_seg, exdata->c2s_rx_tcp_bytes,
exdata->s2c_rx_tcp_seg, exdata->s2c_rx_tcp_bytes);
if (exdata->c2s_tcp_seg_hexdump_fd > 0) if (exdata->c2s_tcp_seg_hexdump_fd > 0)
{ {
close(exdata->c2s_tcp_seg_hexdump_fd); close(exdata->c2s_tcp_seg_hexdump_fd);
@@ -142,9 +128,29 @@ static void session_debugger_exdata_free_callback(int idx, void *ex_ptr, void *a
session_debugger_exdata_free((struct session_debugger_exdata *)ex_ptr); session_debugger_exdata_free((struct session_debugger_exdata *)ex_ptr);
} }
static void on_sess_msg(struct session *sess, struct packet *pkt, void *args) static void on_sess_free(struct session *sess, void *arg)
{ {
struct session_debugger *dbg = (struct session_debugger *)args; struct session_debugger *dbg = (struct session_debugger *)arg;
struct session_debugger_exdata *exdata = (struct session_debugger_exdata *)session_get_exdata(sess, dbg->sess_exdata_idx);
char buff[PATH_MAX] = {0};
session_to_str(exdata->sess, 0, buff, sizeof(buff) - 1);
session_debugger_log(exdata->dbg->fd, "sess free: %s", buff);
session_debugger_log(exdata->dbg->fd, "session %lu %s stat:\n"
"C2S rx packets: %6lu, C2S rx bytes: %6lu\n"
"S2C rx packets: %6lu, S2C rx bytes: %6lu\n"
"C2S rx TCP segments: %6lu, C2S rx TCP bytes: %6lu\n"
"S2C rx TCP segments: %6lu, S2C rx TCP bytes: %6lu\n",
session_get_id(exdata->sess), session_get0_readable_addr(exdata->sess),
exdata->c2s_rx_pkts, exdata->c2s_rx_bytes,
exdata->s2c_rx_pkts, exdata->s2c_rx_bytes,
exdata->c2s_rx_tcp_seg, exdata->c2s_rx_tcp_bytes,
exdata->s2c_rx_tcp_seg, exdata->s2c_rx_tcp_bytes);
}
static void on_sess_packet(struct session *sess, struct packet *pkt, void *arg)
{
struct session_debugger *dbg = (struct session_debugger *)arg;
char buff[PATH_MAX]; char buff[PATH_MAX];
struct session_debugger_exdata *exdata = (struct session_debugger_exdata *)session_get_exdata(sess, dbg->sess_exdata_idx); struct session_debugger_exdata *exdata = (struct session_debugger_exdata *)session_get_exdata(sess, dbg->sess_exdata_idx);
@@ -178,9 +184,9 @@ static void on_sess_msg(struct session *sess, struct packet *pkt, void *args)
pthread_spin_unlock(&dbg->lock); pthread_spin_unlock(&dbg->lock);
} }
static void on_tcp_stream_msg(struct session *sess, const char *tcp_payload, uint32_t tcp_payload_len, void *args) static void on_tcp_stream(struct session *sess, const char *tcp_payload, uint32_t tcp_payload_len, void *arg)
{ {
struct session_debugger *dbg = (struct session_debugger *)args; struct session_debugger *dbg = (struct session_debugger *)arg;
char buff[PATH_MAX]; char buff[PATH_MAX];
struct session_debugger_exdata *exdata = (struct session_debugger_exdata *)session_get_exdata(sess, dbg->sess_exdata_idx); struct session_debugger_exdata *exdata = (struct session_debugger_exdata *)session_get_exdata(sess, dbg->sess_exdata_idx);
@@ -251,22 +257,27 @@ static struct session_debugger *session_debugger_new(struct session_manager *ses
goto error_out; goto error_out;
} }
if (session_manager_subscribe_tcp(sess_mgr, on_sess_msg, dbg) == -1) if (session_manager_subscribe_free(sess_mgr, on_sess_free, dbg) == -1)
{
session_debugger_log(STDERR_FILENO, "subscribe free failed\n");
goto error_out;
}
if (session_manager_subscribe_tcp(sess_mgr, on_sess_packet, dbg) == -1)
{ {
session_debugger_log(STDERR_FILENO, "subscribe tcp failed\n"); session_debugger_log(STDERR_FILENO, "subscribe tcp failed\n");
goto error_out; goto error_out;
} }
if (session_manager_subscribe_udp(sess_mgr, on_sess_msg, dbg) == -1) if (session_manager_subscribe_udp(sess_mgr, on_sess_packet, dbg) == -1)
{ {
session_debugger_log(STDERR_FILENO, "subscribe udp failed\n"); session_debugger_log(STDERR_FILENO, "subscribe udp failed\n");
goto error_out; goto error_out;
} }
if (session_manager_subscribe_control_packet(sess_mgr, on_sess_msg, dbg) == -1) if (session_manager_subscribe_control_packet(sess_mgr, on_sess_packet, dbg) == -1)
{ {
session_debugger_log(STDERR_FILENO, "subscribe control packet failed\n"); session_debugger_log(STDERR_FILENO, "subscribe control packet failed\n");
goto error_out; goto error_out;
} }
if (session_manager_subscribe_tcp_stream(sess_mgr, on_tcp_stream_msg, dbg) == -1) if (session_manager_subscribe_tcp_stream(sess_mgr, on_tcp_stream, dbg) == -1)
{ {
session_debugger_log(STDERR_FILENO, "subscribe tcp stream failed\n"); session_debugger_log(STDERR_FILENO, "subscribe tcp stream failed\n");
goto error_out; goto error_out;