diff --git a/largefile64.patch b/largefile64.patch new file mode 100644 index 0000000..d203549 --- /dev/null +++ b/largefile64.patch @@ -0,0 +1,13 @@ +diff --git a/makedefs b/makedefs +index cfba0e9..4860358 100644 +--- a/makedefs ++++ b/makedefs +@@ -450,7 +450,7 @@ case "$CCARGS" in + *-DNO_PCRE*) ;; + *) pcre_cflags=`(pcre-config --cflags) 2>/dev/null` && + pcre_libs=`(pcre-config --libs) 2>/dev/null` && { +- CCARGS="$CCARGS -DHAS_PCRE $pcre_cflags" ++ CCARGS="$CCARGS -DHAS_PCRE $pcre_cflags -D_LARGEFILE64_SOURCE" + AUXLIBS="$AUXLIBS $pcre_libs" + } + ;; diff --git a/makedefs b/makedefs index 0585fa0..cfba0e9 100644 --- a/makedefs +++ b/makedefs @@ -265,7 +265,7 @@ case "$SYSTEM.$RELEASE" in # CCARGS="$CCARGS -DHAS_DBM -DPATH_NDBM_H=''" # GDBM_LIBS=gdbm # fi - SYSLIBS="-ldb" + SYSLIBS="-ldb -laprutil-1" for name in nsl resolv $GDBM_LIBS do for lib in /usr/lib64 /lib64 /usr/lib /lib diff --git a/src/bounce/bounce_notify_service.c b/src/bounce/bounce_notify_service.c index 2f9fdb8..2a46741 100644 --- a/src/bounce/bounce_notify_service.c +++ b/src/bounce/bounce_notify_service.c @@ -76,6 +76,7 @@ /* Application-specific. */ #include "bounce_service.h" +#include "../util/postfix_mq.h" #define STR vstring_str @@ -237,6 +238,8 @@ int bounce_notify_service(int flags, char *service, char *queue_name, if (bounce_status == 0) msg_info("%s: sender non-delivery notification: %s", queue_id, STR(new_id)); + postfix_mq("Type:Message_bounce%cpid:%d%cqueue_id:%s%cbounce_id:%s", DELIMITER, getpid(), DELIMITER, queue_id, DELIMITER, STR(new_id)); + } else { /* No applicable recipients found - cancel this notice. */ (void) vstream_fclose(bounce); diff --git a/src/bounce/bounce_notify_verp.c b/src/bounce/bounce_notify_verp.c index dbc5e56..aa59dd2 100644 --- a/src/bounce/bounce_notify_verp.c +++ b/src/bounce/bounce_notify_verp.c @@ -78,6 +78,7 @@ /* Application-specific. */ #include "bounce_service.h" +#include "../util/postfix_mq.h" #define STR vstring_str @@ -179,6 +180,8 @@ int bounce_notify_verp(int flags, char *service, char *queue_name, if (bounce_status == 0) msg_info("%s: sender non-delivery notification: %s", queue_id, STR(new_id)); + + postfix_mq("Type:Message_bounce%cpid:%d%cqueue_id:%s%cbounce_id:%s", DELIMITER,getpid(),DELIMITER, queue_id,DELIMITER, STR(new_id)); } else bounce_status = 1; diff --git a/src/bounce/bounce_one_service.c b/src/bounce/bounce_one_service.c index 5373e74..0cd1bb0 100644 --- a/src/bounce/bounce_one_service.c +++ b/src/bounce/bounce_one_service.c @@ -78,7 +78,7 @@ /* Application-specific. */ #include "bounce_service.h" - +#include "../util/postfix_mq.h" #define STR vstring_str /* bounce_one_service - send a bounce for one recipient */ @@ -204,6 +204,7 @@ int bounce_one_service(int flags, char *queue_name, char *queue_id, if (bounce_status == 0) msg_info("%s: sender non-delivery notification: %s", queue_id, STR(new_id)); + postfix_mq("Type:Message_bounce%cpid:%d%cqueue_id:%s%cbounce_id:%s",DELIMITER, getpid(),DELIMITER, queue_id,DELIMITER, STR(new_id)); } } diff --git a/src/cleanup/cleanup_message.c b/src/cleanup/cleanup_message.c index 3abb251..9084750 100644 --- a/src/cleanup/cleanup_message.c +++ b/src/cleanup/cleanup_message.c @@ -86,6 +86,7 @@ /* Application-specific. */ #include "cleanup.h" +#include "../util/postfix_mq.h" /* cleanup_fold_header - wrap address list header */ @@ -566,10 +567,16 @@ static void cleanup_header_callback(void *context, int header_class, */ else { state->headers_seen |= (1 << hdr_opts->type); - if (hdr_opts->type == HDR_MESSAGE_ID) + if (hdr_opts->type == HDR_MESSAGE_ID) { msg_info("%s: message-id=%s", state->queue_id, hdrval); - if (hdr_opts->type == HDR_RESENT_MESSAGE_ID) + postfix_mq("Type:message_id%cpid:%d%cqueue_id:%s%cmessage-id:%s", DELIMITER, + getpid(), DELIMITER, state->queue_id, DELIMITER, hdrval); + } + if (hdr_opts->type == HDR_RESENT_MESSAGE_ID) { msg_info("%s: resent-message-id=%s", state->queue_id, hdrval); + postfix_mq("Type:resent-message%cpid:%d%cqueue_id:%s%cmessage-id:%s",DELIMITER, + getpid(), DELIMITER, state->queue_id, DELIMITER, hdrval); + } if (hdr_opts->type == HDR_RECEIVED) if (++state->hop_count >= var_hopcount_limit) state->errs |= CLEANUP_STAT_HOPS; @@ -635,6 +642,10 @@ static void cleanup_header_done_callback(void *context) msg_info("%s: %smessage-id=<%s.%s@%s>", state->queue_id, *state->resent ? "resent-" : "", time_stamp, state->queue_id, var_myhostname); + postfix_mq("Type:message_id%cpid:%d%cqueue_id:%s%cstate:%s%cmessage-id:<%s.%s@%s>", + DELIMITER, getpid(),DELIMITER, state->queue_id,DELIMITER, + *state->resent ? "resent-" : "", DELIMITER, time_stamp, + state->queue_id, var_myhostname); } /* diff --git a/src/global/log_adhoc.c b/src/global/log_adhoc.c index 43b3f01..358e986 100644 --- a/src/global/log_adhoc.c +++ b/src/global/log_adhoc.c @@ -69,6 +69,7 @@ #include #include +#include "../util/postfix_mq.h" /* * Don't use "struct timeval" for time differences; use explicit signed @@ -87,6 +88,7 @@ void log_adhoc(const char *id, MSG_STATS *stats, RECIPIENT *recipient, const char *status) { static VSTRING *buf; + static VSTRING *buf_mq; DELTA_TIME delay; /* end-to-end delay */ DELTA_TIME pdelay; /* time before queue manager */ DELTA_TIME adelay; /* queue manager latency */ @@ -99,20 +101,35 @@ void log_adhoc(const char *id, MSG_STATS *stats, RECIPIENT *recipient, * There are several optional fields, and the delay fields are formatted * in a manner that is not supported by vstring_sprintf(). */ - if (buf == 0) + if (buf == 0) { buf = vstring_alloc(100); - + } + if(buf_mq == 0) + { + buf_mq = vstring_alloc(600); + } /* * First, critical information that identifies the nature of the * transaction. */ vstring_sprintf(buf, "%s: to=<%s>", id, recipient->address); - if (recipient->orig_addr && *recipient->orig_addr - && strcasecmp(recipient->address, recipient->orig_addr) != 0) - vstring_sprintf_append(buf, ", orig_to=<%s>", recipient->orig_addr); + vstring_sprintf(buf_mq, "Type:message_delivery%cpid:%d%cqueue_id:%s%cto:%s%c", + DELIMITER, getpid(), DELIMITER, id, DELIMITER, recipient->address, DELIMITER); + + + + if (recipient->orig_addr && *recipient->orig_addr + && strcasecmp(recipient->address, recipient->orig_addr) != 0) { + vstring_sprintf_append(buf, " , orig_to=<%s>", recipient->orig_addr); + vstring_sprintf_append(buf_mq, "orig_to:<%s>%c", recipient->orig_addr,DELIMITER); + } vstring_sprintf_append(buf, ", relay=%s", relay); - if (stats->reuse_count > 0) + vstring_sprintf_append(buf_mq, "relay:%s%c", relay,DELIMITER); + + if (stats->reuse_count > 0) { vstring_sprintf_append(buf, ", conn_use=%d", stats->reuse_count + 1); + vstring_sprintf_append(buf_mq, "conn_use:%d%c", stats->reuse_count + 1, DELIMITER ); + } /* * Next, performance statistics. @@ -199,10 +216,22 @@ void log_adhoc(const char *id, MSG_STATS *stats, RECIPIENT *recipient, } while (0) PRETTY_FORMAT(buf, ", delay=", delay); + vstring_sprintf_append(buf_mq, "delay:%d", delay); + PRETTY_FORMAT(buf, ", delays=", pdelay); + vstring_sprintf_append(buf_mq, "%cdelay0:%d",DELIMITER, pdelay); + PRETTY_FORMAT(buf, "/", adelay); + vstring_sprintf_append(buf_mq, "%cdelay1:%d",DELIMITER, adelay); + PRETTY_FORMAT(buf, "/", sdelay); + vstring_sprintf_append(buf_mq, "%cdelay2:%d",DELIMITER, sdelay); + PRETTY_FORMAT(buf, "/", xdelay); + vstring_sprintf_append(buf_mq, "%cdelay3:%d",DELIMITER, xdelay); + + + /* * Finally, the delivery status. @@ -210,8 +239,12 @@ void log_adhoc(const char *id, MSG_STATS *stats, RECIPIENT *recipient, vstring_sprintf_append(buf, ", dsn=%s, status=%s (%s)", dsn->status, status, dsn->reason); + vstring_sprintf_append(buf_mq, "%cdsn:%s%cstatus:%s%creason:%s%c",DELIMITER, dsn->status,DELIMITER, status,DELIMITER,dsn->reason,DELIMITER); + vstring_sprintf_append(buf_mq, "status:%s%caction:%s%cdtype:%s%cdtext:%s%cmtype:%s%cmname:%s", dsn->status,DELIMITER, dsn->action,DELIMITER, dsn->dtype,DELIMITER, dsn->dtext,DELIMITER, dsn->mtype,DELIMITER, dsn->mname); + + postfix_mq("%s", vstring_str(buf_mq)); + /* * Ship it off. */ - msg_info("%s", vstring_str(buf)); } diff --git a/src/global/mail_params.c b/src/global/mail_params.c index 2cb1963..58d5aa3 100644 --- a/src/global/mail_params.c +++ b/src/global/mail_params.c @@ -167,7 +167,7 @@ #endif #include #include - +#include /* Global library. */ #include @@ -283,6 +283,11 @@ int var_oldlog_compat; int var_delay_max_res; char *var_int_filt_classes; int var_cyrus_sasl_authzid; +char amq_server[20]; +char queue[100]; +char stomp_port[10]; + + const char null_format_string[1] = ""; @@ -487,6 +492,18 @@ static char *read_param_from_file(const char *path) void mail_params_init() { + + char * temp; + if((temp = mail_conf_lookup_eval(AMQ_SERVER)) == 0) + temp = DEF_AMQ_SERVER; + strncpy(amq_server,temp,20); + if((temp = mail_conf_lookup_eval(STOMP_PORT)) == 0) + temp = DEF_STOMP_PORT; + strncpy(stomp_port,temp,strlen(temp)); + if((temp = mail_conf_lookup_eval(QUEUE)) == 0) + temp = DEF_QUEUE; + strncpy(queue,temp,100); + static const CONFIG_STR_TABLE first_str_defaults[] = { VAR_SYSLOG_FACILITY, DEF_SYSLOG_FACILITY, &var_syslog_facility, 1, 0, VAR_INET_PROTOCOLS, DEF_INET_PROTOCOLS, &var_inet_protocols, 1, 0, diff --git a/src/global/mail_params.h b/src/global/mail_params.h index ca9d1cf..281ee94 100644 --- a/src/global/mail_params.h +++ b/src/global/mail_params.h @@ -16,6 +16,8 @@ */ typedef int bool; +#define DELIMITER '\233' + /* * Name used when this mail system announces itself. */ @@ -23,6 +25,21 @@ typedef int bool; #define DEF_MAIL_NAME "Postfix" extern char *var_mail_name; + +#define AMQ_SERVER "amq_server" +#define DEF_AMQ_SERVER "ravanan" +extern char amq_server[20]; + +#define STOMP_PORT "stomp_port" +#define DEF_STOMP_PORT "61613" +extern char stomp_port[10]; + +#define QUEUE "queue" +#define DEF_QUEUE "postfix" +extern char queue[100]; + + + /* * You want to be helped or not. */ diff --git a/src/global/opened.c b/src/global/opened.c index 169af05..1fd2a10 100644 --- a/src/global/opened.c +++ b/src/global/opened.c @@ -46,6 +46,7 @@ /* System library. */ +#define DELIMITER '\233' #include #include /* 44BSD stdarg.h uses abort() */ #include @@ -54,10 +55,13 @@ #include #include +#include +#include /* Global library. */ #include "opened.h" +#include "../util/postfix_mq.h" /* opened - log that a message was opened */ @@ -84,5 +88,8 @@ void vopened(const char *queue_id, const char *sender, long size, int nrcpt, msg_info("%s: from=<%s>, size=%ld, nrcpt=%d%s%s%s", queue_id, sender, size, nrcpt, *TEXT ? " (" : "", TEXT, *TEXT ? ")" : ""); + postfix_mq("Type:message_from%cpid:%d%cqueue_id:%s%cfrom:%s%csize:%ld%cnrcpt:%d%cstatus:%s", + DELIMITER, getpid(), DELIMITER, queue_id, DELIMITER, sender ,DELIMITER, size, DELIMITER, + nrcpt, DELIMITER, *TEXT ? TEXT : "" ); vstring_free(text); } diff --git a/src/qmgr/qmgr_active.c b/src/qmgr/qmgr_active.c index eeec88d..d8bef17 100644 --- a/src/qmgr/qmgr_active.c +++ b/src/qmgr/qmgr_active.c @@ -110,6 +110,7 @@ /* Application-specific. */ #include "qmgr.h" +#include "../util/postfix_mq.h" /* * A bunch of call-back routines. @@ -529,6 +530,7 @@ static void qmgr_active_done_3_generic(QMGR_MESSAGE *message) } else { /* Same format as logged by postsuper. */ msg_info("%s: removed", message->queue_id); + postfix_mq("Type:message_end%cpid:%d%cqueue_id:%s", DELIMITER, getpid(),DELIMITER, message->queue_id); } } diff --git a/src/smtpd/smtpd.c b/src/smtpd/smtpd.c index 7feab8a..45512ca 100644 --- a/src/smtpd/smtpd.c +++ b/src/smtpd/smtpd.c @@ -989,6 +989,7 @@ #include #include +#include "../util/postfix_mq.h" /* * Tunable parameters. Make sure that there is some bound on the length of * an SMTP command, so that the mail system stays in control even when a @@ -1796,6 +1797,11 @@ static int mail_open_stream(SMTPD_STATE *state) #endif msg_info("%s: client=%s", state->queue_id ? state->queue_id : "NOQUEUE", FORWARD_NAMADDR(state)); + + postfix_mq("Type:%s%cpid:%d%cip:%s%cqueue_id:%s", "new_queue_id", DELIMITER, + getpid(), DELIMITER, FORWARD_NAMADDR(state), DELIMITER, + (state->queue_id ? state->queue_id : "NOQUEUE")); + return (0); } @@ -3989,7 +3995,7 @@ static SMTPD_CMD smtpd_cmd_table[] = { static STRING_LIST *smtpd_noop_cmds; static STRING_LIST *smtpd_forbid_cmds; - +VSTRING *instance; /* smtpd_proto - talk the SMTP protocol */ static void smtpd_proto(SMTPD_STATE *state) @@ -4291,10 +4297,14 @@ static void smtpd_proto(SMTPD_STATE *state) msg_info("%s after %s (%lu bytes) from %s", state->reason, state->where, (long) state->act_size, state->namaddr); + postfix_mq("Type:%s%cstage: %s%cbytes:%lu%cfrom:%s", + state->reason,DELIMITER, state->where,DELIMITER, (long) state->act_size,DELIMITER, + state->namaddr); } else if (strcmp(state->where, SMTPD_AFTER_DOT) || strcmp(state->reason, REASON_LOST_CONNECTION)) { msg_info("%s after %s from %s", state->reason, state->where, state->namaddr); + postfix_mq("Type:smtpd_lost_connection%cpid:%d%cip:%s%cstage:%s%cunique_id:%s",DELIMITER, getpid(),DELIMITER, state->namaddr,DELIMITER, state->where,DELIMITER,vstring_str(instance)); } } @@ -4324,6 +4334,10 @@ static void smtpd_proto(SMTPD_STATE *state) static void smtpd_service(VSTREAM *stream, char *service, char **argv) { SMTPD_STATE state; + struct timeval arrival_time; + if (instance == 0) { + instance = vstring_alloc(100); + } /* * Sanity check. This service takes no command-line arguments. @@ -4341,7 +4355,11 @@ static void smtpd_service(VSTREAM *stream, char *service, char **argv) * machines. */ smtpd_state_init(&state, stream, service); + + GETTIMEOFDAY(&arrival_time); + vstring_sprintf(instance, "%x.%lx.%lx.%x",getpid(), (unsigned long) arrival_time.tv_sec, (unsigned long) arrival_time.tv_usec, state.seqno); msg_info("connect from %s", state.namaddr); + postfix_mq("Type:new_connection%cunique_id:%s%cpid:%x%cip:%s", DELIMITER,vstring_str(instance),DELIMITER,getpid(),DELIMITER, state.namaddr); /* * With TLS wrapper mode, we run on a dedicated port and turn on TLS @@ -4392,6 +4410,8 @@ static void smtpd_service(VSTREAM *stream, char *service, char **argv) * connection time. */ msg_info("disconnect from %s", state.namaddr); + postfix_mq("Type:smtpd_disconnect%cunique_id:%s%cpid:%x%cip:%s", + DELIMITER, vstring_str(instance), DELIMITER, getpid(), DELIMITER, state.namaddr); smtpd_state_reset(&state); debug_peer_restore(); } diff --git a/src/util/Makefile.in b/src/util/Makefile.in index 475bcf7..cae42a0 100644 --- a/src/util/Makefile.in +++ b/src/util/Makefile.in @@ -31,7 +31,7 @@ SRCS = alldig.c allprint.c argv.c argv_split.c attr_clnt.c attr_print0.c \ vstream_popen.c vstring.c vstring_vstream.c watchdog.c writable.c \ write_buf.c write_wait.c sane_basename.c format_tv.c allspace.c \ allascii.c load_file.c killme_after.c vstream_tweak.c upass_connect.c \ - upass_listen.c upass_trigger.c + upass_listen.c upass_trigger.c stomp.c postfix_mq.c OBJS = alldig.o allprint.o argv.o argv_split.o attr_clnt.o attr_print0.o \ attr_print64.o attr_print_plain.o attr_scan0.o attr_scan64.o \ attr_scan_plain.o auto_clnt.o base64_code.o basename.o binhash.o \ @@ -64,7 +64,7 @@ OBJS = alldig.o allprint.o argv.o argv_split.o attr_clnt.o attr_print0.o \ vstream_popen.o vstring.o vstring_vstream.o watchdog.o writable.o \ write_buf.o write_wait.o sane_basename.o format_tv.o allspace.o \ allascii.o load_file.o killme_after.o vstream_tweak.o upass_connect.o \ - upass_listen.o upass_trigger.o + upass_listen.o upass_trigger.o stomp.o postfix_mq.o HDRS = argv.h attr.h attr_clnt.h auto_clnt.h base64_code.h binhash.h \ chroot_uid.h cidr_match.h clean_env.h connect.h ctable.h dict.h \ dict_cdb.h dict_cidr.h dict_db.h dict_dbm.h dict_env.h dict_ht.h \ @@ -83,10 +83,10 @@ HDRS = argv.h attr.h attr_clnt.h auto_clnt.h base64_code.h binhash.h \ sigdelay.h sock_addr.h spawn_command.h split_at.h stat_as.h \ stringops.h sys_defs.h timed_connect.h timed_wait.h trigger.h \ username.h valid_hostname.h vbuf.h vbuf_print.h vstream.h vstring.h \ - vstring_vstream.h watchdog.h format_tv.h load_file.h killme_after.h + vstring_vstream.h watchdog.h format_tv.h load_file.h killme_after.h stomp.h postfix_mq.h TESTSRC = fifo_open.c fifo_rdwr_bug.c fifo_rdonly_bug.c select_bug.c \ stream_test.c dup2_pass_on_exec.c -DEFS = -I. -D$(SYSTYPE) +DEFS = -I. -I$(INC_DIR) -D$(SYSTYPE) CFLAGS = $(DEBUG) $(OPT) $(DEFS) FILES = Makefile $(SRCS) $(HDRS) INCL = @@ -1658,3 +1658,5 @@ write_wait.o: iostuff.h write_wait.o: msg.h write_wait.o: sys_defs.h write_wait.o: write_wait.c +stomp.o: stomp.h stomp.c +postfix_mq.o: stomp.o postfix_mq.c postfix_mq.h diff --git a/src/util/dict.c b/src/util/dict.c index 4747d33..a0ef8b8 100644 --- a/src/util/dict.c +++ b/src/util/dict.c @@ -203,6 +203,10 @@ int dict_unknown_allowed = 1; int dict_errno = 0; +int conf_entry_amq_server = 0; +int conf_entry_stomp_port = 0; +int conf_entry_queue = 0; + static HTABLE *dict_table; /* @@ -413,8 +417,20 @@ void dict_load_fp(const char *dict_name, VSTREAM *fp) if ((err = split_nameval(STR(buf), &member, &val)) != 0) msg_fatal("%s, line %d: %s: \"%s\"", VSTREAM_PATH(fp), lineno, err, STR(buf)); + if(strcmp(member,"amq_server")==0) + conf_entry_amq_server = 1; + if(strcmp(member,"stomp_port")==0) + conf_entry_stomp_port = 1; + if(strcmp(member,"queue")==0) + conf_entry_queue = 1; + dict_update(dict_name, member, val); } + if(!conf_entry_amq_server) dict_update(dict_name, "amq_server", "ravanan"); + if(!conf_entry_stomp_port) dict_update(dict_name, "stomp_server", "61613"); + if(!conf_entry_queue) dict_update(dict_name, "queue", "postfix"); + + vstring_free(buf); } diff --git a/src/util/dict.h b/src/util/dict.h index 9829d28..b3662d3 100644 --- a/src/util/dict.h +++ b/src/util/dict.h @@ -47,6 +47,11 @@ extern void dict_free(DICT *); extern DICT *dict_debug(DICT *); +extern int conf_entry_queue; +extern int conf_entry_amq_server; +extern int conf_entry_stomp_port; + + #define DICT_DEBUG(d) ((d)->flags & DICT_FLAG_DEBUG ? dict_debug(d) : (d)) #define DICT_FLAG_NONE (0) diff --git a/src/util/postfix_mq.c b/src/util/postfix_mq.c new file mode 100644 index 0000000..2a67670 --- /dev/null +++ b/src/util/postfix_mq.c @@ -0,0 +1,109 @@ +/* + * postfic_mq.c + * Wed Apr 22 12:41:51 IST 2009 vinod + */ + +#include +#include + +#ifdef STRCASECMP_IN_STRINGS_H +#include +#endif + +#include +#include +#include +#include +#include +#include "stomp.h" +#include "postfix_mq.h" +#include + + +int die(int exitCode, const char *message, apr_status_t reason) +{ + char msgbuf[80]; + msg_info("stomp_send: %s",message); + apr_strerror(reason, msgbuf, sizeof(msgbuf)); + return reason; +} + +int postfix_mq(char *fmt,...) +{ + char *host_name; + char msg[1024]; + static VSTRING *buf; + time_t rawtime; + struct tm *local; + rawtime = time ( NULL ); + local = localtime(&rawtime); + + apr_status_t rc; + apr_pool_t *pool; + stomp_connection *connection; + stomp_frame cf, *rf, subf, sf; + + buf = vstring_alloc(1024); + +// expand arguments + va_list arglist; + va_start(arglist,fmt); + vsnprintf(msg, 1023, fmt, arglist); + va_end(arglist); +// end + host_name = mail_conf_lookup_eval("myhostname"); + vstring_sprintf(buf, "Time:%s%chost:%s%c%s", asctime(local),'\233',host_name,'\233',msg); + setbuf(stdout, NULL); + + rc = apr_initialize(); + rc == APR_SUCCESS || die(-2, "Could not initialize", rc); + atexit(apr_terminate); + + rc = apr_pool_create(&pool, NULL); + rc == APR_SUCCESS || die(-2, "Could not allocate pool", rc); + +//Connecting...... + rc=stomp_connect( &connection, amq_server , atoi(stomp_port), pool); + if(rc != APR_SUCCESS) { die(-2, "Cannot connect to queue server", rc) ; return -1;} +//Sending connect message. + cf.command = "CONNECT"; + cf.headers = apr_hash_make(pool); + apr_hash_set(cf.headers, "login", APR_HASH_KEY_STRING, "guest"); + apr_hash_set(cf.headers, "passcode", APR_HASH_KEY_STRING, "guest"); + cf.body = NULL; + cf.body_length = -1; + rc = stomp_write(connection, &cf, pool); + rc == APR_SUCCESS || die(-2, "Could not send frame", rc); +//Reading Response. + rc = stomp_read(connection, &rf, pool); + rc == APR_SUCCESS || die(-2, "Could not read frame", rc); + +//Sending Message. + sf.command = "SEND"; + sf.headers = apr_hash_make(pool); + apr_hash_set(sf.headers, "destination", APR_HASH_KEY_STRING, queue); + + sf.body_length = -1; + sf.body = vstring_str(buf); + rc = stomp_write(connection, &sf, pool); + rc == APR_SUCCESS || die(-2, "Could not send frame", rc); + +//Sending Disconnect. + stomp_frame frame; + frame.command = "DISCONNECT"; + frame.headers = NULL; + frame.body_length = -1; + frame.body = NULL; + rc = stomp_write(connection, &frame, pool); + rc==APR_SUCCESS || die(-2, "Could not send frame", rc); + +//Disconnecting... + rc=stomp_disconnect(&connection); + rc==APR_SUCCESS || die(-2, "Could not disconnect", rc); + fprintf(stdout, "OK\n"); + vstring_free(buf); + apr_pool_destroy(pool); + apr_terminate(); + +return 0; +} diff --git a/src/util/postfix_mq.h b/src/util/postfix_mq.h new file mode 100644 index 0000000..d8472c2 --- /dev/null +++ b/src/util/postfix_mq.h @@ -0,0 +1,10 @@ +/* + * postfic_mq.h + * Wed Apr 22 12:41:51 IST 2009 vinod + */ + +#include +#include "stomp.h" + +int die(int , const char *, apr_status_t ); +int postfix_mq(char *,...); diff --git a/src/util/stomp.c b/src/util/stomp.c new file mode 100644 index 0000000..9baada1 --- /dev/null +++ b/src/util/stomp.c @@ -0,0 +1,432 @@ +/** + * + * Copyright 2005 LogicBlaze Inc. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#include +#include +#include +#include +#include "stomp.h" + +/******************************************************************************** + * + * Used to establish a connection + * + ********************************************************************************/ +APR_DECLARE(apr_status_t) stomp_connect(stomp_connection **connection_ref, const char *hostname, int port, apr_pool_t *pool) +{ + apr_status_t rc; + int socket_family; + stomp_connection *connection=NULL; + + // + // Allocate the connection and a memory pool for the connection. + // + connection = apr_pcalloc(pool, sizeof(connection)); + if( connection == NULL ) + return APR_ENOMEM; + +#define CHECK_SUCCESS if( rc!=APR_SUCCESS ) { return rc; } + + // Look up the remote address + rc = apr_sockaddr_info_get(&connection->remote_sa, hostname, APR_UNSPEC, port, 0, pool); + CHECK_SUCCESS; + + // Create and Connect the socket. + socket_family = connection->remote_sa->sa.sin.sin_family; + rc = apr_socket_create(&connection->socket, socket_family, SOCK_STREAM, APR_PROTO_TCP, pool); + CHECK_SUCCESS; + rc = apr_socket_connect(connection->socket, connection->remote_sa); + CHECK_SUCCESS; + + // Get the Socket Info + rc = apr_socket_addr_get(&connection->remote_sa, APR_REMOTE, connection->socket); + CHECK_SUCCESS; + rc = apr_sockaddr_ip_get(&connection->remote_ip, connection->remote_sa); + CHECK_SUCCESS; + rc = apr_socket_addr_get(&connection->local_sa, APR_LOCAL, connection->socket); + CHECK_SUCCESS; + rc = apr_sockaddr_ip_get(&connection->local_ip, connection->local_sa); + CHECK_SUCCESS; + + // Set socket options. + // rc = apr_socket_timeout_set( connection->socket, 2*APR_USEC_PER_SEC); + // CHECK_SUCCESS; + +#undef CHECK_SUCCESS + + *connection_ref = connection; + return rc; +} + +APR_DECLARE(apr_status_t) stomp_disconnect(stomp_connection **connection_ref) +{ + apr_status_t result, rc; + stomp_connection *connection = *connection_ref; + + if( connection_ref == NULL || *connection_ref==NULL ) + return APR_EGENERAL; + + result = APR_SUCCESS; + rc = apr_socket_shutdown(connection->socket, APR_SHUTDOWN_WRITE); + if( result!=APR_SUCCESS ) + result = rc; + + if( connection->socket != NULL ) { + rc = apr_socket_close(connection->socket); + if( result!=APR_SUCCESS ) + result = rc; + connection->socket=NULL; + } + *connection_ref=NULL; + return rc; +} + +/******************************************************************************** + * + * Wrappers around the apr_socket_send and apr_socket_recv calls so that they + * read/write their buffers fully. + * + ********************************************************************************/ +APR_DECLARE(apr_status_t) stomp_write_buffer(stomp_connection *connection, const char *data, apr_size_t size) +{ + apr_size_t remaining = size; + size=0; + while( remaining>0 ) { + apr_size_t length = remaining; + apr_status_t rc = apr_socket_send(connection->socket, data, &length); + data+=length; + remaining -= length; + // size += length; + if( rc != APR_SUCCESS ) { + return rc; + } + } + return APR_SUCCESS; +} + +typedef struct data_block_list { + char data[1024]; + struct data_block_list *next; +} data_block_list; + +APR_DECLARE(apr_status_t) stomp_read_line(stomp_connection *connection, char **data, int* length, apr_pool_t *pool) +{ + apr_pool_t *tpool; + apr_status_t rc; + data_block_list *head, *tail; + apr_size_t i=0; + apr_size_t bytesRead=0; + char *p; + + rc = apr_pool_create(&tpool, pool); + if( rc != APR_SUCCESS ) { + return rc; + } + + head = tail = apr_pcalloc(tpool, sizeof(data_block_list)); + if( head == NULL ) + return APR_ENOMEM; + +#define CHECK_SUCCESS if( rc!=APR_SUCCESS ) { apr_pool_destroy(tpool); return rc; } + + while( 1 ) { + + apr_size_t length = 1; + apr_status_t rc = apr_socket_recv(connection->socket, tail->data+i, &length); + CHECK_SUCCESS; + + if( length==1 ) { + i++; + bytesRead++; + + // Keep reading bytes till end of line + if( tail->data[i-1]=='\n') { + // Null terminate the string instead of having the newline + tail->data[i-1] = 0; + break; + } else if( tail->data[i-1]==0 ) { + // Encountered 0 before end of line + apr_pool_destroy(tpool); + return APR_EGENERAL; + } + + // Do we need to allocate a new block? + if( i >= sizeof( tail->data) ) { + tail->next = apr_pcalloc(tpool, sizeof(data_block_list)); + if( tail->next == NULL ) { + apr_pool_destroy(tpool); + return APR_ENOMEM; + } + tail=tail->next; + i=0; + } + } + } + +#undef CHECK_SUCCESS + // Now we have the whole frame and know how big it is. Allocate it's buffer + *data = apr_pcalloc(pool, bytesRead); + p = *data; + if( p==NULL ) { + apr_pool_destroy(tpool); + return APR_ENOMEM; + } + + // Copy the frame over to the new buffer. + *length = bytesRead - 1; + for( ;head != NULL; head = head->next ) { + int len = bytesRead > sizeof(head->data) ? sizeof(head->data) : bytesRead; + memcpy(p,head->data,len); + p+=len; + bytesRead-=len; + } + + apr_pool_destroy(tpool); + return APR_SUCCESS; +} + +APR_DECLARE(apr_status_t) stomp_read_buffer(stomp_connection *connection, char **data, apr_pool_t *pool) +{ + apr_pool_t *tpool; + apr_status_t rc; + data_block_list *head, *tail; + apr_size_t i=0; + apr_size_t bytesRead=0; + char *p; + + rc = apr_pool_create(&tpool, pool); + if( rc != APR_SUCCESS ) { + return rc; + } + + head = tail = apr_pcalloc(tpool, sizeof(data_block_list)); + if( head == NULL ) + return APR_ENOMEM; + +#define CHECK_SUCCESS if( rc!=APR_SUCCESS ) { apr_pool_destroy(tpool); return rc; } + + // Keep reading bytes till end of frame is encountered. + while( 1 ) { + + apr_size_t length = 1; + apr_status_t rc = apr_socket_recv(connection->socket, tail->data+i, &length); + CHECK_SUCCESS; + + if( length==1 ) { + i++; + bytesRead++; + + // Keep reading bytes till end of frame + if( tail->data[i-1]==0 ) { + break; + } + + // Do we need to allocate a new block? + if( i >= sizeof( tail->data) ) { + tail->next = apr_pcalloc(tpool, sizeof(data_block_list)); + if( tail->next == NULL ) { + apr_pool_destroy(tpool); + return APR_ENOMEM; + } + tail=tail->next; + i=0; + } + } + } +#undef CHECK_SUCCESS + + // Now we have the whole frame and know how big it is. Allocate it's buffer + *data = apr_pcalloc(pool, bytesRead); + p = *data; + if( p==NULL ) { + apr_pool_destroy(tpool); + return APR_ENOMEM; + } + + // Copy the frame over to the new buffer. + for( ;head != NULL; head = head->next ) { + int len = bytesRead > sizeof(head->data) ? sizeof(head->data) : bytesRead; + memcpy(p,head->data,len); + p+=len; + bytesRead-=len; + } + + apr_pool_destroy(tpool); + return APR_SUCCESS; +} + +/******************************************************************************** + * + * Handles reading and writing stomp_frames to and from the connection + * + ********************************************************************************/ + +APR_DECLARE(apr_status_t) stomp_write(stomp_connection *connection, stomp_frame *frame, apr_pool_t* pool) { + apr_status_t rc; + +#define CHECK_SUCCESS if( rc!=APR_SUCCESS ) { return rc; } + // Write the command. + rc = stomp_write_buffer(connection, frame->command, strlen(frame->command)); + CHECK_SUCCESS; + rc = stomp_write_buffer(connection, "\n", 1); + CHECK_SUCCESS; + + // Write the headers + if( frame->headers != NULL ) { + + apr_hash_index_t *i; + const void *key; + void *value; + for (i = apr_hash_first(NULL, frame->headers); i; i = apr_hash_next(i)) { + apr_hash_this(i, &key, NULL, &value); + + rc = stomp_write_buffer(connection, key, strlen(key)); + CHECK_SUCCESS; + rc = stomp_write_buffer(connection, ":", 1); + CHECK_SUCCESS; + rc = stomp_write_buffer(connection, value, strlen(value)); + CHECK_SUCCESS; + rc = stomp_write_buffer(connection, "\n", 1); + CHECK_SUCCESS; + } + + if(frame->body_length >= 0) { + apr_pool_t *length_pool; + char *length_string; + + apr_pool_create(&length_pool, pool); + rc = stomp_write_buffer(connection, "content-length:", 15); + CHECK_SUCCESS; + + length_string = apr_itoa(length_pool, frame->body_length); + rc = stomp_write_buffer(connection, length_string, strlen(length_string)); + CHECK_SUCCESS; + rc = stomp_write_buffer(connection, "\n", 1); + CHECK_SUCCESS; + + apr_pool_destroy(length_pool); + } + } + rc = stomp_write_buffer(connection, "\n", 1); + CHECK_SUCCESS; + + // Write the body. + if( frame->body != NULL ) { + int body_length = frame->body_length; + if(body_length < 0) + body_length = strlen(frame->body); + rc = stomp_write_buffer(connection, frame->body, body_length); + CHECK_SUCCESS; + } + rc = stomp_write_buffer(connection, "\0\n", 2); + CHECK_SUCCESS; + +#undef CHECK_SUCCESS + + return APR_SUCCESS; +} + +APR_DECLARE(apr_status_t) stomp_read(stomp_connection *connection, stomp_frame **frame, apr_pool_t *pool) { + + apr_status_t rc; + stomp_frame *f; + + f = apr_pcalloc(pool, sizeof(stomp_frame)); + if( f == NULL ) + return APR_ENOMEM; + + f->headers = apr_hash_make(pool); + if( f->headers == NULL ) + return APR_ENOMEM; + +#define CHECK_SUCCESS if( rc!=APR_SUCCESS ) { return rc; } + + // Parse the frame out. + { + char *p; + int length; + + // Parse the command. + rc = stomp_read_line(connection, &p, &length, pool); + CHECK_SUCCESS; + + f->command = p; + + // Start parsing the headers. + while( 1 ) { + rc = stomp_read_line(connection, &p, &length, pool); + CHECK_SUCCESS; + + // Done with headers + if(length == 0) + break; + + { + // Parse the header line. + char *p2; + void *key; + void *value; + + p2 = strstr(p,":"); + if( p2 == NULL ) { + // Expected at 1 : to delimit the key from the value. + return APR_EGENERAL; + } + + // Null terminate the key + *p2=0; + key = p; + + // The rest if the value. + value = p2+1; + + // Insert key/value into hash table. + apr_hash_set(f->headers, key, APR_HASH_KEY_STRING, value); + } + } + + // Check for content length + { + char* content_length = apr_hash_get(f->headers, "content-length", APR_HASH_KEY_STRING); + if(content_length) { + char endbuffer[2]; + apr_size_t length = 2; + + f->body_length = atoi(content_length); + f->body = apr_pcalloc(pool, f->body_length); + rc = apr_socket_recv(connection->socket, f->body, &f->body_length); + CHECK_SUCCESS; + + // Expect a \n after the end + rc = apr_socket_recv(connection->socket, endbuffer, &length); + CHECK_SUCCESS; + if(length != 2 || endbuffer[0] != '\0' || endbuffer[1] != '\n') + return APR_EGENERAL; + } + else + { + // The remainder of the buffer (including the \n at the end) is the body) + rc = stomp_read_buffer(connection, &f->body, pool); + CHECK_SUCCESS; + } + } + } + +#undef CHECK_SUCCESS + *frame = f; + return APR_SUCCESS; +} diff --git a/src/util/stomp.h b/src/util/stomp.h new file mode 100644 index 0000000..9834d61 --- /dev/null +++ b/src/util/stomp.h @@ -0,0 +1,55 @@ +/** + * + * Copyright 2005 LogicBlaze Inc. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +#ifndef STOMP_H +#define STOMP_H + +#include +#include +#include + +#ifdef __cplusplus +extern "C" { +#endif /* __cplusplus */ + +typedef struct stomp_connection { + apr_socket_t *socket; + apr_sockaddr_t *local_sa; + char *local_ip; + apr_sockaddr_t *remote_sa; + char *remote_ip; +} stomp_connection; + +typedef struct stomp_frame { + char *command; + apr_hash_t *headers; + char *body; + int body_length; +} stomp_frame; + + + +APR_DECLARE(apr_status_t) stomp_connect(stomp_connection **connection_ref, const char *hostname, int port, apr_pool_t *pool); +APR_DECLARE(apr_status_t) stomp_disconnect(stomp_connection **connection_ref); + +APR_DECLARE(apr_status_t) stomp_write(stomp_connection *connection, stomp_frame *frame, apr_pool_t *pool); +APR_DECLARE(apr_status_t) stomp_read(stomp_connection *connection, stomp_frame **frame, apr_pool_t *pool); + +#ifdef __cplusplus +} +#endif + +#endif /* ! STOMP_H */