Index: /trunk/Monitoring/Broker/README
===================================================================
--- /trunk/Monitoring/Broker/README	(revision 852)
+++ /trunk/Monitoring/Broker/README	(revision 852)
@@ -0,0 +1,53 @@
+- installation (as root)
+	- RabbitMQ server
+		yum install erlang.x86_64
+		yum install libxslt-devel.x86_64
+		yum install libxslt.x86_64
+		yum install xmlto.x86_64
+		cd /opt
+		mkdir -p src
+		cd src
+		# rabbitmq install
+		cd /opt/src
+		wget http://www.rabbitmq.com/releases/rabbitmq-server/v3.0.2/rabbitmq-server-3.0.2.tar.gz
+		tar xzvf rabbitmq-server-3.0.2.tar.gz
+		rm rabbitmq-server-3.0.2.tar.gz
+		cd rabbitmq-server-3.0.2/
+		export TARGET_DIR=/opt/rabbitmq-server-3.0.2
+		export SBIN_DIR=/opt/rabbitmq-server-3.0.2/sbin
+		export MAN_DIR=/opt/rabbitmq-server-3.0.2/man
+		make
+		make install
+	- console admin tools (http://www.rabbitmq.com/management-cli.html)
+		- wget http://hg.rabbitmq.com/rabbitmq-management/raw-file/rabbitmq_v3_0_2/bin/rabbitmqadmin
+		- mv -i rabbitmqadmin /usr/sbin
+		- chmod +x /usr/sbin/rabbitmqadmin
+- conf
+	- vi /opt/rabbitmq-server-3.0.2/sbin/rabbitmq-defaults
+	- as default disk size is too high for VESG4 VM, do that
+		- edit rabbit config file so result looks like this:
+			$ cat /etc/rabbitmq/rabbitmq.config
+			[   
+				{rabbit, [{disk_free_limit, {mem_relative, 0.1}}]}
+			].
+	-   
+		-   
+			- $RABBITMQ_HOME/etc/rabbitmq/rabbitmq-env.conf
+			- $RABBITMQ_HOME/etc/rabbitmq/rabbitmq.config
+		- note that neither of these files exist after installation
+		  i.e. you need to create them
+	-   
+		log         : /var/log/rabbitmq/rabbit.log
+		sasl log    : /var/log/rabbitmq/rabbit-sasl.log
+		database dir: /var/lib/rabbitmq/mnesia/rabbit
+- run
+	- to start the daemon, use command below
+		- ./rabbitmq-server -detached
+	- to stop the daemon, use this
+		- kill -TERM $(pidof epmd)
+- operation
+	- to print RabbitMQ status, use
+		- rabbitmqctl report
+	- to print RabbitMQ queues, use:
+		- rabbitmqadmin list queues
+		- rabbitmqctl list_queues
Index: /trunk/Monitoring/CNClient/README
===================================================================
--- /trunk/Monitoring/CNClient/README	(revision 851)
+++ /trunk/Monitoring/CNClient/README	(revision 852)
@@ -5,14 +5,26 @@
 	- 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
Index: /trunk/Monitoring/CNClient/send_AMQP_msg.c
===================================================================
--- /trunk/Monitoring/CNClient/send_AMQP_msg.c	(revision 851)
+++ /trunk/Monitoring/CNClient/send_AMQP_msg.c	(revision 852)
@@ -104,5 +104,22 @@
     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,5 +149,6 @@
 }
 
-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,4 +158,6 @@
   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,5 +182,18 @@
   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,2 +203,146 @@
   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);
+}
+
Index: /trunk/Monitoring/doc/message_specification.txt
===================================================================
--- /trunk/Monitoring/doc/message_specification.txt	(revision 852)
+++ /trunk/Monitoring/doc/message_specification.txt	(revision 852)
@@ -0,0 +1,19 @@
+- simulation progression message
+	- format
+		- job_id
+			- config-card file
+			- center
+			- machine
+			- user
+			- start
+			- timestamp
+		- sequence number
+			- 1..N
+		- 6CN_QCHOSE
+		- timestamp
+		- nesting
+		- in / out
+		- status
+			- OK / NOK
+		- errcode
+			- 001.002
Index: /trunk/Monitoring/test/README
===================================================================
--- /trunk/Monitoring/test/README	(revision 851)
+++ /trunk/Monitoring/test/README	(revision 852)
@@ -1,1 +1,2 @@
 test_twisted.py   consumer/producer test showing how to use RabbitMQ, Twisted and txamqp together
