init mail decoderdev-mail-decoder

9 files changed, 147 insertions, 187 deletions

diff --git a/decoders/mail/mail_decoder_email.c b/decoders/mail/mail_decoder_email.c
index d6963a4..ac8ff37 100644
--- a/decoders/mail/mail_decoder_email.c
+++ b/decoders/mail/mail_decoder_email.c
@@ -843,123 +843,93 @@ struct mail_eml *mail_eml_new(void)
     return eml;
 }
 
-int mail_publish_command(struct mail_decoder *env, struct session *sess,
+int mail_publish_command(struct mq_runtime *mq, int topic_id, struct session *sess,
                                         enum MAIL_PROTOCOL protocol, size_t mail_seq, struct mail_command *cmd)
 {
-    struct mq_runtime *runtime;
     struct mail_message *msg;
 
     if ((cmd->arg == NULL && cmd->cmd_line == NULL) || (cmd->arg_len == 0 && cmd->cmd_line_len == 0)) {
         return -1;
     }
 
-    runtime = module_manager_get_mq_runtime(env->mod_mgr_ref);
-    if (runtime == NULL) {
-        return -1;
-    }
-
     msg = (struct mail_message *)calloc(1, sizeof(struct mail_message));
     msg->sess_ref = sess;
     msg->mail_seq = mail_seq;
     msg->mail_protocol = protocol;
     msg->command = mail_command_clone(cmd);
 
-    mq_runtime_publish_message(runtime, env->command_topic_id, msg);
+    mq_runtime_publish_message(mq, topic_id, msg);
     return 0;
 }
 
-int mail_publish_header(struct mail_decoder *env, struct session *sess, enum MAIL_PROTOCOL protocol, size_t mail_seq, struct mail_header *header)
+int mail_publish_header(struct mq_runtime *mq, int topic_id, struct session *sess, enum MAIL_PROTOCOL protocol, size_t mail_seq, struct mail_header *header)
 {
-    struct mq_runtime *runtime;
     struct mail_message *msg;
 
     if (header == NULL || header->n_header_fields == 0) {
         return -1;
     }
 
-    runtime = module_manager_get_mq_runtime(env->mod_mgr_ref);
-    if (runtime == NULL) {
-        return -1;
-    }
-
     msg = (struct mail_message *)calloc(1, sizeof(struct mail_message));
     msg->sess_ref = sess;
     msg->mail_seq = mail_seq;
     msg->mail_protocol = protocol;
     msg->header = header;
 
-    mq_runtime_publish_message(runtime, env->header_topic_id, msg);
+    mq_runtime_publish_message(mq, topic_id, msg);
     return 0;
 }
 
-int mail_publish_body(struct mail_decoder *env, struct session *sess, enum MAIL_PROTOCOL protocol, size_t mail_seq, struct mail_body *body)
+int mail_publish_body(struct mq_runtime *mq, int topic_id, struct session *sess, enum MAIL_PROTOCOL protocol, size_t mail_seq, struct mail_body *body)
 {
-    struct mq_runtime *runtime;
     struct mail_message *msg;
 
     if (body == NULL || body->body_offset == 0) {
         return -1;
     }
 
-    runtime = module_manager_get_mq_runtime(env->mod_mgr_ref);
-    if (runtime == NULL) {
-        return -1;
-    }
-
     msg = (struct mail_message *)calloc(1, sizeof(struct mail_message));
     msg->sess_ref = sess;
     msg->mail_seq = mail_seq;
     msg->mail_protocol = protocol;
     msg->body = mail_body_clone(body);
 
-    mq_runtime_publish_message(runtime, env->body_topic_id, msg);
+    mq_runtime_publish_message(mq, topic_id, msg);
     return 0;
 }
 
-int mail_publish_attachment(struct mail_decoder *env, struct session *sess, enum MAIL_PROTOCOL protocol, size_t mail_seq, struct mail_attachment *attachment)
+int mail_publish_attachment(struct mq_runtime *mq, int topic_id, struct session *sess, enum MAIL_PROTOCOL protocol, size_t mail_seq, struct mail_attachment *attachment)
 {
-    struct mq_runtime *runtime;
     struct mail_message *msg;
 
     if (attachment == NULL || attachment->attachment_name_len == 0 || attachment->attachment_offset == 0) {
         return -1;
     }
 
-    runtime = module_manager_get_mq_runtime(env->mod_mgr_ref);
-    if (runtime == NULL) {
-        return -1;
-    }
-
     msg = (struct mail_message *)calloc(1, sizeof(struct mail_message));
     msg->sess_ref = sess;
     msg->mail_seq = mail_seq;
     msg->mail_protocol = protocol;
     msg->attachment = mail_attachment_clone(attachment);
 
-    mq_runtime_publish_message(runtime, env->attachment_topic_id, msg);
+    mq_runtime_publish_message(mq, topic_id, msg);
     return 0;
 }
 
-int mail_publish_eml(struct mail_decoder *env, struct session *sess, enum MAIL_PROTOCOL protocol, struct mail_eml *eml)
+int mail_publish_eml(struct mq_runtime *mq, int topic_id, struct session *sess, enum MAIL_PROTOCOL protocol, struct mail_eml *eml)
 {
-    struct mq_runtime *runtime;
     struct mail_message *msg;
 
     if (eml == NULL || eml->eml_offset == 0) {
         return -1;
     }
 
-    runtime = module_manager_get_mq_runtime(env->mod_mgr_ref);
-    if (runtime == NULL) {
-        return -1;
-    }
-
     msg = (struct mail_message *)calloc(1, sizeof(struct mail_message));
     msg->sess_ref = sess;
     msg->mail_protocol = protocol;
     msg->eml = mail_eml_clone(eml);
 
-    mq_runtime_publish_message(runtime, env->eml_topic_id, msg);
+    mq_runtime_publish_message(mq, topic_id, msg);
     return 0;
 }
 
@@ -1313,7 +1283,7 @@ void email_parser_free(struct email_parser *eml_parser)
     free(eml_parser); 
 }
 
