Add support for ssl connection

META.json

@@ -1,10 +1,10 @@
 {
    "name": "pg_amqp",
    "abstract": "AMQP protocol support for PostgreSQL",
-   "version": "0.4.1",
+   "version": "0.4.2",
    "description": "The pg_amqp package provides the ability for postgres statements to directly publish messages to an AMQP broker.",
    "maintainer": 
-       [ "Theo Schlossnagle <[email protected]>", "Keith Fiske <[email protected]" ] 
+       [ "Theo Schlossnagle <[email protected]>", "Keith Fiske <[email protected]", "Marcello Ceschia <[email protected]>" ] 
    "license": [ "bsd", "mozilla_1_0" ],
    "generated_by": "Keith Fiske",
    "status": "stable",
@@ -17,9 +17,9 @@
     },
    "provides": {
      "amqp": {
-       "file": "sql/amqp--0.4.1.sql",
+       "file": "sql/amqp--0.4.2.sql",
         "docfile": "doc/amqp.md",
-       "version": "0.4.1",
+       "version": "0.4.2",
        "abstract": "AMQP protocol support for PostgreSQL"
      }
    },
@@ -43,4 +43,4 @@
      "rabbitmq",
      "queue"
    ]
-}
+}

README.md

@@ -1,8 +1,8 @@
 pg_amqp
-=============
 
 The pg_amqp package provides the ability for postgres statements to directly