+recv_AMQP_msg.c   basic consumer test
Index: /trunk/Monitoring/test/recv_AMQP_msg.c
===================================================================
--- /trunk/Monitoring/test/recv_AMQP_msg.c	(revision 852)
+++ /trunk/Monitoring/test/recv_AMQP_msg.c	(revision 852)
@@ -0,0 +1,291 @@
+#include <stdlib.h>
+#include <stdio.h>
+#include <string.h>
+#include <stdint.h>
+#include <sys/time.h>
+#include <unistd.h>
+
+
+#include <ctype.h>
+
+#include <amqp.h>
+#include <amqp_framing.h>
+
+#include <assert.h>
+
+#include "utils.h"
+
+
+#define SUMMARY_EVERY_US 1000000
+
+
+
+
+uint64_t now_microseconds(void)
+{
+  struct timeval tv;
+  gettimeofday(&tv, NULL);
+  return (uint64_t) tv.tv_sec * 1000000 + (uint64_t) tv.tv_usec;
+}
+
+void microsleep(int usec)
+{
+  usleep(usec);
+}
+
+void die_on_error(int x, char const *context) {
+  if (x < 0) {
+    char *errstr = amqp_error_string(-x);
+    fprintf(stderr, "%s: %s\n", context, errstr);
+    free(errstr);
+    exit(1);
+  }
+}
+
+void die_on_amqp_error(amqp_rpc_reply_t x, char const *context) {
+  switch (x.reply_type) {
+    case AMQP_RESPONSE_NORMAL:
+      return;
+
+    case AMQP_RESPONSE_NONE:
+      fprintf(stderr, "%s: missing RPC reply type!\n", context);
+      break;
+
+    case AMQP_RESPONSE_LIBRARY_EXCEPTION:
+      fprintf(stderr, "%s: %s\n", context, amqp_error_string(x.library_error));
+      break;
+
+    case AMQP_RESPONSE_SERVER_EXCEPTION:
+      switch (x.reply.id) {
+	case AMQP_CONNECTION_CLOSE_METHOD: {
+	  amqp_connection_close_t *m = (amqp_connection_close_t *) x.reply.decoded;
+	  fprintf(stderr, "%s: server connection error %d, message: %.*s\n",
+		  context,
+		  m->reply_code,
+		  (int) m->reply_text.len, (char *) m->reply_text.bytes);
+	  break;
+	}
+	case AMQP_CHANNEL_CLOSE_METHOD: {
+	  amqp_channel_close_t *m = (amqp_channel_close_t *) x.reply.decoded;
+	  fprintf(stderr, "%s: server channel error %d, message: %.*s\n",
+		  context,
+		  m->reply_code,
+		  (int) m->reply_text.len, (char *) m->reply_text.bytes);
+	  break;
+	}
+	default:
+	  fprintf(stderr, "%s: unknown server error, method id 0x%08X\n", context, x.reply.id);
+	  break;
+      }
+      break;
+  }
+
+  exit(1);
+}
+
+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);
+}
+
+static void run(amqp_connection_state_t conn)
+{
+  uint64_t start_time = now_microseconds();
+  int received = 0;
+  int previous_received = 0;
+  uint64_t previous_report_time = start_time;
+  uint64_t next_summary_time = start_time + SUMMARY_EVERY_US;
+
+  amqp_frame_t frame;
+  int result;
+  size_t body_received;
+  size_t body_target;
+
+  uint64_t now;
+
+  while (1) {
+    now = now_microseconds();
+    if (now > next_summary_time) {
+      int countOverInterval = received - previous_received;
+      double intervalRate = countOverInterval / ((now - previous_report_time) / 1000000.0);
+      printf("%d ms: Received %d - %d since last report (%d Hz)\n",
+(int)(now - start_time) / 1000, received, countOverInterval, (int) intervalRate);
+
+      previous_received = received;
+      previous_report_time = now;
+      next_summary_time += SUMMARY_EVERY_US;
+    }
+
+    amqp_maybe_release_buffers(conn);
+    result = amqp_simple_wait_frame(conn, &frame);
+    if (result < 0)
+      return;
+
+    if (frame.frame_type != AMQP_FRAME_METHOD)
+      continue;
+
+    if (frame.payload.method.id != AMQP_BASIC_DELIVER_METHOD)
+      continue;
+
+    result = amqp_simple_wait_frame(conn, &frame);
+    if (result < 0)
+      return;
+
+    if (frame.frame_type != AMQP_FRAME_HEADER) {
+      fprintf(stderr, "Expected header!");
+      abort();
+    }
+
+    body_target = frame.payload.properties.body_size;
+    body_received = 0;
+
+    while (body_received < body_target) {
+      result = amqp_simple_wait_frame(conn, &frame);
+      if (result < 0)
+return;
+
+      if (frame.frame_type != AMQP_FRAME_BODY) {
+fprintf(stderr, "Expected body!");
+abort();
+      }
+
+      body_received += frame.payload.body_fragment.len;
+      assert(body_received <= body_target);
+    }
+
+    received++;
+  }
+}
+
+int main(int argc, char const * const *argv) {
+  char const *hostname;
+  int port;
+  char const *exchange;
+  char const *bindingkey;
+
+  int sockfd;
+  amqp_connection_state_t conn;
+
+  amqp_bytes_t queuename;
+
+  if (argc < 3) {
+    fprintf(stderr, "Usage: amqp_consumer host port\n");
+    return 1;
+  }
+
+  hostname = argv[1];
+  port = atoi(argv[2]);
+  exchange = "amq.direct"; /* argv[3]; */
+  bindingkey = "test queue"; /* argv[4]; */
+
+  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");
+
+  /*
+  {
+    amqp_queue_declare_ok_t *r = amqp_queue_declare(conn, 1, amqp_empty_bytes, 0, 0, 0, 1, amqp_empty_table);
+    die_on_amqp_error(amqp_get_rpc_reply(conn), "Declaring queue");
+    queuename = amqp_bytes_malloc_dup(r->queue);
+    if (queuename.bytes == NULL) {
+      fprintf(stderr, "Out of memory while copying queue name");
+      return 1;
+    }
+  }
+
+  amqp_queue_bind(conn, 1, queuename, amqp_cstring_bytes(exchange), amqp_cstring_bytes(bindingkey), amqp_empty_table);
+
+  die_on_amqp_error(amqp_get_rpc_reply(conn), "Binding queue");
+
+  amqp_basic_consume(conn, 1, queuename, amqp_empty_bytes, 0, 1, 0, amqp_empty_table);
+  die_on_amqp_error(amqp_get_rpc_reply(conn), "Consuming");
+  */
+
+  amqp_basic_consume(conn, 1, amqp_cstring_bytes("test queue"), amqp_empty_bytes, 0, 1, 0, amqp_empty_table);
+  die_on_amqp_error(amqp_get_rpc_reply(conn), "Consuming");
+
+  run(conn);
+
+  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;
+}
