Changeset 852 for trunk/Monitoring

Timestamp:

04/23/13 20:26:50 (11 years ago)

Author:

jripsl

Message:

• AMQP C-client implementation.

Location:

trunk/Monitoring

Files:

• Broker/README (added)
• CNClient/README (modified) (1 diff)
• CNClient/send_AMQP_msg.c (modified) (5 diffs)
• doc/message_specification.txt (added)
• test/README (modified) (1 diff)
• test/recv_AMQP_msg.c (added)

trunk/Monitoring/CNClient/README

@@ -5 +5 @@
         - Library installation
                 - using system package
-                        - aptitude install librabbitmq0
-                        - aptitude install librabbitmq-dev
+                        - install using commands below
+                                - aptitude install librabbitmq0
+                                - aptitude install librabbitmq-dev
+                        - note that it is likely that system packages versions are out of date (we need v0.3.0)
+                          so better use installation from source
                 - from source
                         tar xzvf rabbitmq-c-v0.3.0.zip
                         cd rabbitmq-c-v0.3.0
                         autoreconf -i
-                        ./configure
+                        ./configure --enable-static
                         make
                         make install
-        - gcc -I/usr/local/include -L/usr/local/lib -Wall send_AMQP_msg.c -lrabbitmq -o sendAMQPMsg
+        - gcc -static -I/usr/local/include -L/usr/local/lib -Wall -o sendAMQPMsg send_AMQP_msg.c -lrabbitmq
+                - we get warning below during compilation
+                        - 
+                          <---
+                          /usr/local/lib/librabbitmq.a(librabbitmq_librabbitmq_la-amqp_socket.o): In function `amqp_open_socket':
+                          rabbitmq-c-rabbitmq-c-v0.3.0/librabbitmq/amqp_socket.c:66: warning: Using 'getaddrinfo' in statically 
+                          linked applications requires at runtime the shared libraries from the glibc version used for linking
+                          --->
+                        - it means that you may need to be sure all computing node have the same glibc version !!!!
+                                - also means that a different binary must be use in each computing center
 - Run
         - ./sendAMQPMsg localhost 5672 1 10

trunk/Monitoring/CNClient/send_AMQP_msg.c

@@ -104 +104 @@
     uint64_t now = now_microseconds();
 
-    die_on_error(amqp_basic_publish(conn, 1, amqp_cstring_bytes("amq.direct"), amqp_cstring_bytes(queue_name), 0, 0, NULL, message_bytes), "Publishing");
+    { 
+      amqp_basic_properties_t props;
+      props._flags = AMQP_BASIC_CONTENT_TYPE_FLAG | AMQP_BASIC_DELIVERY_MODE_FLAG;
+      props.content_type = amqp_cstring_bytes("text/plain");
+      props.delivery_mode = 2; /* persistent delivery mode */
+
+      // works ! (queue get populated even if no consumer)
+      die_on_error(amqp_basic_publish(conn, 1, amqp_cstring_bytes(""), amqp_cstring_bytes(queue_name), 0, 0, &props, message_bytes), "Publishing");
+
+      // don't works ! (queue get populated only if consumer up)
+      //die_on_error(amqp_basic_publish(conn, 1, amqp_cstring_bytes("amq.direct"), amqp_cstring_bytes(queue_name), 0, 0, &props, message_bytes), "Publishing");
+    }
+
+        // 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");
+
+
     sent++;
     if (now > next_summary_time) {
@@ -132 +149 @@
 }
 
-int main(int argc, char const * const *argv) {
+//int main(int argc, char const * const *argv) {
+int main_light(int argc, char const * const *argv) {
   char const *hostname;
   int port;
@@ -140 +158 @@
   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");
@@ -162 +182 @@
   die_on_amqp_error(amqp_get_rpc_reply(conn), "Opening channel");
 
-  send_batch(conn, "test queue", rate_limit, message_count);
+  //JRA
+  /*
+  {
+    amqp_queue_declare_ok_t *r = amqp_queue_declare(conn, 1, amqp_cstring_bytes("test queue"), 0, 1, 0, 0, amqp_empty_table); // durable && no auto-delete
+    die_on_amqp_error(amqp_get_rpc_reply(conn), "Declaring queue");
+    reply_to_queue = amqp_bytes_malloc_dup(r->queue);
+    if (reply_to_queue.bytes == NULL) {
+      fprintf(stderr, "Out of memory while copying queue name");
+      return 1;
+    }
+  }
+  */
+
+  send_batch(conn, "test queue", rate_limit, message_count); // note that "test queue" here is used as the routing key
 
   die_on_amqp_error(amqp_channel_close(conn, 1, AMQP_REPLY_SUCCESS), "Closing channel");
@@ -170 +203 @@
   return 0;
 }
+
+
+//int main_complex(int argc, char const * const *argv) {
+int main(int argc, char const * const *argv) {
+  char const *hostname;
+  int port;
+  char const *exchange;
+  char const *routingkey;
+  char const *messagebody;
+
+
+  
+  int sockfd;
+  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];
+
+  conn = amqp_new_connection();
+
+  die_on_error(sockfd = amqp_open_socket(hostname, port), "Opening socket");
+  amqp_set_sockfd(conn, sockfd);
+  die_on_amqp_error(amqp_login(conn, "/", 0, 131072, 0, AMQP_SASL_METHOD_PLAIN, "guest", "guest"), "Logging in");
+  amqp_channel_open(conn, 1);
+  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
+  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");
+
+
+
+  { 
+    amqp_basic_properties_t props;
+    props._flags = AMQP_BASIC_CONTENT_TYPE_FLAG | AMQP_BASIC_DELIVERY_MODE_FLAG;
+    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_amqp_error(amqp_channel_close(conn, 1, AMQP_REPLY_SUCCESS), "Closing channel");
+  die_on_amqp_error(amqp_connection_close(conn, AMQP_REPLY_SUCCESS), "Closing connection");
+  die_on_error(amqp_destroy_connection(conn), "Ending connection");
+  return 0;
+}
+
+// -- not used -- //
+
+static void dump_row(long count, int numinrow, int *chs) {
+  int i;
+
+  printf("%08lX:", count - numinrow);
+
+  if (numinrow > 0) {
+    for (i = 0; i < numinrow; i++) {
+      if (i == 8)
+    printf(" :");
+      printf(" %02X", chs[i]);
+    }
+    for (i = numinrow; i < 16; i++) {
+      if (i == 8)
+    printf(" :");
+      printf("   ");
+    }
+    printf("  ");
+    for (i = 0; i < numinrow; i++) {
+      if (isprint(chs[i]))
+    printf("%c", chs[i]);
+      else
+    printf(".");
+    }
+  }
+  printf("\n");
+}
+
+static int rows_eq(int *a, int *b) {
+  int i;
+
+  for (i=0; i<16; i++)
+    if (a[i] != b[i])
+      return 0;
+
+  return 1;
+}
+
+void amqp_dump(void const *buffer, size_t len) {
+  unsigned char *buf = (unsigned char *) buffer;
+  long count = 0;
+  int numinrow = 0;
+  int chs[16];
+  int oldchs[16] = {0};
+  int showed_dots = 0;
+  size_t i;
+
+  for (i = 0; i < len; i++) {
+    int ch = buf[i];
+
+    if (numinrow == 16) {
+      int i;
+
+      if (rows_eq(oldchs, chs)) {
+    if (!showed_dots) {
+      showed_dots = 1;
+      printf("          .. .. .. .. .. .. .. .. : .. .. .. .. .. .. .. ..\n");
+    } 
+      } else {
+    showed_dots = 0;
+    dump_row(count, numinrow, chs);
+      }
+
+      for (i=0; i<16; i++)
+    oldchs[i] = chs[i];
+
+      numinrow = 0;
+    }
+
+    count++;
+    chs[numinrow++] = ch;
+  }
+
+  dump_row(count, numinrow, chs);
+
+  if (numinrow != 0)
+    printf("%08lX:\n", count);
+}
+

trunk/Monitoring/test/README

@@ -1 +1 @@
 test_twisted.py   consumer/producer test showing how to use RabbitMQ, Twisted and txamqp together
+recv_AMQP_msg.c   basic consumer test