Changeset 854 for trunk/Monitoring/CNClient/send_AMQP_msg.c

Timestamp:

04/26/13 16:57:16 (11 years ago)

Author:

jripsl

Message:

add JSON deserialization, mail notification, logging.

File:

• trunk/Monitoring/CNClient/send_AMQP_msg.c (modified) (9 diffs)

trunk/Monitoring/CNClient/send_AMQP_msg.c

@@ -16 +16 @@
 
 #define SUMMARY_EVERY_US 1000000
-
 
 
@@ -117 +116 @@
     }
 
-        // note that "amq.direct" is a special exchange
-        // (The empty exchange name is an alias for amq.direct)
+          // note that "amq.direct" is a special exchange
+          // (The empty exchange name is an alias for amq.direct)
     //die_on_error(amqp_basic_publish(conn, 1, amqp_cstring_bytes("amq.direct"), amqp_cstring_bytes(queue_name), 0, 0, NULL, message_bytes), "Publishing");
 
@@ -158 +157 @@
   amqp_connection_state_t conn;
 
-  amqp_bytes_t reply_to_queue;
-
-  if (argc < 5) {
-    fprintf(stderr, "Usage: amqp_producer host port rate_limit message_count\n");
-    return 1;
-  }
+  //amqp_bytes_t reply_to_queue;
+
 
   hostname = argv[1];
@@ -204 +199 @@
 }
 
