/** * * 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 "stomp.h" #include int die(int exitCode, const char *message, apr_status_t reason) { char msgbuf[80]; apr_strerror(reason, msgbuf, sizeof(msgbuf)); fprintf(stderr, "%s: %s (%d)\n", message, msgbuf, reason); exit(exitCode); return reason; } static void terminate() { apr_terminate(); } int Subscribe(); int Publish(); void usageExit(char* prog) { printf("Usage:%s -d pub|sub [-q queue_name] [-n node_name] [-p portnum] [-m maxmsgs] [-s sleep_seconds]" "\n",prog); printf ("Where maxmsgs default = 10\n"); exit(1); } char* qname="/queue/TEST.FOO"; char* nodename="127.0.0.1"; int port=61613; int direction=0; int maxmsg = 10; int sleeptime = 0; int main(int argc, char **argv) { char c = ' '; while((c = getopt(argc, argv, "d:q:n:p:m:s:")) != EOF) { switch(c) { case 'd' : if (!strcmp(optarg,"pub")) {direction=2;} else if (!strcmp(optarg,"sub")) {direction=1;} if (!direction) {printf ("argument for -d must be 'pub' or 'sub'\n");exit(1);} break; case 'q' : qname = strdup(optarg); break; case 'n' : nodename = strdup(optarg); break; case 'p' : port = atoi(optarg); break; case 's' : sleeptime = atoi(optarg); break; case 'm' : maxmsg = atoi(optarg); break; default : usageExit(argv[0]); break; } } switch (direction) { case 1: return Subscribe(); break; case 2: return Publish(); break; default: usageExit(argv[0]); return -1; break; } } int Subscribe() { int i = 0; apr_status_t rc; apr_pool_t *pool; stomp_connection *connection; setbuf(stdout, NULL); rc = apr_initialize(); rc==APR_SUCCESS || die(-2, "Could not initialize", rc); atexit(terminate); rc = apr_pool_create(&pool, NULL); rc==APR_SUCCESS || die(-2, "Could not allocate pool", rc); fprintf(stdout, "Connecting......"); rc=stomp_connect( &connection, nodename, port, pool); rc==APR_SUCCESS || die(-2, "Could not connect", rc); fprintf(stdout, "OK\n"); fprintf(stdout, "Sending connect message."); { stomp_frame frame; frame.command = "CONNECT"; frame.headers = apr_hash_make(pool); frame.body = NULL; rc = stomp_write(connection, &frame); rc==APR_SUCCESS || die(-2, "Could not send frame", rc); } fprintf(stdout, "OK\n"); fprintf(stdout, "Reading Response."); { stomp_frame *frame; rc = stomp_read(connection, &frame, pool); rc==APR_SUCCESS || die(-2, "Could not read frame", rc); fprintf(stdout, "Response: %s, %s\n", frame->command, frame->body); } fprintf(stdout, "OK\n"); fprintf(stdout, "Sending Subscribe."); { stomp_frame frame; frame.command = "SUB"; frame.headers = apr_hash_make(pool); //apr_hash_set(frame.headers, "destination", APR_HASH_KEY_STRING, "/queue/FOO.BAR"); apr_hash_set(frame.headers, "destination", APR_HASH_KEY_STRING, qname); frame.body = NULL; rc = stomp_write(connection, &frame); rc==APR_SUCCESS || die(-2, "Could not send frame", rc); } fprintf(stdout, "OK\n"); fprintf(stdout, "Reading Subscribed Messsages."); for (i=0; i < maxmsg; i++) { stomp_frame *frame; rc = stomp_read(connection, &frame, pool); rc==APR_SUCCESS || die(-2, "Could not read frame", rc); fprintf(stdout, "Received: %s, %s\n", frame->command, frame->body); sleep(sleeptime); } fprintf(stdout, "OK\n"); fprintf(stdout, "Sending Disconnect."); { stomp_frame frame; frame.command = "DISCONNECT"; frame.headers = NULL; frame.body = NULL; rc = stomp_write(connection, &frame); rc==APR_SUCCESS || die(-2, "Could not send frame", rc); } fprintf(stdout, "OK\n"); fprintf(stdout, "Disconnecting..."); rc=stomp_disconnect(&connection); rc==APR_SUCCESS || die(-2, "Could not disconnect", rc); fprintf(stdout, "OK\n"); apr_pool_destroy(pool); return 0; } int Publish() { int i; char c_message[256]; c_message[255]='\0'; apr_status_t rc; apr_pool_t *pool; stomp_connection *connection; setbuf(stdout, NULL); rc = apr_initialize(); rc==APR_SUCCESS || die(-2, "Could not initialize", rc); atexit(terminate); rc = apr_pool_create(&pool, NULL); rc==APR_SUCCESS || die(-2, "Could not allocate pool", rc); fprintf(stdout, "Connecting......"); rc=stomp_connect( &connection, nodename, port, pool); rc==APR_SUCCESS || die(-2, "Could not connect", rc); fprintf(stdout, "OK\n"); fprintf(stdout, "Sending connect message."); { stomp_frame frame; frame.command = "CONNECT"; frame.headers = apr_hash_make(pool); frame.body = NULL; rc = stomp_write(connection, &frame); rc==APR_SUCCESS || die(-2, "Could not send frame", rc); } fprintf(stdout, "OK\n"); fprintf(stdout, "Reading Response."); { stomp_frame *frame; rc = stomp_read(connection, &frame, pool); rc==APR_SUCCESS || die(-2, "Could not read frame", rc); fprintf(stdout, "Response: %s, %s\n", frame->command, frame->body); } fprintf(stdout, "OK\n"); fprintf(stdout, "Sending Message."); for (i=0; i < maxmsg; i++) { stomp_frame frame; frame.command = "SEND"; frame.headers = apr_hash_make(pool); apr_hash_set(frame.headers, "destination", APR_HASH_KEY_STRING, qname); snprintf(c_message, 255, "This is message number %d",i); frame.body = c_message; rc = stomp_write(connection, &frame); rc==APR_SUCCESS || die(-2, "Could not send frame", rc); printf ("Sending Message: %s\n",c_message); sleep(sleeptime); } fprintf(stdout, "OK\n"); fprintf(stdout, "Sending Disconnect."); { stomp_frame frame; frame.command = "DISCONNECT"; frame.headers = NULL; frame.body = NULL; rc = stomp_write(connection, &frame); rc==APR_SUCCESS || die(-2, "Could not send frame", rc); } fprintf(stdout, "OK\n"); fprintf(stdout, "Disconnecting..."); rc=stomp_disconnect(&connection); rc==APR_SUCCESS || die(-2, "Could not disconnect", rc); fprintf(stdout, "OK\n"); apr_pool_destroy(pool); return 0; }