WIP: rabbitMQ interface

build.com

@@ -2,7 +2,11 @@
     /name=(as_is,short) -
     /warn=disa=EXTRASEMI -
     /include=([-.wasd.wasd_root.src.misc], LIBCJSON, LIBRABBITMQ)
+$ cc utils.c -
+    /name=(as_is,short) -
+    /warn=disa=EXTRASEMI -
+    /include=([-.wasd.wasd_root.src.misc], LIBCJSON, LIBRABBITMQ)
 $ cc [-.wasd.wasd_root.src.misc]cgilib.c -
     /name=(as_is,short) -
     /include=([-.wasd.wasd_root.src.misc])
 $ link rmqplus.obj, cgilib.obj, sys$input:/opt
@@ -5,6 +9,7 @@
 $ cc [-.wasd.wasd_root.src.misc]cgilib.c -
     /name=(as_is,short) -
     /include=([-.wasd.wasd_root.src.misc])
 $ link rmqplus.obj, cgilib.obj, sys$input:/opt
+utils.obj
 CJSONSHR32/share
 RABBITMQSHR32/share

rmqplus.c

@@ -45,6 +45,8 @@
 #include <amqp.h>
 #include <amqp_framing.h>
 
+#include "utils.h"
+
 char Utility[] = "RMQPLUS";
 
 int Debug,
@@ -59,6 +61,18 @@
 unsigned long UnixTime;
 struct tm *UnixTmPtr;
 
+char const *amqp_hostname;
+char const *amqp_vhost;
+char const *amqp_username;
+char const *amqp_password;
+int amqp_port, status;
+char const *amqp_exchange;
+char const *amqp_routingkey;
+char const *amqp_messagebody;
+amqp_socket_t *socket = NULL;
+amqp_connection_state_t conn;
+amqp_bytes_t reply_to_queue;
+
 /*
  *  
  */
@@ -74,7 +88,34 @@
     return root;
 }
 
+
+int amqp_init() {
+    /*
+     * establish a channel that is used to connect RabbitMQ server
+     */
+
+    conn = amqp_new_connection();
+
+    socket = amqp_tcp_socket_new(conn);
+    if (!socket) {
+      die("creating TCP socket");
+    }
+
+    status = amqp_socket_open(socket, amqp_hostname, amqp_port);
+    if (status) {
+      die("opening TCP socket");
+    }
+
+    die_on_amqp_error(amqp_login(conn, amqp_vhost, 0, 131072, 0, AMQP_SASL_METHOD_PLAIN,
+                                 amqp_username, amqp_password),
+                      "Logging in");
+    amqp_channel_open(conn, 1);
+    die_on_amqp_error(amqp_get_rpc_reply(conn), "Opening channel");
+    return 1;
+}
+
+
 int rmqplus() {
     cJSON *cjson_root, *cjson_in, *cjson_out, *cjson_header, *cjson_obj;
     char *cptr, *cname, *cvalue;
   
@@ -77,11 +118,7 @@
 int rmqplus() {
     cJSON *cjson_root, *cjson_in, *cjson_out, *cjson_header, *cjson_obj;
     char *cptr, *cname, *cvalue;
   
-    /*********/
-    /* begin */
-    /*********/
-
     cjson_root = cjson_create_root();
     cjson_in = cJSON_GetObjectItem(cjson_root, "in");
     cjson_out = cJSON_GetObjectItem(cjson_root, "out");
@@ -128,12 +165,11 @@
  */
 
 main() {
-    /*********/
-    /* begin */
-    /*********/
+    char* cptr;
+
     if ((stdout = freopen("SYS$OUTPUT", "w", stdout, "ctx=bin")) == NULL)
         exit(vaxc$errno);
 
     if (getenv("RMQPLUS$DBUG"))
         Debug = 1;
 
@@ -134,9 +170,25 @@
     if ((stdout = freopen("SYS$OUTPUT", "w", stdout, "ctx=bin")) == NULL)
         exit(vaxc$errno);
 
     if (getenv("RMQPLUS$DBUG"))
         Debug = 1;
 
+    if (!(amqp_hostname = getenv("RMQPLUS_HOSTNAME")))
+        amqp_hostname = "localhost";
+
+    if (cptr = getenv("RMQPLUS_VHOST"))
+        amqp_vhost = cptr;
+    else
+        amqp_vhost = "/";
+
+    if (cptr = getenv("RMQPLUS_PORT"))
+        amqp_port = atoi(cptr);
+    else
+        amqp_port = 5672;
+
+    if (!(amqp_exchange = getenv("RMQPLUS_EXCHANGE")))
+        amqp_exchange = "";
+
     CgiLibEnvironmentSetDebug(Debug);
 
     time(&UnixTime);