-
-//int main_complex(int argc, char const * const *argv) {
-int main(int argc, char const * const *argv) {
+int read_file(char *f,char **source) {
+        FILE *fp = fopen(f, "r");
+        if (fp != NULL) {
+                /* Go to the end of the file. */
+                if (fseek(fp, 0L, SEEK_END) == 0) {
+                        /* Get the size of the file. */
+                        long bufsize = ftell(fp);
+                        if (bufsize == -1) { /* Error */ }
+
+                        /* Allocate our buffer to that size. */
+                        *source = malloc(sizeof(char) * (bufsize + 1));
+
+                        /* Go back to the start of the file. */
+                        if (fseek(fp, 0L, SEEK_SET) == 0) { /* Error */ }
+
+                        /* Read the entire file into memory. */
+                        size_t newLen = fread(*source, sizeof(char), bufsize, fp);
+                        if (newLen == 0) {
+                                fputs("Error reading file\n", stderr);
+        return 1;
+                        } else {
+                                //source[++newLen] = '\0'; /* Just to be safe. */
+                        }
+                }
+                fclose(fp);
+        } else {
+                fputs("File not found\n", stderr);
+                return 1;
+        }
+
+  return 0;
+}
+
+int main(int argc, char * const *argv) {
   char const *hostname;
   int port;
-  char const *exchange;
-  char const *routingkey;
-  char const *messagebody;
+  //char const *exchange;
+  //char const *routingkey;
+  char *body_tmp;
+  char const *body;
+  char *body_final = NULL;
+  char *buf = NULL;
+  int c;
+  char *filepath = NULL;
+  int file_flag=0;
 
 
@@ -218 +250 @@
   amqp_connection_state_t conn;
 
-  if (argc < 4) {
-    fprintf(stderr, "Usage: amqp_sendstring host port messagebody\n");
-    return 1;
-  }
-
-  hostname = argv[1];
-  port = atoi(argv[2]);
-  //exchange = argv[3];
-  //routingkey = argv[4];
-  messagebody = argv[5];
+  //parsing args
+  opterr = 0;
+  while ((c = getopt (argc, argv, "b:ef:h:p:")) != -1)
+    switch (c)
+      {
+      case 'b':
+        body_tmp = optarg;
+        break;
+      case 'e':
+        fprintf(stderr, "Usage: amqp_producer host port rate_limit message_count\n");
+        fprintf(stderr, "Example 1: ./sendAMQPMsg -h localhost -p 5672 -b '{\"jobid\":\"toto\"}'\n");
+        fprintf(stderr, "Example 2: ./sendAMQPMsg -h localhost -p 5672 -f /home/foobar/config.card -b '{\"jobid\":\"toto\"}'\n");
+        exit(0);
+      case 'f':
+        filepath = optarg;
+        file_flag = 1;
+        break;
+      case 'h':
+        hostname = optarg;
+        break;
+      case 'p':
+        port = atoi(optarg);
+        break;
+      case '?':
+        fprintf (stderr, "ERR001: incorrect argument '-%c'.\n", optopt);
+        exit(EXIT_FAILURE);
+      default:
+        fprintf (stderr, "ERR002: incorrect argument\n");
+        exit(EXIT_FAILURE);
+      }
+
+  //retrieve non-option argument
+  /*
+  int index;
+  for (index = optind; index < argc; index++)
+    printf ("Non-option argument %s\n", argv[index]);
+  */
+
+  // add checks here
+  // (for example, body_tmp is mandatory)
+
+  if(file_flag==1) {
+    // retrieve file contents
+
+    if( access( filepath, F_OK ) != -1 ) {
+        // file exists
+        
+        ;
+    } else {
+      // file doesn't exist
+
+      fprintf(stderr, "File not found (%s)\n",filepath);
+
+      exit(EXIT_FAILURE);
+    }
+
+    int res = read_file(filepath,&buf);
+    if (res != 0) {
+      exit(EXIT_FAILURE);
+    }
+
+    body_final=malloc(strlen(body_tmp) + 6 + strlen(buf) + 1);
+    //strcpy(body_final,"\0");
+    strcat(body_final,body_tmp);
+    strcat(body_final,",file=");
+    strcat(body_final,buf);
+
+    body=body_final;
+
+    //debug
+    //fprintf(stderr, "hostname=%s, port=%d, body=%s, filepath=%s\n", hostname, port, body, filepath);
+
+  } else {
+    // retrieve msg body from argument
+
+    body=body_tmp;
+
+  }
+
+  //exit(0);
 
   conn = amqp_new_connection();
@@ -237 +339 @@
   die_on_amqp_error(amqp_get_rpc_reply(conn), "Opening channel");
 
-
-
-
-  //JRA
   amqp_exchange_declare(conn, 1, amqp_cstring_bytes("myexchange"), amqp_cstring_bytes("fanout"), 0, 0, amqp_empty_table);
   die_on_amqp_error(amqp_get_rpc_reply(conn), "Declaring exchange");
 
-  //JRA
-  amqp_queue_declare_ok_t *r = amqp_queue_declare(conn, 1, amqp_cstring_bytes("myqueue"), 0, 1, 0, 0, amqp_empty_table); // durable && no auto-delete
+  amqp_queue_declare(conn, 1, amqp_cstring_bytes("myqueue"), 0, 1, 0, 0, amqp_empty_table); // durable && no auto-delete
   die_on_amqp_error(amqp_get_rpc_reply(conn), "Declaring queue");
 
   amqp_queue_bind(conn, 1, amqp_cstring_bytes("myqueue"), amqp_cstring_bytes("myexchange"), amqp_cstring_bytes(""), amqp_empty_table); //no need for binding key as we use the fanout exchange type
   die_on_amqp_error(amqp_get_rpc_reply(conn), "Binding");
-
-
 
   { 
@@ -258 +353 @@
     props.content_type = amqp_cstring_bytes("text/plain");
     props.delivery_mode = 2; /* persistent delivery mode */
-    die_on_error(amqp_basic_publish(conn, 1, amqp_cstring_bytes("myexchange"), amqp_cstring_bytes(""), 0, 0, &props, amqp_cstring_bytes(messagebody)), "Publishing");
+    die_on_error(amqp_basic_publish(conn, 1, amqp_cstring_bytes("myexchange"), amqp_cstring_bytes(""), 0, 0, &props, amqp_cstring_bytes(body)), "Publishing");
   }
 
@@ -264 +359 @@
   die_on_amqp_error(amqp_connection_close(conn, AMQP_REPLY_SUCCESS), "Closing connection");
   die_on_error(amqp_destroy_connection(conn), "Ending connection");
+
+  if(file_flag==1)
+    free(buf);
+
   return 0;
 }
@@ -347 +446 @@
 }
 
+/* vi: set et ts=2 sw=2: */