#include <sys_defs.h>
#include <sys/socket.h>
#include <sys/wait.h>
#include <netinet/in.h>
#include <sys/un.h>
#include <stdlib.h>
#include <unistd.h>
#include <signal.h>
#include <errno.h>
#include <string.h>
#include <msg.h>
#include <msg_vstream.h>
#include <vstring.h>
#include <vstream.h>
#include <get_hostname.h>
#include <split_at.h>
#include <connect.h>
#include <mymalloc.h>
#include <events.h>
#include <find_inet.h>
#include <iostuff.h>
#include <netstring.h>
#include <mail_date.h>
#include <qmqp_proto.h>
typedef struct SESSION {
int xfer_count;
int rcpt_done;
int rcpt_count;
VSTREAM *stream;
int connect_count;
struct SESSION *next;
} SESSION;
static SESSION *last_session;
static VSTRING *buffer;
static int var_line_limit = 10240;
static int var_timeout = 300;
static const char *var_myhostname;
static int session_count;
static int message_count = 1;
static struct sockaddr_in sin;
#undef sun
static struct sockaddr_un sun;
static struct sockaddr *sa;
static int sa_length;
static int recipients = 1;
static char *defaddr;
static char *recipient;
static char *sender;
static int message_length = 1024;
static int count = 0;
static int counter = 0;
static int connect_count = 1;
static int random_delay = 0;
static int fixed_delay = 0;
static const char *mydate;
static int mypid;
static void enqueue_connect(SESSION *);
static void start_connect(SESSION *);
static void connect_done(int, char *);
static void send_data(SESSION *);
static void receive_reply(int, char *);
static VSTRING *message_buffer;
static VSTRING *sender_buffer;
static VSTRING *recipient_buffer;
#define STR(x) vstring_str(x)
#define LEN(x) VSTRING_LEN(x)
static int random_interval(int interval)
{
return (rand() % (interval + 1));
}
static int socket_error(int sock)
{
int error;
SOCKOPT_SIZE error_len;
error = 0;
error_len = sizeof(error);
if (getsockopt(sock, SOL_SOCKET, SO_ERROR, (char *) &error, &error_len) < 0)
return (-1);
if (error) {
errno = error;
return (-1);
}
return (0);
}
static char *exception_text(int except)
{
;
switch (except) {
case NETSTRING_ERR_EOF:
return ("lost connection");
case NETSTRING_ERR_TIME:
return ("timeout");
case NETSTRING_ERR_FORMAT:
return ("netstring format error");
case NETSTRING_ERR_SIZE:
return ("netstring size exceeds limit");
default:
msg_panic("exception_text: unknown exception %d", except);
}
}
static void startup(SESSION *session)
{
if (message_count-- <= 0) {
myfree((char *) session);
session_count--;
return;
}
enqueue_connect(session);
}
static void start_event(int unused_event, char *context)
{
SESSION *session = (SESSION *) context;
startup(session);
}
static void start_another(SESSION *session)
{
if (random_delay > 0) {
event_request_timer(start_event, (char *) session,
random_interval(random_delay));
} else if (fixed_delay > 0) {
event_request_timer(start_event, (char *) session, fixed_delay);
} else {
startup(session);
}
}
static void enqueue_connect(SESSION *session)
{
session->next = 0;
if (last_session == 0) {
last_session = session;
start_connect(session);
} else {
last_session->next = session;
last_session = session;
}
}
static void dequeue_connect(SESSION *session)
{
if (session == last_session) {
if (session->next != 0)
msg_panic("dequeue_connect: queue ends after last");
last_session = 0;
} else {
if (session->next == 0)
msg_panic("dequeue_connect: queue ends before last");
start_connect(session->next);
}
}
static void fail_connect(SESSION *session)
{
if (session->connect_count-- == 1)
msg_fatal("connect: %m");
msg_warn("connect: %m");
event_disable_readwrite(vstream_fileno(session->stream));
vstream_fclose(session->stream);
session->stream = 0;
#ifdef MISSING_USLEEP
doze(10);
#else
usleep(10);
#endif
start_connect(session);
}
static void start_connect(SESSION *session)
{
int fd;
if ((fd = socket(sa->sa_family, SOCK_STREAM, 0)) < 0)
msg_fatal("socket: %m");
(void) non_blocking(fd, NON_BLOCKING);
session->stream = vstream_fdopen(fd, O_RDWR);
event_enable_write(fd, connect_done, (char *) session);
netstring_setup(session->stream, var_timeout);
if (connect(fd, sa, sa_length) < 0 && errno != EINPROGRESS)
fail_connect(session);
}
static void connect_done(int unused_event, char *context)
{
SESSION *session = (SESSION *) context;
int fd = vstream_fileno(session->stream);
if (socket_error(fd) < 0) {
fail_connect(session);
} else {
dequeue_connect(session);
non_blocking(fd, BLOCKING);
event_disable_readwrite(fd);
send_data(session);
}
}
static void send_data(SESSION *session)
{
int fd = vstream_fileno(session->stream);
int except;
if ((except = vstream_setjmp(session->stream)) != 0)
msg_fatal("%s while sending message", exception_text(except));
netstring_put_multi(session->stream,
STR(message_buffer), LEN(message_buffer),
STR(sender_buffer), LEN(sender_buffer),
STR(recipient_buffer), LEN(recipient_buffer),
0);
netstring_fflush(session->stream);
event_enable_read(fd, receive_reply, (char *) session);
}
static void receive_reply(int unused_event, char *context)
{
SESSION *session = (SESSION *) context;
int except;
if ((except = vstream_setjmp(session->stream)) != 0)
msg_fatal("%s while receiving server reply", exception_text(except));
netstring_get(session->stream, buffer, var_line_limit);
if (msg_verbose)
vstream_printf("<< %.*s\n", LEN(buffer), STR(buffer));
if (STR(buffer)[0] != QMQP_STAT_OK)
msg_fatal("%s error: %.*s",
STR(buffer)[0] == QMQP_STAT_RETRY ? "recoverable" :
STR(buffer)[0] == QMQP_STAT_HARD ? "unrecoverable" :
"unknown", LEN(buffer) - 1, STR(buffer) + 1);
if (count) {
counter++;
vstream_printf("%d\r", counter);
vstream_fflush(VSTREAM_OUT);
}
event_disable_readwrite(vstream_fileno(session->stream));
vstream_fclose(session->stream);
session->stream = 0;
start_another(session);
}
static void usage(char *myname)
{
msg_fatal("usage: %s -s sess -l msglen -m msgs -c -C count -f from -t to -R delay -v -w delay host[:port]", myname);
}
int main(int argc, char **argv)
{
SESSION *session;
char *host;
char *port;
char *path;
int path_len;
int sessions = 1;
int ch;
int n;
int i;
signal(SIGPIPE, SIG_IGN);
msg_vstream_init(argv[0], VSTREAM_ERR);
while ((ch = GETOPT(argc, argv, "cC:f:l:m:r:R:s:t:vw:")) > 0) {
switch (ch) {
case 'c':
count++;
break;
case 'C':
if ((connect_count = atoi(optarg)) <= 0)
usage(argv[0]);
break;
case 'f':
sender = optarg;
break;
case 'l':
if ((message_length = atoi(optarg)) <= 0)
usage(argv[0]);
break;
case 'm':
if ((message_count = atoi(optarg)) <= 0)
usage(argv[0]);
break;
case 'r':
if ((recipients = atoi(optarg)) <= 0)
usage(argv[0]);
break;
case 'R':
if (fixed_delay > 0 || (random_delay = atoi(optarg)) <= 0)
usage(argv[0]);
break;
case 's':
if ((sessions = atoi(optarg)) <= 0)
usage(argv[0]);
break;
case 't':
recipient = optarg;
break;
case 'v':
msg_verbose++;
break;
case 'w':
if (random_delay > 0 || (fixed_delay = atoi(optarg)) <= 0)
usage(argv[0]);
break;
default:
usage(argv[0]);
}
}
if (argc - optind != 1)
usage(argv[0]);
if (random_delay > 0)
srand(getpid());
if (strncmp(argv[optind], "unix:", 5) == 0) {
path = argv[optind] + 5;
path_len = strlen(path);
if (path_len >= (int) sizeof(sun.sun_path))
msg_fatal("unix-domain name too long: %s", path);
memset((char *) &sun, 0, sizeof(sun));
sun.sun_family = AF_UNIX;
#ifdef HAS_SUN_LEN
sun.sun_len = path_len + 1;
#endif
memcpy(sun.sun_path, path, path_len);
sa = (struct sockaddr *) & sun;
sa_length = sizeof(sun);
} else {
if (strncmp(argv[optind], "inet:", 5) == 0)
argv[optind] += 5;
if ((port = split_at(host = argv[optind], ':')) == 0 || *port == 0)
port = "628";
memset((char *) &sin, 0, sizeof(sin));
sin.sin_family = AF_INET;
sin.sin_addr.s_addr = find_inet_addr(host);
sin.sin_port = find_inet_port(port, "tcp");
sa = (struct sockaddr *) & sin;
sa_length = sizeof(sin);
}
buffer = vstring_alloc(100);
var_myhostname = get_hostname();
if (sender == 0 || recipient == 0) {
vstring_sprintf(buffer, "foo@%s", var_myhostname);
defaddr = mystrdup(vstring_str(buffer));
if (sender == 0)
sender = defaddr;
if (recipient == 0)
recipient = defaddr;
}
mydate = mail_date(time((time_t *) 0));
mypid = getpid();
message_buffer = vstring_alloc(message_length + 200);
vstring_sprintf(buffer,
"From: <%s>\nTo: <%s>\nDate: %s\nMessage-Id: <%d@%s>\n\n",
sender, recipient, mydate, mypid, var_myhostname);
for (n = 1; LEN(buffer) < message_length; n++) {
for (i = 0; i < n && i < 79; i++)
VSTRING_ADDCH(buffer, 'X');
VSTRING_ADDCH(buffer, '\n');
}
STR(buffer)[message_length - 1] = '\n';
netstring_memcpy(message_buffer, STR(buffer), message_length);
n = strlen(sender);
sender_buffer = vstring_alloc(n);
netstring_memcpy(sender_buffer, sender, n);
if (recipients == 1) {
n = strlen(recipient);
recipient_buffer = vstring_alloc(n);
netstring_memcpy(recipient_buffer, recipient, n);
} else {
recipient_buffer = vstring_alloc(100);
for (n = 0; n < recipients; n++) {
vstring_sprintf(buffer, "%d%s", n, recipient);
netstring_memcat(recipient_buffer, STR(buffer), LEN(buffer));
}
}
while (sessions-- > 0) {
session = (SESSION *) mymalloc(sizeof(*session));
session->stream = 0;
session->xfer_count = 0;
session->connect_count = connect_count;
session->next = 0;
session_count++;
startup(session);
}
for (;;) {
event_loop(-1);
if (session_count <= 0 && message_count <= 0) {
if (count) {
VSTREAM_PUTC('\n', VSTREAM_OUT);
vstream_fflush(VSTREAM_OUT);
}
exit(0);
}
}
}