-publish messages to an [AMQP](http://www.amqp.org/) broker.
+publish messages to an [AMQP](http://www.amqp.org/) broker. Version 0.4.2 also provides an implementation 
+for secured connections using open ssl. To build v0.4.2 openssl dev package is required.
 
 All bug reports, feature requests and general questions can be directed to the Issues section on Github. - http://github.com/omniti-labs/pg_amqp
 

amqp.control

@@ -1,6 +1,6 @@
 # amqp extension
 comment = 'AMQP protocol support for PostgreSQL'
-default_version = '0.4.1'
+default_version = '0.4.2'
 module_pathname = '$libdir/pg_amqp'
 relocatable = false
 schema = amqp

sql/amqp--0.4.2.sql

@@ -0,0 +1,91 @@
+CREATE TABLE @[email protected] (
+  broker_id serial NOT NULL,
+  host text NOT NULL,
+  port integer NOT NULL DEFAULT 5672,
+  vhost text,
+  username text,
+  password text,
+  requiressl boolean DEFAULT false,
+  verify_cert boolean DEFAULT true,
+  verify_cn boolean DEFAULT true,
+  cert text,
+  key text,
+  key_password character varying,
+  ca text,
+  PRIMARY KEY (broker_id, host, port)
+);
+
+
+CREATE FUNCTION @[email protected]_publish(
+    broker_id integer
+    , exchange varchar
+    , routing_key varchar
+    , message varchar
+    , delivery_mode integer default null
+    , content_type varchar default null
+    , reply_to varchar default null
+    , correlation_id varchar default null
+)
+
+RETURNS boolean AS 'pg_amqp.so', 'pg_amqp_autonomous_publish'
+LANGUAGE C IMMUTABLE;
+
+COMMENT ON FUNCTION @[email protected]_publish(integer, varchar, varchar, varchar, integer, varchar, varchar, varchar) IS
+'Works as amqp.publish does, but the message is published immediately irrespective of the
+current transaction state.  PostgreSQL commit and rollback at a later point will have no
+effect on this message being sent to AMQP.';
+
+
+CREATE FUNCTION amqp.disconnect(broker_id integer)
+RETURNS void AS 'pg_amqp.so', 'pg_amqp_disconnect'
+LANGUAGE C IMMUTABLE STRICT;
+
+COMMENT ON FUNCTION amqp.disconnect(integer) IS
+'Explicitly disconnect the specified (broker_id) if it is current connected. Broker
+connections, once established, live until the PostgreSQL backend terminated.  This
+allows for more precise control over that.
+select amqp.disconnect(broker_id) from amqp.broker
+will disconnect any brokers that may be connected.';
+
+
+CREATE FUNCTION amqp.exchange_declare(
+    broker_id integer
+    , exchange varchar 
+    , exchange_type varchar
+    , passive boolean
+    , durable boolean
+    , auto_delete boolean DEFAULT false
+)
+RETURNS boolean AS 'pg_amqp.so', 'pg_amqp_exchange_declare'
+LANGUAGE C IMMUTABLE;
+
+COMMENT ON FUNCTION amqp.exchange_declare(integer, varchar, varchar, boolean, boolean, boolean) IS
+'Declares a exchange (broker_id, exchange_name, exchange_type, passive, durable, auto_delete)
+auto_delete should be set to false (default) as unexpected errors can cause disconnect/reconnect which
+would trigger the auto deletion of the exchange.';
+
+
+CREATE FUNCTION @[email protected](
+    broker_id integer
+    , exchange varchar
+    , routing_key varchar
+    , message varchar
+    , delivery_mode integer default null
+    , content_type varchar default null
+    , reply_to varchar default null
+    , correlation_id varchar default null
+    , headers varchar[][2] default null
+    , properties varchar[][2] default null
+)
+RETURNS boolean AS 'pg_amqp.so', 'pg_amqp_publish'
+LANGUAGE C IMMUTABLE;
+
+COMMENT ON FUNCTION @[email protected](integer, varchar, varchar, varchar, integer, varchar, varchar, varchar, varchar[][2], varchar[][2]) IS
+'Publishes a message (broker_id, exchange, routing_key, message, delivery_mode, content_type, reply_to, correlation_id, headers, properties). 
+The message will only be published if the containing PostgreSQL transaction successfully commits.  
+Under certain circumstances, the AMQP commit might fail.  In this case, a WARNING is emitted. 
+The last four parameters are optional and set the following message properties: 
+delivery_mode (either 1 or 2), content_type, reply_to and correlation_id.
+
+Publish returns a boolean indicating if the publish command was successful.  Note that as
+AMQP publish is asynchronous, you may find out later it was unsuccessful.';

sql/functions/functions.sql

@@ -56,17 +56,18 @@ CREATE FUNCTION @[email protected](
     , content_type varchar default null
     , reply_to varchar default null
     , correlation_id varchar default null
+    , headers varchar[][2] default null
+    , properties varchar[][2] default null
 )
 RETURNS boolean AS 'pg_amqp.so', 'pg_amqp_publish'
 LANGUAGE C IMMUTABLE;
 
-COMMENT ON FUNCTION @[email protected](integer, varchar, varchar, varchar, integer, varchar, varchar, varchar) IS
-'Publishes a message (broker_id, exchange, routing_key, message). 
+COMMENT ON FUNCTION @[email protected](integer, varchar, varchar, varchar, integer, varchar, varchar, varchar, varchar[][2], varchar[][2]) IS
+'Publishes a message (broker_id, exchange, routing_key, message, delivery_mode, content_type, reply_to, correlation_id, headers, properties). 
 The message will only be published if the containing PostgreSQL transaction successfully commits.  
 Under certain circumstances, the AMQP commit might fail.  In this case, a WARNING is emitted. 
 The last four parameters are optional and set the following message properties: 
 delivery_mode (either 1 or 2), content_type, reply_to and correlation_id.
 
 Publish returns a boolean indicating if the publish command was successful.  Note that as
 AMQP publish is asynchronous, you may find out later it was unsuccessful.';
-

sql/tables/tables.sql

@@ -5,6 +5,13 @@ CREATE TABLE @[email protected] (
   vhost text,
   username text,
   password text,
+  requiressl boolean DEFAULT false,
+  verify_cert boolean DEFAULT true,
+  verify_cn boolean DEFAULT true,
+  cert text,
+  key text,
+  key_password character varying,
+  ca text,
   PRIMARY KEY (broker_id, host, port)
 );
 

src/librabbitmq/amqp.h

@@ -98,6 +98,11 @@ typedef enum amqp_response_type_enum_ {
   AMQP_RESPONSE_SERVER_EXCEPTION
 } amqp_response_type_enum;
 
+typedef enum amqp_ssl_flags_type_enum_ {
+  AMQP_SSL_FLAG_VERIFY		= 1 << 0,
+  AMQP_SSL_FLAG_CHECK_CN	= 1 << 2,
+} amqp_ssl_flags_type_enum;
+
 typedef struct amqp_rpc_reply_t_ {
   amqp_response_type_enum reply_type;
   amqp_method_t reply;
@@ -113,8 +118,7 @@ typedef enum amqp_sasl_method_enum_ {
 
 typedef struct amqp_basic_return_t_ amqp_basic_return_t;
 typedef int (*amqp_output_fn_t)(void *context, void *buffer, size_t count);
-typedef void (*amqp_basic_return_fn_t)(amqp_channel_t, amqp_basic_return_t *,
-                                       void *);
+typedef void (*amqp_basic_return_fn_t)(amqp_channel_t, amqp_basic_return_t *, void *);
 
 
 /* Opaque struct. */
@@ -140,6 +144,7 @@ extern amqp_bytes_t amqp_bytes_malloc_dup(amqp_bytes_t src);
     }						\
   })
 
+extern amqp_connection_state_t amqp_new_ssl_connection(const char *certificate, const char *key, const char *password, const char *ca, unsigned short flags);
 extern amqp_connection_state_t amqp_new_connection(void);
 extern int amqp_get_sockfd(amqp_connection_state_t state);
 extern void amqp_set_sockfd(amqp_connection_state_t state,
@@ -173,8 +178,7 @@ extern int amqp_send_frame_to(amqp_connection_state_t state,
 
 extern int amqp_table_entry_cmp(void const *entry1, void const *entry2);
 
-extern int amqp_open_socket(char const *hostname, int portnumber,
-                            struct timeval *timeout);
+extern int amqp_connect(amqp_connection_state_t state, char const *hostname, int portnumber, struct timeval *timeout);
 
 extern int amqp_send_header(amqp_connection_state_t state);
 extern int amqp_send_header_to(amqp_connection_state_t state,
@@ -343,6 +347,7 @@ extern amqp_boolean_t amqp_data_in_buffer(amqp_connection_state_t state);
  */
 extern amqp_rpc_reply_t *amqp_get_rpc_reply(void);
 
+
 #ifdef __cplusplus
 }
 #endif

src/librabbitmq/amqp_api.c

@@ -4,6 +4,8 @@
 #include <stdint.h>
 #include <errno.h>
 
+#define DISABLE_THREADS
+
 #ifndef DISABLE_THREADS
 #include <pthread.h>
 #endif
@@ -103,14 +105,9 @@ amqp_rpc_reply_t amqp_channel_close(amqp_connection_state_t state,
 			 code, amqp_cstring_bytes(codestr), 0, 0);
 }
 
-amqp_rpc_reply_t amqp_connection_close(amqp_connection_state_t state,
-				       int code)
+amqp_rpc_reply_t amqp_connection_close(amqp_connection_state_t state, int code)
 {
-  char codestr[13];
-  snprintf(codestr, sizeof(codestr), "%d", code);
-  return AMQP_SIMPLE_RPC(state, 0, CONNECTION, CLOSE, CLOSE_OK,
-			 amqp_connection_close_t,
-			 code, amqp_cstring_bytes(codestr), 0, 0);
+  return state->close_connection(state, code);
 }
 
 amqp_exchange_declare_ok_t *amqp_exchange_declare(amqp_connection_state_t state,