-struct email_parser *email_parser_new(struct mail_decoder *env, enum MAIL_PROTOCOL protocol, int is_c2s)
+struct email_parser *email_parser_new(struct mq_runtime *mq, struct mail_topics *topics, enum MAIL_PROTOCOL protocol, int is_c2s)
 {
     struct email_parser* eml_parser = NULL;
 
@@ -1346,7 +1316,11 @@ struct email_parser *email_parser_new(struct mail_decoder *env, enum MAIL_PROTOC
     eml_parser->body = mail_body_new();
     eml_parser->current_attachment = mail_attachment_new();
     eml_parser->eml = mail_eml_new();
-    eml_parser->mail_env_ref = env;
+    eml_parser->header_topic_id = topics->topic_ids[MAIL_TOPIC_HEADER];
+    eml_parser->body_topic_id = topics->topic_ids[MAIL_TOPIC_BODY];
+    eml_parser->attachment_topic_id = topics->topic_ids[MAIL_TOPIC_ATTACHMENT];
+    eml_parser->eml_topic_id = topics->topic_ids[MAIL_TOPIC_EML];
+    eml_parser->mq = mq;
     return eml_parser;
 }
 
@@ -1359,7 +1333,7 @@ int email_parser_entry(struct email_parser *eml_parser, struct session *sess, co
     } else {
         mail_eml_update(eml_parser->eml, NULL, 0, 1);
     }
-    mail_publish_eml(eml_parser->mail_env_ref, sess, eml_parser->protocol, eml_parser->eml);
+    mail_publish_eml(eml_parser->mq, eml_parser->eml_topic_id, sess, eml_parser->protocol, eml_parser->eml);
 
     if (is_eml_header_field(line_type) && eml_parser->header == NULL) {
         eml_parser->header = mail_header_new();
@@ -1434,11 +1408,11 @@ int email_parser_entry(struct email_parser *eml_parser, struct session *sess, co
             // body
             if (eml_parser->MailInfoState == MAIL_GET_COUNT) {
                 mail_body_update(eml_parser->body, eml_parser->mimeinfo->dst, eml_parser->mimeinfo->actLen, 0);
-                mail_publish_body(eml_parser->mail_env_ref, sess, eml_parser->protocol, eml_parser->mail_seq, eml_parser->body);
+                mail_publish_body(eml_parser->mq, eml_parser->body_topic_id, sess, eml_parser->protocol, eml_parser->mail_seq, eml_parser->body);
             }
             if (eml_parser->MailInfoState == MAIL_GET_COUNT_END) {
                 mail_body_update(eml_parser->body, NULL, 0, 1);
-                mail_publish_body(eml_parser->mail_env_ref, sess, eml_parser->protocol, eml_parser->mail_seq,eml_parser->body);
+                mail_publish_body(eml_parser->mq, eml_parser->body_topic_id, sess, eml_parser->protocol, eml_parser->mail_seq,eml_parser->body);
                 mail_body_reset(eml_parser->body);
             }
 
@@ -1450,17 +1424,17 @@ int email_parser_entry(struct email_parser *eml_parser, struct session *sess, co
             }
             if (eml_parser->MailInfoState == MAIL_GET_FILECOUNT) {
                 mail_attachment_update(eml_parser->current_attachment, eml_parser->mimeinfo->dst, eml_parser->mimeinfo->actLen, 0);
-                mail_publish_attachment(eml_parser->mail_env_ref, sess, eml_parser->protocol, eml_parser->mail_seq, eml_parser->current_attachment);
+                mail_publish_attachment(eml_parser->mq, eml_parser->attachment_topic_id, sess, eml_parser->protocol, eml_parser->mail_seq, eml_parser->current_attachment);
             }
 
             if (eml_parser->MailInfoState == MAIL_GET_FILECOUNT_END) {
                 mail_attachment_update(eml_parser->current_attachment, NULL, 0, 1);
-                mail_publish_attachment(eml_parser->mail_env_ref, sess, eml_parser->protocol, eml_parser->mail_seq, eml_parser->current_attachment);
+                mail_publish_attachment(eml_parser->mq, eml_parser->attachment_topic_id, sess, eml_parser->protocol, eml_parser->mail_seq, eml_parser->current_attachment);
                 mail_attachment_reset(eml_parser->current_attachment);
             }
             break;
         case EML_DATA_HEAD_BODY_BORDER:
-            mail_publish_header(eml_parser->mail_env_ref, sess, eml_parser->protocol, eml_parser->mail_seq, eml_parser->header);
+            mail_publish_header(eml_parser->mq, eml_parser->header_topic_id, sess, eml_parser->protocol, eml_parser->mail_seq, eml_parser->header);
             eml_parser->header = NULL;
 
             eml_parser->EmlAnalyseState = MAIL_STAT_BODY;
diff --git a/decoders/mail/mail_decoder_email.h b/decoders/mail/mail_decoder_email.h
index f6a2e7a..caf3935 100644
--- a/decoders/mail/mail_decoder_email.h
+++ b/decoders/mail/mail_decoder_email.h
@@ -211,13 +211,19 @@ struct email_parser
 	long pre_uid;		//IMAPЭ�鰴���ֽ�����ȡ�ʼ����ݵ����
 	long uid;
 
-    struct mail_decoder *mail_env_ref;
     size_t mail_seq;
     enum MAIL_PROTOCOL protocol;
     struct mail_header *header;
     struct mail_body *body;
     struct mail_attachment *current_attachment;
     struct mail_eml *eml;
+
+    int eml_topic_id;
+    int header_topic_id;
+    int body_topic_id;
+    int attachment_topic_id;
+    struct mq_runtime *mq;
+
 };
 
 int mail_command_update(struct mail_command *command, enum MAIL_COMMAND cmd, const char *cmd_arg, size_t cmd_arg_len, const char *cmd_line, size_t cmd_line_len);
@@ -250,11 +256,11 @@ int mail_eml_reset(struct mail_eml *eml);
 void mail_eml_free(struct mail_eml *eml);
 struct mail_eml *mail_eml_clone(struct mail_eml *eml);
 struct mail_eml *mail_eml_new(void);
-int mail_publish_command(struct mail_decoder *env, struct session *sess, enum MAIL_PROTOCOL protocol, size_t mail_seq, struct mail_command *cmd);
-int mail_publish_header(struct mail_decoder *env, struct session *sess, enum MAIL_PROTOCOL protocol, size_t mail_seq, struct mail_header *header);
-int mail_publish_body(struct mail_decoder *env, struct session *sess, enum MAIL_PROTOCOL protocol, size_t mail_seq, struct mail_body *body);
-int mail_publish_attachment(struct mail_decoder *env, struct session *sess, enum MAIL_PROTOCOL protocol, size_t mail_seq, struct mail_attachment *attachment);
-int mail_publish_eml(struct mail_decoder *env, struct session *sess, enum MAIL_PROTOCOL protocol, struct mail_eml *eml);
+int mail_publish_command(struct mq_runtime *mq, int topic_id, struct session *sess, enum MAIL_PROTOCOL protocol, size_t mail_seq, struct mail_command *cmd);
+int mail_publish_header(struct mq_runtime *mq, int topic_id, struct session *sess, enum MAIL_PROTOCOL protocol, size_t mail_seq, struct mail_header *header);
+int mail_publish_body(struct mq_runtime *mq, int topic_id, struct session *sess, enum MAIL_PROTOCOL protocol, size_t mail_seq, struct mail_body *body);
+int mail_publish_attachment(struct mq_runtime *mq, int topic_id, struct session *sess, enum MAIL_PROTOCOL protocol, size_t mail_seq, struct mail_attachment *attachment);
+int mail_publish_eml(struct mq_runtime *mq, int topic_id, struct session *sess, enum MAIL_PROTOCOL protocol, struct mail_eml *eml);
 void mail_message_free(void *msg, void *arg);
 char *mail_message_to_json(struct mail_message *msg);
 
@@ -269,7 +275,7 @@ int email_parser_get_tcpdata(struct email_parser* eml_parser, const char *payloa
 int email_parser_entry(struct email_parser *eml_parser, struct session *sess, const char *email_line, size_t line_len, enum EML_LINE_TYPE line_type);
 int email_parser_reset(struct email_parser * eml_parser);
 void email_parser_free(struct email_parser *eml_parser);
-struct email_parser * email_parser_new(struct mail_decoder *env, enum MAIL_PROTOCOL protocol, int is_c2s);
+struct email_parser * email_parser_new(struct mq_runtime *mq, struct mail_topics *topics, enum MAIL_PROTOCOL protocol, int is_c2s);
 
 #endif
 
diff --git a/decoders/mail/mail_decoder_module.c b/decoders/mail/mail_decoder_module.c
index 343f792..39689ec 100644
--- a/decoders/mail/mail_decoder_module.c
+++ b/decoders/mail/mail_decoder_module.c
@@ -20,6 +20,14 @@
 //#include "mail_decoder_imap.h"
 #include "mail_decoder_module.h"
 
+const char * g_mail_topic_name[MAIL_TOPIC_MAX] = {
+    "MAIL_COMMAND",
+    "MAIL_HEADER",
+    "MAIL_BODY",
+    "MAIL_ATTACHMENT",
+    "MAIL_EML",
+};
+
 static unsigned short get_session_dest_port(struct session *sess)
 {
     const struct layer *first_pkt_layer;
@@ -90,17 +98,11 @@ static int is_current_packet_c2s(struct session *sess)
     return 1;
 }
 
-static int mail_subscribe_command(struct mail_decoder *mail_env, mail_command_callback_func *cb, void *arg)
+static int mail_subscribe_command(struct mq_schema *schema, mail_command_callback_func *cb, void *arg)
 {
     int topic;
-    struct mq_schema *schema;
-
-    schema = module_manager_get_mq_schema(mail_env->mod_mgr_ref);
-    if (schema == NULL) {
-        return -1;
-    }
 
-    topic = mq_schema_get_topic_id(schema, MAIL_COMMAND_TOPIC_NAME);
+    topic = mq_schema_get_topic_id(schema, g_mail_topic_name[MAIL_TOPIC_COMMAND]);
     if (topic < 0) {
         return -1;
     }
@@ -108,17 +110,11 @@ static int mail_subscribe_command(struct mail_decoder *mail_env, mail_command_ca
     return mq_schema_subscribe(schema, topic, (on_msg_cb_func *)(void *)cb, arg);
 }
 
-static int mail_subscribe_header(struct mail_decoder *mail_env, mail_header_callback_func *cb, void *arg)
+static int mail_subscribe_header(struct mq_schema *schema, mail_header_callback_func *cb, void *arg)
 {
     int topic;
-    struct mq_schema *schema;
-
-    schema = module_manager_get_mq_schema(mail_env->mod_mgr_ref);
-    if (schema == NULL) {
-        return -1;
-    }
 
-    topic = mq_schema_get_topic_id(schema, MAIL_HEADER_TOPIC_NAME);
+    topic = mq_schema_get_topic_id(schema, g_mail_topic_name[MAIL_TOPIC_HEADER]);
     if (topic < 0) {
         return -1;
     }
@@ -126,17 +122,11 @@ static int mail_subscribe_header(struct mail_decoder *mail_env, mail_header_call
     return mq_schema_subscribe(schema, topic, (on_msg_cb_func *)(void *)cb, arg);
 }
 
-static int mail_subscribe_body(struct mail_decoder *mail_env, mail_body_callback_func *cb, void *arg)
+static int mail_subscribe_body(struct mq_schema *schema, mail_body_callback_func *cb, void *arg)
 {
     int topic;
-    struct mq_schema *schema;
 
-    schema = module_manager_get_mq_schema(mail_env->mod_mgr_ref);
-    if (schema == NULL) {
-        return -1;
-    }
-
-    topic = mq_schema_get_topic_id(schema, MAIL_BODY_TOPIC_NAME);
+    topic = mq_schema_get_topic_id(schema, g_mail_topic_name[MAIL_TOPIC_BODY]);
     if (topic < 0) {
         return -1;
     }
@@ -144,17 +134,11 @@ static int mail_subscribe_body(struct mail_decoder *mail_env, mail_body_callback
     return mq_schema_subscribe(schema, topic, (on_msg_cb_func *)(void *)cb, arg);
 }
 
-static int mail_subscribe_attachment(struct mail_decoder *mail_env, mail_attachment_callback_func *cb, void *arg)
+static int mail_subscribe_attachment(    struct mq_schema *schema, mail_attachment_callback_func *cb, void *arg)
 {
     int topic;
-    struct mq_schema *schema;
 
-    schema = module_manager_get_mq_schema(mail_env->mod_mgr_ref);
-    if (schema == NULL) {
-        return -1;
-    }
-
-    topic = mq_schema_get_topic_id(schema, MAIL_ATTACHMENT_TOPIC_NAME);
+    topic = mq_schema_get_topic_id(schema, g_mail_topic_name[MAIL_TOPIC_ATTACHMENT]);
     if (topic < 0) {
         return -1;
     }
@@ -162,17 +146,11 @@ static int mail_subscribe_attachment(struct mail_decoder *mail_env, mail_attachm
     return mq_schema_subscribe(schema, topic, (on_msg_cb_func *)(void *)cb, arg);
 }
 
-static int mail_subscribe_eml(struct mail_decoder *mail_env, mail_eml_callback_func *cb, void *arg)
+static int mail_subscribe_eml(struct mq_schema *schema, mail_eml_callback_func *cb, void *arg)
 {
     int topic;
-    struct mq_schema *schema;
 
-    schema = module_manager_get_mq_schema(mail_env->mod_mgr_ref);
-    if (schema == NULL) {
-        return -1;
-    }
-
-    topic = mq_schema_get_topic_id(schema, MAIL_EML_TOPIC_NAME);
+    topic = mq_schema_get_topic_id(schema, g_mail_topic_name[MAIL_TOPIC_EML]);
     if (topic < 0) {
         return -1;
     }
@@ -183,16 +161,18 @@ static int mail_subscribe_eml(struct mail_decoder *mail_env, mail_eml_callback_f
 static void mail_dispatch(int topic, void *msg, on_msg_cb_func *msg_cb, void *arg, void *dispatch_arg)
 {
     struct mail_message *mail_msg;
-    struct mail_decoder *mail_env;
+    struct mail_decoder *decoder;
+    struct mail_topics *topics;
 
     mail_msg = (struct mail_message *)msg;
-    mail_env = (struct mail_decoder *)dispatch_arg;
+    decoder = (struct mail_decoder *)dispatch_arg;
+    topics = &decoder->topics;
 
     //char *msg_json = mail_message_to_json(mail_msg);
     //printf("%s\n", msg_json);
     //free(msg_json);
 
-    if (topic == mail_env->command_topic_id) {
+    if (topic == topics->topic_ids[MAIL_TOPIC_COMMAND]) {
         mail_command_callback_func *command_cb = (mail_command_callback_func *)(void *)msg_cb;
         command_cb(mail_msg->sess_ref, mail_msg->mail_protocol, mail_msg->mail_seq,
                 mail_msg->command->cmd, (const char *)mail_msg->command->arg, mail_msg->command->arg_len,
@@ -201,13 +181,13 @@ static void mail_dispatch(int topic, void *msg, on_msg_cb_func *msg_cb, void *ar
         return;
     }
 
-    if (topic == mail_env->header_topic_id) {
+    if (topic == topics->topic_ids[MAIL_TOPIC_HEADER]) {
         mail_header_callback_func *header_cb = (mail_header_callback_func *)(void *)msg_cb;
         header_cb(mail_msg->sess_ref, mail_msg->mail_protocol, mail_msg->mail_seq, (const struct mail_header *)mail_msg->header, arg);
         return;
     }
 
-    if (topic == mail_env->body_topic_id) {
+    if (topic == topics->topic_ids[MAIL_TOPIC_BODY]) {
         mail_body_callback_func *body_cb = (mail_body_callback_func *)(void *)msg_cb;
         body_cb(mail_msg->sess_ref, mail_msg->mail_protocol, mail_msg->mail_seq,
                 (const char *)mail_msg->body->body, mail_msg->body->body_len,
@@ -216,7 +196,7 @@ static void mail_dispatch(int topic, void *msg, on_msg_cb_func *msg_cb, void *ar
         return;
     }
 
-    if (topic == mail_env->attachment_topic_id) {
+    if (topic == topics->topic_ids[MAIL_TOPIC_ATTACHMENT]) {
         mail_attachment_callback_func *attachment_cb = (mail_attachment_callback_func *)(void *)msg_cb;
         attachment_cb(mail_msg->sess_ref, mail_msg->mail_protocol, mail_msg->mail_seq,
                 (const char *)mail_msg->attachment->attachment_name, mail_msg->attachment->attachment_name_len,
@@ -226,7 +206,7 @@ static void mail_dispatch(int topic, void *msg, on_msg_cb_func *msg_cb, void *ar
         return;
     }
 
-    if (topic == mail_env->eml_topic_id) {
+    if (topic == topics->topic_ids[MAIL_TOPIC_EML]) {
         mail_eml_callback_func *eml_cb = (mail_eml_callback_func *)(void *)msg_cb;
         eml_cb(mail_msg->sess_ref, mail_msg->mail_protocol,
                 (const char *)mail_msg->eml->eml, mail_msg->eml->eml_len,
@@ -236,7 +216,7 @@ static void mail_dispatch(int topic, void *msg, on_msg_cb_func *msg_cb, void *ar
     }
 }
 
-int mail_subscribe(struct mail_decoder *mail_env,
+int mail_subscribe(struct mail_decoder *decoder,
                         mail_command_callback_func command_cb,
                         mail_header_callback_func *header_cb,
                         mail_body_callback_func *body_cb,
@@ -246,27 +226,27 @@ int mail_subscribe(struct mail_decoder *mail_env,
 {
     int ret;
 
-    ret = mail_subscribe_command(mail_env, command_cb, arg);
+    ret = mail_subscribe_command(decoder->mq_schema, command_cb, arg);
     if (ret < 0) {
         return ret;
     }
 
-    ret = mail_subscribe_header(mail_env, header_cb, arg);
+    ret = mail_subscribe_header(decoder->mq_schema, header_cb, arg);
     if (ret < 0) {
         return ret;
     }
 
-    ret = mail_subscribe_body(mail_env, body_cb, arg);
+    ret = mail_subscribe_body(decoder->mq_schema, body_cb, arg);
     if (ret < 0) {
         return ret;
     }
 
-    ret = mail_subscribe_attachment(mail_env, attachment_cb, arg);
+    ret = mail_subscribe_attachment(decoder->mq_schema, attachment_cb, arg);
     if (ret < 0) {
         return ret;
     }
 
-    ret = mail_subscribe_eml(mail_env, eml_cb, arg);
+    ret = mail_subscribe_eml(decoder->mq_schema, eml_cb, arg);
     if (ret < 0) {
         return ret;
     }
@@ -309,15 +289,18 @@ static struct mail_session_ctx *mail_decoder_session_ctx_new(void)
 //    return 0;
 //}
 
-static int mail_decoder_smtp_process(struct mail_decoder *env, struct mail_session_ctx *session_ctx, const char *payload, size_t payload_len, int is_c2s)
+static int mail_decoder_smtp_process(struct mail_decoder *decoder, struct mail_session_ctx *session_ctx, const char *payload, size_t payload_len, int is_c2s)
 {
-    char ret;;
+    char ret;
+    struct mq_runtime *mq;
 
-    if (session_ctx->smtp_parser == NULL) {
-        session_ctx->smtp_parser = smtp_parser_new(env, is_c2s);
+    mq = module_manager_get_mq_runtime(decoder->mod_mgr);
+
+    if (session_ctx->smtp_parser == NULL && mq != NULL) {
+        session_ctx->smtp_parser = smtp_parser_new(mq, &decoder->topics, is_c2s);
     }
 
-    ret = smtp_parser_entry(session_ctx->smtp_parser, session_ctx->sess_ref, payload, payload_len,is_c2s);
+    ret = smtp_parser_entry(session_ctx->smtp_parser, session_ctx->sess_ref, payload, payload_len, is_c2s);
     if (ret != 0) {
         return -1;
     }
@@ -331,7 +314,7 @@ static void mail_decoder_on_tcp_payload(struct session *sess, enum session_state
     int ret;
     int is_packet_c2s;
     struct mail_session_ctx *session_ctx;
-    struct mail_decoder *env = (struct mail_decoder *)arg;
+    struct mail_decoder *decoder = (struct mail_decoder *)arg;
 
     if (payload == NULL || payload_len == 0) {
         return;
@@ -339,14 +322,13 @@ static void mail_decoder_on_tcp_payload(struct session *sess, enum session_state
 
     is_packet_c2s = is_current_packet_c2s(sess);
 
-    session_ctx = (struct mail_session_ctx *)session_get_exdata(sess, env->exdata_id);
+    session_ctx = (struct mail_session_ctx *)session_get_exdata(sess, decoder->exdata_id);
     if (session_ctx == NULL) {
         session_ctx = mail_decoder_session_ctx_new();
-        session_ctx->mail_env_ref = env;
         session_ctx->sess_ref = sess;
         session_ctx->protocol = MAIL_PROTOCOL_MAX;
         session_ctx->is_droped = 0;
-        session_set_exdata(sess, env->exdata_id, session_ctx);
+        session_set_exdata(sess, decoder->exdata_id, session_ctx);
 
         unsigned short session_dest_port = get_session_dest_port(sess);
 
@@ -383,7 +365,7 @@ static void mail_decoder_on_tcp_payload(struct session *sess, enum session_state
 
     switch (session_ctx->protocol) {
         case MAIL_PROTOCOL_SMTP:
-            ret = mail_decoder_smtp_process(env, session_ctx, payload, payload_len, is_packet_c2s);
+            ret = mail_decoder_smtp_process(decoder, session_ctx, payload, payload_len, is_packet_c2s);
             break;
         case MAIL_PROTOCOL_POP3:
             //ret = mail_decoder_pop3_process(decoder, session_ctx, payload, payload_len , is_packet_c2s);
@@ -414,19 +396,18 @@ void mail_decoder_on_exdata_free(int idx, void *ex_ptr, void *arg)
 void mail_exit(struct module_manager *mod_mgr, struct module *mod)
 {
     (void)(mod_mgr);
-    struct mail_decoder *mail_env;
+    struct mail_decoder *decoder;
     struct mq_schema *schema;
 
     if (mod) {
-        mail_env = (struct mail_decoder *)module_get_ctx(mod);
-        if (mail_env) {
+        decoder = (struct mail_decoder *)module_get_ctx(mod);
+        if (decoder) {
             schema = module_manager_get_mq_schema(mod_mgr);
-            mq_schema_destroy_topic(schema, mail_env->command_topic_id);
-            mq_schema_destroy_topic(schema, mail_env->header_topic_id);
-            mq_schema_destroy_topic(schema, mail_env->body_topic_id);
-            mq_schema_destroy_topic(schema, mail_env->attachment_topic_id);
-            mq_schema_destroy_topic(schema, mail_env->eml_topic_id);
-            free(mail_env);
+            int i;
+            for (i = 0; i < MAIL_TOPIC_MAX; i++) {
+                mq_schema_destroy_topic(schema, decoder->topics.topic_ids[i]);
+            }
+            free(decoder);
         }
 
         module_free(mod);
@@ -437,49 +418,39 @@ struct module* mail_init(struct module_manager *mod_mgr)
 {
     int ret;
     struct module *mod;
-    struct mq_schema *schema;
     struct session_manager *sess_mgr;
-    struct mail_decoder *env;
+    struct mail_decoder *decoder;
+    struct mail_topics *topics;
 
-    env = (struct mail_decoder *)calloc(1, sizeof(struct mail_decoder));
-    env->mod_mgr_ref = mod_mgr;
-    mod = module_new(MAIL_MODULE_NAME, env);
-    sess_mgr = module_to_session_manager(module_manager_get_module(mod_mgr, SESSION_MANAGER_MODULE_NAME));
-    schema = module_manager_get_mq_schema(mod_mgr);
+    decoder = (struct mail_decoder *)calloc(1, sizeof(struct mail_decoder));
+    decoder->mod_mgr = mod_mgr;
+    mod = module_new(MAIL_MODULE_NAME, decoder);
 
-    if (sess_mgr == NULL || schema == NULL) {
+    sess_mgr = module_to_session_manager(module_manager_get_module(mod_mgr, SESSION_MANAGER_MODULE_NAME));
+    if (sess_mgr == NULL) {
         goto exit;
     }
 
-    ret = session_manager_subscribe_tcp_stream(sess_mgr, mail_decoder_on_tcp_payload, env);
+    ret = session_manager_subscribe_tcp_stream(sess_mgr, mail_decoder_on_tcp_payload, decoder);
     if (ret < 0) {
         goto exit;
     }
 
-    env->exdata_id = session_manager_new_session_exdata_index(sess_mgr, MAIL_EXDATA_NAME, mail_decoder_on_exdata_free, NULL);	
-    if (env->exdata_id < 0) {
+    decoder->exdata_id = session_manager_new_session_exdata_index(sess_mgr, MAIL_EXDATA_NAME, mail_decoder_on_exdata_free, NULL);	
+    if (decoder->exdata_id < 0) {
         goto exit;
     }
 
-    env->command_topic_id = mq_schema_create_topic(schema, MAIL_COMMAND_TOPIC_NAME, mail_dispatch, env, mail_message_free, NULL);
-    if (env->command_topic_id < 0) {
-        goto exit;
-    }
-    env->header_topic_id = mq_schema_create_topic(schema, MAIL_HEADER_TOPIC_NAME, mail_dispatch, env, mail_message_free, NULL);
-    if (env->header_topic_id < 0) {
-        goto exit;
-    }
-    env->body_topic_id = mq_schema_create_topic(schema, MAIL_BODY_TOPIC_NAME, mail_dispatch, env, mail_message_free, NULL);
-    if (env->body_topic_id < 0) {
-        goto exit;
-    }
-    env->attachment_topic_id = mq_schema_create_topic(schema, MAIL_ATTACHMENT_TOPIC_NAME, mail_dispatch, env, mail_message_free, NULL);
-    if (env->attachment_topic_id < 0) {
-        goto exit;
-    }
-    env->eml_topic_id = mq_schema_create_topic(schema, MAIL_EML_TOPIC_NAME, mail_dispatch, env, mail_message_free, NULL);
-    if (env->eml_topic_id < 0) {
-        goto exit;
+    decoder->mq_schema = module_manager_get_mq_schema(mod_mgr);
+
+    int i;
+    topics = &decoder->topics;
+    for (i = 0; i < MAIL_TOPIC_MAX; i++) {
+    topics->topic_ids[i] = mq_schema_create_topic(decoder->mq_schema, g_mail_topic_name[i], mail_dispatch, decoder, mail_message_free, NULL);
+        if (topics->topic_ids[i] < 0) {
+            goto exit;
+        }
+        topics->topic_names[i] = g_mail_topic_name[i];
     }
 
     return mod;
diff --git a/decoders/mail/mail_decoder_module.h b/decoders/mail/mail_decoder_module.h
index ddd4fa9..4ba5515 100644
--- a/decoders/mail/mail_decoder_module.h
+++ b/decoders/mail/mail_decoder_module.h
@@ -4,11 +4,6 @@
 #include "mail_decoder_util.h"
 
 #define MAIL_EXDATA_NAME            "MAIL_EXDATA"
-#define MAIL_COMMAND_TOPIC_NAME     "MAIL_COMMAND"
-#define MAIL_HEADER_TOPIC_NAME      "MAIL_HEADER"
-#define MAIL_BODY_TOPIC_NAME        "MAIL_BODY"
-#define MAIL_ATTACHMENT_TOPIC_NAME  "MAIL_ATTACHMENT"
-#define MAIL_EML_TOPIC_NAME         "MAIL_EML"
 
 #define PORT_SMTP	25
 #define PORT_POP3	110
@@ -72,25 +67,21 @@ typedef struct _mail_local_info
 struct mail_session_ctx {
     enum MAIL_PROTOCOL protocol;
 
+    int is_droped;
     struct smtp_parser *smtp_parser;
 
-    int is_droped;
-    struct mail_decoder   *mail_env_ref;
     struct session        *sess_ref;
 };
 
 struct mail_decoder {
-    int command_topic_id;
-    int header_topic_id;
-    int body_topic_id;
-    int attachment_topic_id;
-    int eml_topic_id;
     int exdata_id;
 
+    struct mail_topics topics;
+    struct mq_runtime *mq;
+    struct mq_schema *mq_schema;
+    struct module_manager *mod_mgr;
     // tobe delete
     stMailLocalInfo mail_local_info;
-
-    struct module_manager *mod_mgr_ref;
 };
 
 #endif
diff --git a/decoders/mail/mail_decoder_smtp.c b/decoders/mail/mail_decoder_smtp.c
index e8a6fc1..5ad8370 100644
--- a/decoders/mail/mail_decoder_smtp.c
+++ b/decoders/mail/mail_decoder_smtp.c
@@ -268,7 +268,7 @@ static int process_user_pass(struct smtp_parser *parser, struct session *sess, c
                                 email->plug_mailinfo.pMailInfo->username->buf, email->plug_mailinfo.pMailInfo->username->buflen);
 
         mail_seq = email_parser_get_seq(parser->eml_parser);
-        mail_publish_command(parser->mail_env_ref, sess, MAIL_PROTOCOL_SMTP, mail_seq, parser->current_command);
+        mail_publish_command(parser->mq, parser->command_topic_id, sess, MAIL_PROTOCOL_SMTP, mail_seq, parser->current_command);
     }
     if((email->MailInfoState & MAIL_AFTER_PASS) && email->plug_mailinfo.pMailInfo->password->buf != NULL) {
         mail_command_update(parser->current_command, MAIL_CMD_PASSWORD,
@@ -276,7 +276,7 @@ static int process_user_pass(struct smtp_parser *parser, struct session *sess, c
                                 email->plug_mailinfo.pMailInfo->password->buf, email->plug_mailinfo.pMailInfo->password->buflen);
 
         mail_seq = email_parser_get_seq(parser->eml_parser);
-        mail_publish_command(parser->mail_env_ref, sess, MAIL_PROTOCOL_SMTP, mail_seq, parser->current_command);
+        mail_publish_command(parser->mq, parser->command_topic_id, sess, MAIL_PROTOCOL_SMTP, mail_seq, parser->current_command);
     }
 
     return 0;
@@ -312,7 +312,7 @@ int process_oauth2(struct smtp_parser *parser, struct session *sess, char *cmdda
                             email->plug_mailinfo.pMailInfo->username->buf, email->plug_mailinfo.pMailInfo->username->buflen);
 
     mail_seq = email_parser_get_seq(parser->eml_parser);
-    mail_publish_command(parser->mail_env_ref, sess, MAIL_PROTOCOL_SMTP, mail_seq, parser->current_command);
+    mail_publish_command(parser->mq, parser->command_topic_id, sess, MAIL_PROTOCOL_SMTP, mail_seq, parser->current_command);
 
     return 0;
 }
@@ -342,7 +342,7 @@ int process_ehlo(struct smtp_parser *parser, struct session *sess, const char *l
     mail_command_update(parser->current_command, MAIL_CMD_EHLO, arg_start, arg_len, line_start, line_len);
 
     mail_seq = email_parser_get_seq(parser->eml_parser);
-    mail_publish_command(parser->mail_env_ref, sess, MAIL_PROTOCOL_SMTP, mail_seq, parser->current_command);
+    mail_publish_command(parser->mq, parser->command_topic_id, sess, MAIL_PROTOCOL_SMTP, mail_seq, parser->current_command);
     return 0;
 }
 
@@ -377,7 +377,7 @@ int process_mail_from(struct smtp_parser *parser, struct session *sess, char *li
     mail_command_update(parser->current_command, MAIL_CMD_MAIL_FROM, arg_start, arg_len, line_start, line_len);
 
     mail_seq = email_parser_get_seq(parser->eml_parser);
-    mail_publish_command(parser->mail_env_ref, sess, MAIL_PROTOCOL_SMTP, mail_seq, parser->current_command);
+    mail_publish_command(parser->mq, parser->command_topic_id, sess, MAIL_PROTOCOL_SMTP, mail_seq, parser->current_command);
 
     return 0;
 }
@@ -414,7 +414,7 @@ int process_rcpt_to(struct smtp_parser *parser, struct session *sess, char *line
     mail_command_update(parser->current_command, MAIL_CMD_RCPT_TO, arg_start, arg_len, line_start, line_len);
 
     mail_seq = email_parser_get_seq(parser->eml_parser);
-    mail_publish_command(parser->mail_env_ref, sess, MAIL_PROTOCOL_SMTP, mail_seq, parser->current_command);
+    mail_publish_command(parser->mq, parser->command_topic_id, sess, MAIL_PROTOCOL_SMTP, mail_seq, parser->current_command);
     return 0;
 }
 
@@ -433,7 +433,7 @@ int process_starttls(struct smtp_parser *parser, struct session *sess, char *lin
     mail_command_update(parser->current_command, MAIL_CMD_STARTTLS, NULL, 0, line_start, line_len);
 
     mail_seq = email_parser_get_seq(parser->eml_parser);
-    mail_publish_command(parser->mail_env_ref, sess, MAIL_PROTOCOL_SMTP, mail_seq, parser->current_command);
+    mail_publish_command(parser->mq, parser->command_topic_id, sess, MAIL_PROTOCOL_SMTP, mail_seq, parser->current_command);
     return 0;
 }
 
@@ -561,12 +561,13 @@ void smtp_parser_free(struct smtp_parser *parser)
     free(parser);
 }
 
-struct smtp_parser *smtp_parser_new(struct mail_decoder *env, int is_c2s)
+struct smtp_parser *smtp_parser_new(struct mq_runtime *mq, struct mail_topics *topics, int is_c2s)
 {
     struct smtp_parser *parser = (struct smtp_parser *)calloc(1, sizeof(struct smtp_parser));
-    parser->mail_env_ref = env;
+    parser->mq = mq;
+    parser->command_topic_id = topics->topic_ids[MAIL_TOPIC_COMMAND];
     parser->current_command = mail_command_new();
-    parser->eml_parser = email_parser_new(env, MAIL_PROTOCOL_SMTP, is_c2s);
+    parser->eml_parser = email_parser_new(mq, topics, MAIL_PROTOCOL_SMTP, is_c2s);
     return parser;
 }
 
diff --git a/decoders/mail/mail_decoder_smtp.h b/decoders/mail/mail_decoder_smtp.h
index 7d3b18f..5099dc4 100644
--- a/decoders/mail/mail_decoder_smtp.h
+++ b/decoders/mail/mail_decoder_smtp.h
@@ -55,13 +55,16 @@ typedef enum SMTP_CMD_STATE
 
 struct smtp_parser {
     struct mail_command *current_command;
+
+    int command_topic_id;
+    struct mq_runtime *mq;
+
     struct email_parser *eml_parser;
-    struct mail_decoder *mail_env_ref;
 };
 
 int smtp_identify(const char *payload, size_t payload_len, int is_c2s);
 int smtp_parser_entry(struct smtp_parser *parser, struct session *sess, const char *payload, size_t payload_len, int is_c2s);
 void smtp_parser_free(struct smtp_parser *parser);
-struct smtp_parser * smtp_parser_new(struct mail_decoder *env, int is_c2s);
+struct smtp_parser * smtp_parser_new(struct mq_runtime *mq, struct mail_topics *topics, int is_c2s);
 
 #endif
diff --git a/decoders/mail/mail_decoder_util.h b/decoders/mail/mail_decoder_util.h
index 23f70ab..ad25d4d 100644
--- a/decoders/mail/mail_decoder_util.h
+++ b/decoders/mail/mail_decoder_util.h
@@ -29,6 +29,20 @@
 #define MAIL_MAX_CHARSET_LEN 32
 #define MAIL_ELEM_NUM 16
 
+enum MAIL_TOPIC{
+    MAIL_TOPIC_COMMAND,
+    MAIL_TOPIC_HEADER,
+    MAIL_TOPIC_BODY,
+    MAIL_TOPIC_ATTACHMENT,
+    MAIL_TOPIC_EML,
+    MAIL_TOPIC_MAX,
+};
+
+struct mail_topics {
+    const char *topic_names[MAIL_TOPIC_MAX];
+    int topic_ids[MAIL_TOPIC_MAX];
+};
+
 typedef struct _mail_elem_info
 {
 	int buflen;
diff --git a/include/stellar/mail.h b/include/stellar/mail.h
index d75d6f9..c2492e2 100644
--- a/include/stellar/mail.h
+++ b/include/stellar/mail.h
@@ -80,9 +80,9 @@ typedef void mail_eml_callback_func(struct session *sess,
                                                     void *arg);
 
 
-struct mail_env;
-struct mail_env *module_to_mail_decoder(struct module *module);
-int mail_subscribe(struct mail_env *env,
+struct mail_decoder;
+struct mail_decoder *module_to_mail_decoder(struct module *module);
+int mail_subscribe(struct mail_decoder *decoder,
                         mail_command_callback_func *command_cb,
                         mail_header_callback_func *header_cb,
                         mail_body_callback_func *data_cb,
diff --git a/test/decoders/mail/mail_test_module.cpp b/test/decoders/mail/mail_test_module.cpp
index fab6961..4a84d27 100644
--- a/test/decoders/mail/mail_test_module.cpp
+++ b/test/decoders/mail/mail_test_module.cpp
@@ -566,7 +566,7 @@ extern "C" struct module *mail_test_init(struct module_manager *mod_mgr)
     struct module *mod;
     struct mail_test_module_ctx *mod_ctx;
     struct session_manager *sess_mgr;
-    struct mail_env *decoder;
+    struct mail_decoder *decoder;
 
     mod_ctx = (struct mail_test_module_ctx *)calloc(1, sizeof(struct mail_test_module_ctx));
     mod_ctx->mod_mgr_ref = mod_mgr;