rendered paste bodydiff --git a/largefile64.patch b/largefile64.patchnew file mode 100644index 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/makedefsindex 0585fa0..cfba0e9 100644--- a/makedefs+++ b/makedefs@@ -265,7 +265,7 @@ case "$SYSTEM.$RELEASE" in # CCARGS="$CCARGS -DHAS_DBM -DPATH_NDBM_H='<gdbm/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 /libdiff --git a/src/bounce/bounce_notify_service.c b/src/bounce/bounce_notify_service.cindex 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.cindex 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.cindex 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.cindex 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.cindex 43b3f01..358e986 100644--- a/src/global/log_adhoc.c+++ b/src/global/log_adhoc.c@@ -69,6 +69,7 @@ #include <log_adhoc.h> #include <mail_params.h>+#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.cindex 2cb1963..58d5aa3 100644--- a/src/global/mail_params.c+++ b/src/global/mail_params.c@@ -167,7 +167,7 @@ #endif #include <inet_proto.h> #include <vstring_vstream.h>-+#include <dict.h> /* Global library. */ #include <mynetworks.h>@@ -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.hindex 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.cindex 169af05..1fd2a10 100644--- a/src/global/opened.c+++ b/src/global/opened.c@@ -46,6 +46,7 @@ /* System library. */ +#define DELIMITER '\233' #include <sys_defs.h> #include <stdlib.h> /* 44BSD stdarg.h uses abort() */ #include <stdarg.h>@@ -54,10 +55,13 @@ #include <msg.h> #include <vstring.h>+#include <sys/types.h>+#include <unistd.h> /* 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.cindex 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.cindex 7feab8a..45512ca 100644--- a/src/smtpd/smtpd.c+++ b/src/smtpd/smtpd.c@@ -989,6 +989,7 @@ #include <smtpd_proxy.h> #include <smtpd_milter.h> +#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.inindex 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.hdiff --git a/src/util/dict.c b/src/util/dict.cindex 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.hindex 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.cnew file mode 100644index 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 <sys_defs.h>+#include <string.h>++#ifdef STRCASECMP_IN_STRINGS_H+#include <strings.h>+#endif++#include <stdlib.h>+#include <time.h>+#include <strings.h>+#include <vstring.h>+#include <mail_params.h>+#include "stomp.h"+#include "postfix_mq.h"+#include <dict.h>+++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.hnew file mode 100644index 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 <stdlib.h>+#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.cnew file mode 100644index 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 <stdlib.h>+#include <string.h>+#include <apr-1.0/apr.h>+#include <apr-1.0/apr_strings.h>+#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.hnew file mode 100644index 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 <apr-1.0/apr_general.h>+#include <apr-1.0/apr_network_io.h>+#include <apr-1.0/apr_hash.h>++#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 */