diff options
Diffstat (limited to '')
| -rw-r--r-- | tests/miniamqpsrvr.c | 658 |
1 files changed, 658 insertions, 0 deletions
diff --git a/tests/miniamqpsrvr.c b/tests/miniamqpsrvr.c new file mode 100644 index 0000000..5aedacd --- /dev/null +++ b/tests/miniamqpsrvr.c @@ -0,0 +1,658 @@ +/* a very simplistic tcp receiver for the rsyslog testbench. + * + * Author Philippe Duveau + * + * This file is contribution of the rsyslog project. + * + * 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 + * -or- + * see COPYING.ASL20 in the source distribution + * + * 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 "config.h" +#include <stdio.h> +#include <stdlib.h> +#include <string.h> +#include <sys/types.h> +#include <sys/stat.h> +#include <sys/socket.h> +#include <sys/time.h> +#include <fcntl.h> +#include <unistd.h> +#include <arpa/inet.h> +#include <inttypes.h> +#include <pthread.h> +#include <signal.h> +#include <errno.h> +#if defined(__FreeBSD__) +#include <netinet/in.h> +#endif + +#include "rsyslog.h" + + +#include <amqp.h> +#include <amqp_framing.h> + +#define AMQP_STARTING ((uchar)0x10) +#define AMQP_STOP ((uchar)0x00) + +#define AMQP_BEHAVIOR_STANDARD 1 +#define AMQP_BEHAVIOR_NOEXCH 2 +#define AMQP_BEHAVIOR_DECEXCH 3 +#define AMQP_BEHAVIOR_BADEXCH 4 + +uchar connection_start[487] = { + 0x01, 0x00, 0x00, 0x00, 0x00, 0x01, 0xDF, 0x00, 0x0A, 0x00, 0x0A, 0x00, 0x09, 0x00, 0x00, 0x01, + 0xBA, 0x0C, 0x63, 0x61, 0x70, 0x61, 0x62, 0x69, 0x6C, 0x69, 0x74, 0x69, 0x65, 0x73, 0x46, 0x00, + 0x00, 0x00, 0xC7, 0x12, 0x70, 0x75, 0x62, 0x6C, 0x69, 0x73, 0x68, 0x65, 0x72, 0x5F, 0x63, 0x6F, + 0x6E, 0x66, 0x69, 0x72, 0x6D, 0x73, 0x74, 0x01, 0x1A, 0x65, 0x78, 0x63, 0x68, 0x61, 0x6E, 0x67, + 0x65, 0x5F, 0x65, 0x78, 0x63, 0x68, 0x61, 0x6E, 0x67, 0x65, 0x5F, 0x62, 0x69, 0x6E, 0x64, 0x69, + 0x6E, 0x67, 0x73, 0x74, 0x01, 0x0A, 0x62, 0x61, 0x73, 0x69, 0x63, 0x2E, 0x6E, 0x61, 0x63, 0x6B, + 0x74, 0x01, 0x16, 0x63, 0x6F, 0x6E, 0x73, 0x75, 0x6D, 0x65, 0x72, 0x5F, 0x63, 0x61, 0x6E, 0x63, + 0x65, 0x6C, 0x5F, 0x6E, 0x6F, 0x74, 0x69, 0x66, 0x79, 0x74, 0x01, 0x12, 0x63, 0x6F, 0x6E, 0x6E, + 0x65, 0x63, 0x74, 0x69, 0x6F, 0x6E, 0x2E, 0x62, 0x6C, 0x6F, 0x63, 0x6B, 0x65, 0x64, 0x74, 0x01, + 0x13, 0x63, 0x6F, 0x6E, 0x73, 0x75, 0x6D, 0x65, 0x72, 0x5F, 0x70, 0x72, 0x69, 0x6F, 0x72, 0x69, + 0x74, 0x69, 0x65, 0x73, 0x74, 0x01, 0x1C, 0x61, 0x75, 0x74, 0x68, 0x65, 0x6E, 0x74, 0x69, 0x63, + 0x61, 0x74, 0x69, 0x6F, 0x6E, 0x5F, 0x66, 0x61, 0x69, 0x6C, 0x75, 0x72, 0x65, 0x5F, 0x63, 0x6C, + 0x6F, 0x73, 0x65, 0x74, 0x01, 0x10, 0x70, 0x65, 0x72, 0x5F, 0x63, 0x6F, 0x6E, 0x73, 0x75, 0x6D, + 0x65, 0x72, 0x5F, 0x71, 0x6F, 0x73, 0x74, 0x01, 0x0F, 0x64, 0x69, 0x72, 0x65, 0x63, 0x74, 0x5F, + 0x72, 0x65, 0x70, 0x6C, 0x79, 0x5F, 0x74, 0x6F, 0x74, 0x01, 0x0C, 0x63, 0x6C, 0x75, 0x73, 0x74, + 0x65, 0x72, 0x5F, 0x6E, 0x61, 0x6D, 0x65, 0x53, 0x00, 0x00, 0x00, 0x0D, 0x72, 0x61, 0x62, 0x62, + 0x69, 0x74, 0x40, 0x73, 0x65, 0x72, 0x76, 0x65, 0x72, 0x09, 0x63, 0x6F, 0x70, 0x79, 0x72, 0x69, + 0x67, 0x68, 0x74, 0x53, 0x00, 0x00, 0x00, 0x2E, 0x43, 0x6F, 0x70, 0x79, 0x72, 0x69, 0x67, 0x68, + 0x74, 0x20, 0x28, 0x43, 0x29, 0x20, 0x32, 0x30, 0x30, 0x37, 0x2D, 0x32, 0x30, 0x31, 0x36, 0x20, + 0x50, 0x69, 0x76, 0x6F, 0x74, 0x61, 0x6C, 0x20, 0x53, 0x6F, 0x66, 0x74, 0x77, 0x61, 0x72, 0x65, + 0x2C, 0x20, 0x49, 0x6E, 0x63, 0x2E, 0x0B, 0x69, 0x6E, 0x66, 0x6F, 0x72, 0x6D, 0x61, 0x74, 0x69, + 0x6F, 0x6E, 0x53, 0x00, 0x00, 0x00, 0x35, 0x4C, 0x69, 0x63, 0x65, 0x6E, 0x73, 0x65, 0x64, 0x20, + 0x75, 0x6E, 0x64, 0x65, 0x72, 0x20, 0x74, 0x68, 0x65, 0x20, 0x4D, 0x50, 0x4C, 0x2E, 0x20, 0x20, + 0x53, 0x65, 0x65, 0x20, 0x68, 0x74, 0x74, 0x70, 0x3A, 0x2F, 0x2F, 0x77, 0x77, 0x77, 0x2E, 0x72, + 0x61, 0x62, 0x62, 0x69, 0x74, 0x6D, 0x71, 0x2E, 0x63, 0x6F, 0x6D, 0x2F, 0x08, 0x70, 0x6C, 0x61, + 0x74, 0x66, 0x6F, 0x72, 0x6D, 0x53, 0x00, 0x00, 0x00, 0x0A, 0x45, 0x72, 0x6C, 0x61, 0x6E, 0x67, + 0x2F, 0x4F, 0x54, 0x50, 0x07, 0x70, 0x72, 0x6F, 0x64, 0x75, 0x63, 0x74, 0x53, 0x00, 0x00, 0x00, + 0x08, 0x52, 0x61, 0x62, 0x62, 0x69, 0x74, 0x4D, 0x51, 0x07, 0x76, 0x65, 0x72, 0x73, 0x69, 0x6F, + 0x6E, 0x53, 0x00, 0x00, 0x00, 0x05, 0x33, 0x2E, 0x36, 0x2E, 0x32, 0x00, 0x00, 0x00, 0x0E, 0x41, + 0x4D, 0x51, 0x50, 0x4C, 0x41, 0x49, 0x4E, 0x20, 0x50, 0x4C, 0x41, 0x49, 0x4E, 0x00, 0x00, 0x00, + 0x05, 0x65, 0x6E, 0x5F, 0x55, 0x53, 0xCE +}; + +static uchar connection_tune[20] = { + 0x01, 0x00, 0x00, 0x00, 0x00, 0x00, 0x0C, 0x00, 0x0A, 0x00, 0x1E, 0x00, 0x00, 0x00, 0x02, 0x00, + 0x00, 0x00, 0x3C, 0xCE +}; + +static uchar connection_open_ok[13] = { + 0x01, 0x00, 0x00, 0x00, 0x00, 0x00, 0x05, 0x00, 0x0A, 0x00, 0x29, 0x00, 0xCE +}; + +static uchar channel_open_ok[16] = { + 0x01, 0x00, 0x01, 0x00, 0x00, 0x00, 0x08, 0x00, 0x14, 0x00, 0x0B, 0x00, 0x00, 0x00, 0x00, 0xCE +}; + +static uchar exchange_declare_ok[12] = { + 0x01, 0x00, 0x01, 0x00, 0x00, 0x00, 0x04, 0x00, 0x28, 0x00, 0x0B, 0xCE +}; + +static uchar channel_close_ok[12] = { + 0x01, 0x00, 0x01, 0x00, 0x00, 0x00, 0x04, 0x00, 0x14, 0x00, 0x29, 0xCE +}; + +static uchar connection_close_ok[12] = { + 0x01, 0x00, 0x00, 0x00, 0x00, 0x00, 0x04, 0x00, 0x0A, 0x00, 0x33, 0xCE +}; + +static uchar channel_close_ok_on_badexch[148] = { + 0x01, 0x00, 0x01, 0x00, 0x00, 0x00, 0x8C, 0x00, 0x14, 0x00, 0x28, 0x01, + 0x96, 0x81, 0x50, 0x52, 0x45, 0x43, 0x4F, 0x4E, 0x44, 0x49, 0x54, 0x49, + 0x4F, 0x4E, 0x5F, 0x46, 0x41, 0x49, 0x4C, 0x45, 0x44, 0x20, 0x2D, 0x20, + 0x69, 0x6E, 0x65, 0x71, 0x75, 0x69, 0x76, 0x61, 0x6C, 0x65, 0x6E, 0x74, + 0x20, 0x61, 0x72, 0x67, 0x20, 0x27, 0x64, 0x75, 0x72, 0x61, 0x62, 0x6C, + 0x65, 0x27, 0x20, 0x66, 0x6F, 0x72, 0x20, 0x65, 0x78, 0x63, 0x68, 0x61, + 0x6E, 0x67, 0x65, 0x20, 0x27, 0x69, 0x6E, 0x27, 0x20, 0x69, 0x6E, 0x20, + 0x76, 0x68, 0x6F, 0x73, 0x74, 0x20, 0x27, 0x2F, 0x6D, 0x65, 0x74, 0x72, + 0x6F, 0x6C, 0x6F, 0x67, 0x69, 0x65, 0x27, 0x3A, 0x20, 0x72, 0x65, 0x63, + 0x65, 0x69, 0x76, 0x65, 0x64, 0x20, 0x27, 0x66, 0x61, 0x6C, 0x73, 0x65, + 0x27, 0x20, 0x62, 0x75, 0x74, 0x20, 0x63, 0x75, 0x72, 0x72, 0x65, 0x6E, + 0x74, 0x20, 0x69, 0x73, 0x20, 0x27, 0x74, 0x72, 0x75, 0x65, 0x27, 0x00, + 0x28, 0x00, 0x0A, 0xCE +}; + +typedef struct { + uchar type; + ushort ch; + uint32_t method; + uint16_t header_flags; + size_t datalen; + size_t framelen; + uchar *data; +} amqp_frame_type_t; + +#define DBGPRINTF0(f, ...) if (debug>0) { \ + struct timeval dbgtv; \ + gettimeofday(&dbgtv, NULL);\ + fprintf(stderr, "%02d.%03d " f, (int)(dbgtv.tv_sec % 60), \ + (int)(dbgtv.tv_usec/1000), __VA_ARGS__); \ +} +#define DBGPRINTF1(f, ...) if (debug>0) { \ + struct timeval dbgtv; \ + gettimeofday(&dbgtv, NULL);\ + dbgtv.tv_sec -= dbgtv_base.tv_sec; \ + dbgtv.tv_usec -= dbgtv_base.tv_usec; \ + if (dbgtv.tv_usec < 0) { \ + dbgtv.tv_usec += 1000000; \ + dbgtv.tv_sec--; \ + } \ + fprintf(stderr, "%02d.%03d " f, (int)(dbgtv.tv_sec % 60), \ + (int)(dbgtv.tv_usec/1000), __VA_ARGS__); \ +} +#define DBGPRINTF2(f, ...) if (debug==2) { \ + struct timeval dbgtv; \ + gettimeofday(&dbgtv, NULL);\ + dbgtv.tv_sec -= dbgtv_base.tv_sec; \ + dbgtv.tv_usec -= dbgtv_base.tv_usec; \ + if (dbgtv.tv_usec < 0) { \ + dbgtv.tv_usec += 1000000; \ + dbgtv.tv_sec--; \ + } \ + fprintf(stderr, "%02d.%03d " f, (int)(dbgtv.tv_sec % 60), \ + (int)(dbgtv.tv_usec/1000), __VA_ARGS__); \ +} + +static struct timeval dbgtv_base; +static int server_behaviors = 0; +static int behaviors; +static int wait_after_accept = 200; /* milliseconds */ +static char *outfile = NULL; +static int debug = 1; + +FILE* fpout = NULL; + +static ATTR_NORETURN void +errout(const char *reason, int server) +{ + char txt[256]; + snprintf(txt,256,"%s server %d", reason, server); + perror(txt); + if (fpout && fpout != stdout) { fclose(fpout); fpout = NULL; } + if (outfile) unlink(outfile); + exit(1); +} + +static ATTR_NORETURN void +usage(void) +{ + fprintf(stderr, "usage: minirmqsrvr -f outfile [-b behaviour] " + "[-t keep_alive_max] [-w delay_after_fail] [-d]\n"); + exit (1); +} + +/* Those three functions are "endianess" insensitive */ +static uint16_t buf2uint16(uchar*b) { + return ((uint16_t)b[0]) << 8 | ((uint16_t)b[1]); +} + +static uint32_t buf2uint32(uchar*b) { + return ((uint32_t)b[0]) << 24 | ((uint32_t)b[1]) << 16 | ((uint32_t)b[2]) << 8 | ((uint32_t)b[3]); +} + +static uint64_t buf2uint64(uchar*b) { + return ((uint64_t)b[0]) << 56 | ((uint64_t)b[1]) << 48 | ((uint64_t)b[2]) << 40 | ((uint64_t)b[3]) << 32 + | ((uint64_t)b[4]) << 24 | ((uint64_t)b[5]) << 16 | ((uint64_t)b[6]) << 8 | ((uint64_t)b[7]); +} + + +static char AMQP091[8] = { 'A', 'M', 'Q', 'P', 0x00, 0x00, 0x09, 0x01 }; + +static int +decode_frame_type(uchar *buf, amqp_frame_type_t *frame, size_t nread) { + if (nread == 8){ + if (memcmp(buf, AMQP091, sizeof(AMQP091))) + return -1; + frame->framelen = 8; + frame->type = AMQP_STARTING; + frame->ch = 0; + return 0; + } + frame->type = buf[0]; + frame->ch = buf2uint16(buf+1); + frame->datalen = buf2uint32(buf+3); + frame->framelen = frame->datalen + 8; + frame->method = buf2uint32(buf+7); + switch (frame->type) { + case AMQP_FRAME_BODY: + frame->data = buf + 7; + break; + default: + frame->data = buf + 11; + } + return 0; +} + +static ssize_t +amqp_write(int fdc, uchar *buf, size_t blen, unsigned short channel) { + buf[1] = (char) (channel >> 8); + buf[2] = (char) (channel & 0xFF); + return write(fdc, buf, blen); +} + +static uchar * +amqpFieldUint64(uint64_t *d, uchar *s) { + *d = buf2uint64(s); + return s + 8; +} + +static uchar * +amqpFieldUint32(uint32_t *d, uchar *s) { + *d = buf2uint32(s); + return s + 4; +} + +static uchar * +amqpFieldUint16(uint16_t *d, uchar *s) { + *d = buf2uint16(s); + return s + 2; +} + +static uchar * +amqpFieldLenFprintf(const char *pfx, uchar *s, uint32_t len) { + if (fpout) + fprintf(fpout, "%s%.*s", pfx, (int)len, (char*)s); + return s + len; +} + +static uchar * +amqpFieldFprintf(const char *pfx, uchar *s) { + uint32_t len = *s++; + return amqpFieldLenFprintf(pfx, s, len); +} + +static uchar * +amqpHeaderFprintf(uchar *s, uint32_t *size) { + uint32_t len; + uchar *p = amqpFieldFprintf(", ", s); + p++; /* value type */ + p = amqpFieldUint32(&len, p); + *size -= (p - s) + len; + return amqpFieldLenFprintf(":", p, len); +} + +static void +amqp_srvr(int port, int srvr, int fds, int piperead, int pipewrite) +{ + uchar wrkBuf[8192], *p; + size_t nRead = 0, bsize = 0; + ssize_t nSent; + amqp_frame_type_t frame; + uint64_t body_ui64 = 0; + uint32_t props_header_size; + uint16_t props_flags; + int my_behaviour; + struct timeval tv; + fd_set rfds; + int nfds = ((piperead > fds)? piperead : fds) + 1; + + int fdc; + + my_behaviour = behaviors & 0x000F; + behaviors = behaviors >> 4; /* for next server */; + + if(listen(fds, 0) != 0) errout("listen", port); + + DBGPRINTF1("Server AMQP %d on port %d started\n", srvr, port); + + tv.tv_sec = 120; + tv.tv_usec = 0; + FD_ZERO(&rfds); + FD_SET(fds, &rfds); + if (piperead > 0) + FD_SET(piperead, &rfds); + + if (select(nfds,&rfds,NULL,NULL, &tv) == 0) { + exit(1); + } + + if (piperead > 0 && FD_ISSET(piperead, &rfds)) { + char c; + int l = read(piperead, &c, 1); + if (l == 1) { + my_behaviour = behaviors & 0x000F; + if (my_behaviour != 0) { + DBGPRINTF1("Server AMQP %d on port %d switch behaviour", srvr, port); + } else { + DBGPRINTF1("Server AMQP %d on port %d leaving", srvr, port); + if (fpout && fpout != stdout) { fclose(fpout); fpout = NULL; } + exit(1); + } + } + } + + fdc = accept(fds, NULL, NULL); + + if (pipewrite > 0) + nSent = write(pipewrite, "N", 1); + + close(fds); + fds = -1; + + /* this let the os understand that the port is closed */ + usleep(1000 * wait_after_accept); + + frame.type = AMQP_STARTING; + + while(fdc > 0) { + nSent = 0; + ssize_t rd = 0; + if (nRead < 12) { + rd = read(fdc, wrkBuf + nRead, sizeof(wrkBuf) - nRead); + if (rd <= 0) { + DBGPRINTF1("Server AMQP %d on port %d disconnected\n", srvr, port); + close(fdc); + fdc = 0; + break; + }else { + nRead += (size_t)rd; + } + } + + if (decode_frame_type(wrkBuf, &frame, nRead)) { + DBGPRINTF1("Server AMQP %d on port %d killed : bad protocol\n", srvr, port); + close(fdc); + fdc = 0; + break; + } + if (rd > 4) + DBGPRINTF2("Server received : %zd\n", rd); + + switch (frame.type) { + + case AMQP_STARTING: /* starting handshake */ + + DBGPRINTF1("Server AMQP %d on port %d type %d connected\n", srvr, port, my_behaviour); + DBGPRINTF2("Server %d connection.start\n", srvr); + nSent = amqp_write(fdc, connection_start, sizeof(connection_start), frame.ch); + break; + + case AMQP_FRAME_METHOD: + + DBGPRINTF2("Server %d method : 0x%X\n", srvr, frame.method); + + switch (frame.method) { + + case AMQP_CONNECTION_START_OK_METHOD: + + DBGPRINTF2("Server %d connection.tune\n", srvr); + nSent = amqp_write(fdc, connection_tune, sizeof(connection_tune), frame.ch); + break; + + case AMQP_CONNECTION_TUNE_OK_METHOD: + + DBGPRINTF2("Client %d connection.tune-ok\n", srvr); + nSent = 0; + break; + + case AMQP_CONNECTION_OPEN_METHOD: + + nSent = amqp_write(fdc, connection_open_ok, + sizeof(connection_open_ok), frame.ch); + DBGPRINTF2("Server %d connection.open\n", srvr); + break; + + case AMQP_CHANNEL_OPEN_METHOD: + + nSent = amqp_write(fdc, channel_open_ok, + sizeof(channel_open_ok), frame.ch); + DBGPRINTF2("Server %d channel.open\n", srvr); + if (my_behaviour == AMQP_BEHAVIOR_NOEXCH) { + close(fdc); + DBGPRINTF1("Server AMQP %d on port %d stopped\n", srvr, port); + fdc = 0; + frame.type = 0; + } + break; + + case AMQP_EXCHANGE_DECLARE_METHOD: + + if (my_behaviour == AMQP_BEHAVIOR_BADEXCH) { + nSent = amqp_write(fdc, channel_close_ok_on_badexch, + sizeof(channel_close_ok_on_badexch), frame.ch); + }else{ + nSent = amqp_write(fdc, exchange_declare_ok, + sizeof(exchange_declare_ok), frame.ch); + } + DBGPRINTF2("Server %d exchange.declare\n", srvr); + if (my_behaviour == AMQP_BEHAVIOR_DECEXCH) { + close(fdc); + DBGPRINTF1("Server AMQP %d on port %d stopped\n", srvr, port); + fdc = 0; + frame.type = 0; + } + break; + + case AMQP_CHANNEL_CLOSE_METHOD: + + nSent = amqp_write(fdc, channel_close_ok, + sizeof(channel_close_ok), frame.ch); + DBGPRINTF2("Server %d channel.close\n", srvr); + break; + + case AMQP_CONNECTION_CLOSE_METHOD: + + nSent = amqp_write(fdc, connection_close_ok, + sizeof(connection_close_ok), frame.ch); + DBGPRINTF2("Server %d connection.close\n", srvr); + break; + + case AMQP_BASIC_PUBLISH_METHOD: + + p = amqpFieldFprintf("Exchange:", frame.data + 2); + amqpFieldFprintf(", routing-key:", p); + break; + + default: + + nSent = 0; + } + break; + + case AMQP_FRAME_HEADER: + + DBGPRINTF2("Server %d HEADERS\n", srvr); + p = amqpFieldUint64(&body_ui64, frame.data); + bsize = (size_t)body_ui64; + p = amqpFieldUint16(&props_flags, p); + if (props_flags & AMQP_BASIC_CONTENT_TYPE_FLAG) { + p = amqpFieldFprintf(", content-type:", p); + } + if (props_flags & AMQP_BASIC_HEADERS_FLAG) { + p = amqpFieldUint32(&props_header_size, p); + while (props_header_size) { + p = amqpHeaderFprintf(p, &props_header_size); + } + } + if (props_flags & AMQP_BASIC_DELIVERY_MODE_FLAG) { + if (fpout) + fprintf(fpout, ", delivery-mode:%s", (*p++)?"transient":"persistent"); + } + if (props_flags & AMQP_BASIC_EXPIRATION_FLAG) { + p = amqpFieldFprintf(", expiration:", p); + } + if (props_flags & AMQP_BASIC_TIMESTAMP_FLAG) { + if (fpout) + fprintf(fpout, ", timestamp:OK"); + p += sizeof(uint64_t); + } + if (props_flags & AMQP_BASIC_APP_ID_FLAG) { + amqpFieldFprintf(", app-id:", p); + } + if (fpout) + fprintf(fpout, ", msg:"); + break; + + case AMQP_FRAME_BODY: + + DBGPRINTF2("Server %d Body size left : %zu, received : %zu\n", + srvr, bsize, frame.datalen); + bsize -= frame.datalen; + if (fpout) { + fprintf(fpout, "%.*s", (int)frame.datalen, frame.data); + if (frame.data[frame.datalen-1] != '\n') + fprintf(fpout, "\n"); + fflush(fpout); + } + break; + + default: + + DBGPRINTF1("Server %d unsupported frame type %d\n", srvr, frame.type); + close(fdc); + fdc = 0; + frame.type = 0; + frame.framelen = 0; + } /* switch (frame.type) */ + + nRead -= frame.framelen; + if (nRead>0) + memmove(wrkBuf, wrkBuf + frame.framelen, nRead); + + if (nSent < 0) { + close(fdc); + fdc = 0; + } + } /* while(fdc) */ + DBGPRINTF2("Leaving thread %d\n", srvr); +} + +int +main(int argc, char *argv[]) +{ + int port[2], fds[2], i, opt, nb_port = 1; + int pipeS1toS2[2] = { -1, -1 }; + int pipeS2toS1[2] = { -1, -1 }; + int pipeRead[2], pipeWrite[2]; + + + struct sockaddr_in srvAddr[2]; + unsigned int addrLen = sizeof(struct sockaddr_in), len; + pid_t pid[2]; + + fpout = stdout; + + while((opt = getopt(argc, argv, "f:b:w:d")) != -1) { + switch (opt) { + case 'w': + wait_after_accept = atoi(optarg); + break; + case 'd': + debug = 2; + break; + case 'b': + server_behaviors = atoi(optarg); + break; + case 'f': + if(strcmp(optarg, "-")) { + outfile = optarg; + fpout = fopen(optarg, "w"); + if(fpout == NULL){ + fprintf(stderr, "file %s could not be created\n", outfile); + exit(1); + } + } + break; + default: + fprintf(stderr, "invalid option '%c' or value missing - terminating...\n", opt); + usage(); + break; + } + } + + switch (server_behaviors) { + case 0: + behaviors = AMQP_BEHAVIOR_STANDARD; + nb_port = 1; + break; + case 1: /* two standard servers get message successfully */ + behaviors = AMQP_BEHAVIOR_STANDARD; + nb_port = 2; + break; + case 2: /* 2 servers first server which disconnect after after open channel : no declare exchange */ + behaviors = AMQP_BEHAVIOR_NOEXCH | AMQP_BEHAVIOR_STANDARD << 4; + nb_port = 2; + break; + case 3: /* 2 servers first server which disconnect after declare exchange*/ + behaviors = AMQP_BEHAVIOR_DECEXCH | AMQP_BEHAVIOR_STANDARD << 4; + nb_port = 2; + break; + case 4: /* one server with bad exchange declare */ + behaviors = AMQP_BEHAVIOR_BADEXCH; + nb_port = 1; + break; + default: + fprintf(stderr,"Invalid behavior"); + exit(1); + } + + gettimeofday(&dbgtv_base, NULL); + + port[0] = port[1] = -1; + + if (nb_port == 2) { + if(pipe(pipeS1toS2) == -1 || pipe(pipeS2toS1) == -1) { + fprintf(stderr, "Pipe failed !"); + exit(1); + } + } + + pipeRead[0] = pipeS2toS1[0]; + pipeWrite[0] = pipeS1toS2[1]; + + pipeRead[1] = pipeS1toS2[0]; + pipeWrite[1] = pipeS2toS1[1]; + + for (i = 0; i < nb_port; i++) { + fds[i] = socket(AF_INET, SOCK_STREAM, 0); + srvAddr[i].sin_family = AF_INET; + srvAddr[i].sin_addr.s_addr = INADDR_ANY; + srvAddr[i].sin_port = 0; + if(bind(fds[i], (struct sockaddr *)&srvAddr[i], addrLen) != 0) + errout("bind", 1); + len = addrLen; + if (getsockname(fds[i], (struct sockaddr *)&srvAddr[i], &len) == -1) + errout("bind", i+1); + if ((port[i] = ntohs(srvAddr[i].sin_port)) <= 0) + errout("get port", i+1); + } + + for (i = 0; i < nb_port; i++) { + if ((pid[i] = fork()) == -1) { + fprintf(stderr, "Fork failed !"); + exit(1); + } + if (pid[i] == 0) { + /* this is the child */ + if (fds[1-i] > 0) close(fds[1-i]); + amqp_srvr(port[i], i+1, fds[i], pipeRead[i], pipeWrite[i]); + + if (fpout && fpout != stdout) fclose(fpout); + + DBGPRINTF2("%s\n","Leaving server"); + return 0; + } + } + + if (nb_port==2) + printf("export AMQPSRVRPID1=%ld AMQPSRVRPID2=%ld PORT_AMQP1=%d PORT_AMQP2=%d", + (long)pid[0], (long)pid[1], port[0], port[1]); + else + printf("export AMQPSRVRPID1=%ld PORT_AMQP1=%d", + (long)pid[0], port[0]); + return 0; +} |