From da9c2c109ad9740177adfc93e5e92cba92c56134 Mon Sep 17 00:00:00 2001 From: Alan Antonuk Date: Wed, 3 Jul 2013 12:35:27 -0700 Subject: Make connection the socket object owner Improve the socket interface by making the amqp_connection_state_t object the amqp_socket_t owner, and tie its lifetime to the connection's lifetime. This prevents a class of silly errors where the socket object isn't freed, or the socket object is assigned to two different connection objects --- examples/amqp_bind.c | 3 +-- examples/amqp_consumer.c | 3 +-- examples/amqp_exchange_declare.c | 3 +-- examples/amqp_listen.c | 3 +-- examples/amqp_listenq.c | 3 +-- examples/amqp_producer.c | 3 +-- examples/amqp_rpc_sendstring_client.c | 3 +-- examples/amqp_sendstring.c | 3 +-- examples/amqp_unbind.c | 3 +-- examples/amqps_bind.c | 3 +-- examples/amqps_consumer.c | 3 +-- examples/amqps_exchange_declare.c | 3 +-- examples/amqps_listen.c | 3 +-- examples/amqps_listenq.c | 3 +-- examples/amqps_producer.c | 3 +-- examples/amqps_sendstring.c | 3 +-- examples/amqps_unbind.c | 3 +-- 17 files changed, 17 insertions(+), 34 deletions(-) (limited to 'examples') diff --git a/examples/amqp_bind.c b/examples/amqp_bind.c index 765e746..de1e0a5 100644 --- a/examples/amqp_bind.c +++ b/examples/amqp_bind.c @@ -68,7 +68,7 @@ int main(int argc, char const *const *argv) conn = amqp_new_connection(); - socket = amqp_tcp_socket_new(); + socket = amqp_tcp_socket_new(conn); if (!socket) { die("creating TCP socket"); } @@ -78,7 +78,6 @@ int main(int argc, char const *const *argv) die("opening TCP socket"); } - amqp_set_socket(conn, socket); die_on_amqp_error(amqp_login(conn, "/", 0, 131072, 0, AMQP_SASL_METHOD_PLAIN, "guest", "guest"), "Logging in"); amqp_channel_open(conn, 1); diff --git a/examples/amqp_consumer.c b/examples/amqp_consumer.c index 72bf654..21a5b48 100644 --- a/examples/amqp_consumer.c +++ b/examples/amqp_consumer.c @@ -146,7 +146,7 @@ int main(int argc, char const *const *argv) conn = amqp_new_connection(); - socket = amqp_tcp_socket_new(); + socket = amqp_tcp_socket_new(conn); if (!socket) { die("creating TCP socket"); } @@ -156,7 +156,6 @@ int main(int argc, char const *const *argv) die("opening TCP socket"); } - amqp_set_socket(conn, socket); die_on_amqp_error(amqp_login(conn, "/", 0, 131072, 0, AMQP_SASL_METHOD_PLAIN, "guest", "guest"), "Logging in"); amqp_channel_open(conn, 1); diff --git a/examples/amqp_exchange_declare.c b/examples/amqp_exchange_declare.c index 55860e5..9a20a62 100644 --- a/examples/amqp_exchange_declare.c +++ b/examples/amqp_exchange_declare.c @@ -66,7 +66,7 @@ int main(int argc, char const *const *argv) conn = amqp_new_connection(); - socket = amqp_tcp_socket_new(); + socket = amqp_tcp_socket_new(conn); if (!socket) { die("creating TCP socket"); } @@ -76,7 +76,6 @@ int main(int argc, char const *const *argv) die("opening TCP socket"); } - amqp_set_socket(conn, socket); die_on_amqp_error(amqp_login(conn, "/", 0, 131072, 0, AMQP_SASL_METHOD_PLAIN, "guest", "guest"), "Logging in"); amqp_channel_open(conn, 1); diff --git a/examples/amqp_listen.c b/examples/amqp_listen.c index 9385c17..bf5b716 100644 --- a/examples/amqp_listen.c +++ b/examples/amqp_listen.c @@ -70,7 +70,7 @@ int main(int argc, char const *const *argv) conn = amqp_new_connection(); - socket = amqp_tcp_socket_new(); + socket = amqp_tcp_socket_new(conn); if (!socket) { die("creating TCP socket"); } @@ -80,7 +80,6 @@ int main(int argc, char const *const *argv) die("opening TCP socket"); } - amqp_set_socket(conn, socket); die_on_amqp_error(amqp_login(conn, "/", 0, 131072, 0, AMQP_SASL_METHOD_PLAIN, "guest", "guest"), "Logging in"); amqp_channel_open(conn, 1); diff --git a/examples/amqp_listenq.c b/examples/amqp_listenq.c index 54c1189..e76cdb1 100644 --- a/examples/amqp_listenq.c +++ b/examples/amqp_listenq.c @@ -66,7 +66,7 @@ int main(int argc, char const *const *argv) conn = amqp_new_connection(); - socket = amqp_tcp_socket_new(); + socket = amqp_tcp_socket_new(conn); if (!socket) { die("creating TCP socket"); } @@ -76,7 +76,6 @@ int main(int argc, char const *const *argv) die("opening TCP socket"); } - amqp_set_socket(conn, socket); die_on_amqp_error(amqp_login(conn, "/", 0, 131072, 0, AMQP_SASL_METHOD_PLAIN, "guest", "guest"), "Logging in"); amqp_channel_open(conn, 1); diff --git a/examples/amqp_producer.c b/examples/amqp_producer.c index efa1a20..948d8f6 100644 --- a/examples/amqp_producer.c +++ b/examples/amqp_producer.c @@ -130,7 +130,7 @@ int main(int argc, char const *const *argv) conn = amqp_new_connection(); - socket = amqp_tcp_socket_new(); + socket = amqp_tcp_socket_new(conn); if (!socket) { die("creating TCP socket"); } @@ -140,7 +140,6 @@ int main(int argc, char const *const *argv) die("opening TCP socket"); } - amqp_set_socket(conn, socket); die_on_amqp_error(amqp_login(conn, "/", 0, 131072, 0, AMQP_SASL_METHOD_PLAIN, "guest", "guest"), "Logging in"); amqp_channel_open(conn, 1); diff --git a/examples/amqp_rpc_sendstring_client.c b/examples/amqp_rpc_sendstring_client.c index 6688195..84e7fdd 100644 --- a/examples/amqp_rpc_sendstring_client.c +++ b/examples/amqp_rpc_sendstring_client.c @@ -75,7 +75,7 @@ int main(int argc, char *argv[]) conn = amqp_new_connection(); - socket = amqp_tcp_socket_new(); + socket = amqp_tcp_socket_new(conn); if (!socket) { die("creating TCP socket"); } @@ -85,7 +85,6 @@ int main(int argc, char *argv[]) die("opening TCP socket"); } - amqp_set_socket(conn, socket); die_on_amqp_error(amqp_login(conn, "/", 0, 131072, 0, AMQP_SASL_METHOD_PLAIN, "guest", "guest"), "Logging in"); amqp_channel_open(conn, 1); diff --git a/examples/amqp_sendstring.c b/examples/amqp_sendstring.c index 0b64024..bc48054 100644 --- a/examples/amqp_sendstring.c +++ b/examples/amqp_sendstring.c @@ -68,7 +68,7 @@ int main(int argc, char const *const *argv) conn = amqp_new_connection(); - socket = amqp_tcp_socket_new(); + socket = amqp_tcp_socket_new(conn); if (!socket) { die("creating TCP socket"); } @@ -78,7 +78,6 @@ int main(int argc, char const *const *argv) die("opening TCP socket"); } - amqp_set_socket(conn, socket); die_on_amqp_error(amqp_login(conn, "/", 0, 131072, 0, AMQP_SASL_METHOD_PLAIN, "guest", "guest"), "Logging in"); amqp_channel_open(conn, 1); diff --git a/examples/amqp_unbind.c b/examples/amqp_unbind.c index 7948d0b..1ca3e83 100644 --- a/examples/amqp_unbind.c +++ b/examples/amqp_unbind.c @@ -68,7 +68,7 @@ int main(int argc, char const *const *argv) conn = amqp_new_connection(); - socket = amqp_tcp_socket_new(); + socket = amqp_tcp_socket_new(conn); if (!socket) { die("creating TCP socket"); } @@ -78,7 +78,6 @@ int main(int argc, char const *const *argv) die("opening TCP socket"); } - amqp_set_socket(conn, socket); die_on_amqp_error(amqp_login(conn, "/", 0, 131072, 0, AMQP_SASL_METHOD_PLAIN, "guest", "guest"), "Logging in"); amqp_channel_open(conn, 1); diff --git a/examples/amqps_bind.c b/examples/amqps_bind.c index fbde025..35c845f 100644 --- a/examples/amqps_bind.c +++ b/examples/amqps_bind.c @@ -71,7 +71,7 @@ int main(int argc, char const *const *argv) conn = amqp_new_connection(); - socket = amqp_ssl_socket_new(); + socket = amqp_ssl_socket_new(conn); if (!socket) { die("creating SSL/TLS socket"); } @@ -95,7 +95,6 @@ int main(int argc, char const *const *argv) die("opening SSL/TLS connection"); } - amqp_set_socket(conn, socket); die_on_amqp_error(amqp_login(conn, "/", 0, 131072, 0, AMQP_SASL_METHOD_PLAIN, "guest", "guest"), "Logging in"); amqp_channel_open(conn, 1); diff --git a/examples/amqps_consumer.c b/examples/amqps_consumer.c index 137457f..fff6677 100644 --- a/examples/amqps_consumer.c +++ b/examples/amqps_consumer.c @@ -148,7 +148,7 @@ int main(int argc, char const *const *argv) conn = amqp_new_connection(); - socket = amqp_ssl_socket_new(); + socket = amqp_ssl_socket_new(conn); if (!socket) { die("creating SSL/TLS socket"); } @@ -172,7 +172,6 @@ int main(int argc, char const *const *argv) die("opening SSL/TLS connection"); } - amqp_set_socket(conn, socket); die_on_amqp_error(amqp_login(conn, "/", 0, 131072, 0, AMQP_SASL_METHOD_PLAIN, "guest", "guest"), "Logging in"); amqp_channel_open(conn, 1); diff --git a/examples/amqps_exchange_declare.c b/examples/amqps_exchange_declare.c index bae2f57..85a29aa 100644 --- a/examples/amqps_exchange_declare.c +++ b/examples/amqps_exchange_declare.c @@ -69,7 +69,7 @@ int main(int argc, char const *const *argv) conn = amqp_new_connection(); - socket = amqp_ssl_socket_new(); + socket = amqp_ssl_socket_new(conn); if (!socket) { die("creating SSL/TLS socket"); } @@ -93,7 +93,6 @@ int main(int argc, char const *const *argv) die("opening SSL/TLS connection"); } - amqp_set_socket(conn, socket); die_on_amqp_error(amqp_login(conn, "/", 0, 131072, 0, AMQP_SASL_METHOD_PLAIN, "guest", "guest"), "Logging in"); amqp_channel_open(conn, 1); diff --git a/examples/amqps_listen.c b/examples/amqps_listen.c index 0e45162..a5eb692 100644 --- a/examples/amqps_listen.c +++ b/examples/amqps_listen.c @@ -73,7 +73,7 @@ int main(int argc, char const *const *argv) conn = amqp_new_connection(); - socket = amqp_ssl_socket_new(); + socket = amqp_ssl_socket_new(conn); if (!socket) { die("creating SSL/TLS socket"); } @@ -97,7 +97,6 @@ int main(int argc, char const *const *argv) die("opening SSL/TLS connection"); } - amqp_set_socket(conn, socket); die_on_amqp_error(amqp_login(conn, "/", 0, 131072, 0, AMQP_SASL_METHOD_PLAIN, "guest", "guest"), "Logging in"); amqp_channel_open(conn, 1); diff --git a/examples/amqps_listenq.c b/examples/amqps_listenq.c index 321c6a3..0210d88 100644 --- a/examples/amqps_listenq.c +++ b/examples/amqps_listenq.c @@ -69,7 +69,7 @@ int main(int argc, char const *const *argv) conn = amqp_new_connection(); - socket = amqp_ssl_socket_new(); + socket = amqp_ssl_socket_new(conn); if (!socket) { die("creating SSL/TLS socket"); } @@ -93,7 +93,6 @@ int main(int argc, char const *const *argv) die("opening SSL/TLS connection"); } - amqp_set_socket(conn, socket); die_on_amqp_error(amqp_login(conn, "/", 0, 131072, 0, AMQP_SASL_METHOD_PLAIN, "guest", "guest"), "Logging in"); amqp_channel_open(conn, 1); diff --git a/examples/amqps_producer.c b/examples/amqps_producer.c index f8f6dc6..25f850b 100644 --- a/examples/amqps_producer.c +++ b/examples/amqps_producer.c @@ -133,7 +133,7 @@ int main(int argc, char const *const *argv) conn = amqp_new_connection(); - socket = amqp_ssl_socket_new(); + socket = amqp_ssl_socket_new(conn); if (!socket) { die("creating SSL/TLS socket"); } @@ -157,7 +157,6 @@ int main(int argc, char const *const *argv) die("opening SSL/TLS connection"); } - amqp_set_socket(conn, socket); die_on_amqp_error(amqp_login(conn, "/", 0, 131072, 0, AMQP_SASL_METHOD_PLAIN, "guest", "guest"), "Logging in"); amqp_channel_open(conn, 1); diff --git a/examples/amqps_sendstring.c b/examples/amqps_sendstring.c index 7465ef2..fe3ac67 100644 --- a/examples/amqps_sendstring.c +++ b/examples/amqps_sendstring.c @@ -71,7 +71,7 @@ int main(int argc, char const *const *argv) conn = amqp_new_connection(); - socket = amqp_ssl_socket_new(); + socket = amqp_ssl_socket_new(conn); if (!socket) { die("creating SSL/TLS socket"); } @@ -95,7 +95,6 @@ int main(int argc, char const *const *argv) die("opening SSL/TLS connection"); } - amqp_set_socket(conn, socket); die_on_amqp_error(amqp_login(conn, "/", 0, 131072, 0, AMQP_SASL_METHOD_PLAIN, "guest", "guest"), "Logging in"); amqp_channel_open(conn, 1); diff --git a/examples/amqps_unbind.c b/examples/amqps_unbind.c index bae017d..7f4737e 100644 --- a/examples/amqps_unbind.c +++ b/examples/amqps_unbind.c @@ -71,7 +71,7 @@ int main(int argc, char const *const *argv) conn = amqp_new_connection(); - socket = amqp_ssl_socket_new(); + socket = amqp_ssl_socket_new(conn); if (!socket) { die("creating SSL/TLS socket"); } @@ -95,7 +95,6 @@ int main(int argc, char const *const *argv) die("opening SSL/TLS connection"); } - amqp_set_socket(conn, socket); die_on_amqp_error(amqp_login(conn, "/", 0, 131072, 0, AMQP_SASL_METHOD_PLAIN, "guest", "guest"), "Logging in"); amqp_channel_open(conn, 1); -- cgit v1.2.1 From 6ad770dc62f76fa0625d277b521a120b549d9fc2 Mon Sep 17 00:00:00 2001 From: zaq178miami Date: Sun, 23 Jun 2013 19:36:10 +0300 Subject: Add nonblocking connect support --- examples/CMakeLists.txt | 6 ++ examples/amqp_connect_timeout.c | 110 ++++++++++++++++++++++++++++++++ examples/amqps_connect_timeout.c | 132 +++++++++++++++++++++++++++++++++++++++ 3 files changed, 248 insertions(+) create mode 100644 examples/amqp_connect_timeout.c create mode 100644 examples/amqps_connect_timeout.c (limited to 'examples') diff --git a/examples/CMakeLists.txt b/examples/CMakeLists.txt index a184cf6..8dcdcf4 100644 --- a/examples/CMakeLists.txt +++ b/examples/CMakeLists.txt @@ -28,6 +28,9 @@ target_link_libraries(amqp_listen ${RMQ_LIBRARY_TARGET}) add_executable(amqp_producer amqp_producer.c ${COMMON_SRCS}) target_link_libraries(amqp_producer ${RMQ_LIBRARY_TARGET}) +add_executable(amqp_connect_timeout amqp_connect_timeout.c ${COMMON_SRCS}) +target_link_libraries(amqp_connect_timeout ${RMQ_LIBRARY_TARGET}) + add_executable(amqp_consumer amqp_consumer.c ${COMMON_SRCS}) target_link_libraries(amqp_consumer ${RMQ_LIBRARY_TARGET}) @@ -41,6 +44,9 @@ add_executable(amqp_listenq amqp_listenq.c ${COMMON_SRCS}) target_link_libraries(amqp_listenq ${RMQ_LIBRARY_TARGET}) if (ENABLE_SSL_SUPPORT) +add_executable(amqps_connect_timeout amqps_connect_timeout.c ${COMMON_SRCS}) +target_link_libraries(amqps_connect_timeout ${RMQ_LIBRARY_TARGET}) + add_executable(amqps_sendstring amqps_sendstring.c ${COMMON_SRCS}) target_link_libraries(amqps_sendstring ${RMQ_LIBRARY_TARGET}) diff --git a/examples/amqp_connect_timeout.c b/examples/amqp_connect_timeout.c new file mode 100644 index 0000000..c2bd5ec --- /dev/null +++ b/examples/amqp_connect_timeout.c @@ -0,0 +1,110 @@ +/* vim:set ft=c ts=2 sw=2 sts=2 et cindent: */ +/* + * ***** BEGIN LICENSE BLOCK ***** + * Version: MIT + * + * Portions created by Alan Antonuk are Copyright (c) 2012-2013 + * Alan Antonuk. All Rights Reserved. + * + * Portions created by Bogdan Padalko are Copyright (c) 2013. + * Bogdan Padalko. All Rights Reserved. + * + * Portions created by VMware are Copyright (c) 2007-2012 VMware, Inc. + * All Rights Reserved. + * + * Portions created by Tony Garnock-Jones are Copyright (c) 2009-2010 + * VMware, Inc. and Tony Garnock-Jones. All Rights Reserved. + * + * Permission is hereby granted, free of charge, to any person + * obtaining a copy of this software and associated documentation + * files (the "Software"), to deal in the Software without + * restriction, including without limitation the rights to use, copy, + * modify, merge, publish, distribute, sublicense, and/or sell copies + * of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be + * included in all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, + * EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF + * MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND + * NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS + * BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN + * ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN + * CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + * SOFTWARE. + * ***** END LICENSE BLOCK ***** + */ + +#include +#include +#include + +#include +#include +#include + +#include + +#ifdef _WIN32 +# ifndef WIN32_LEAN_AND_MEAN +# define WIN32_LEAN_AND_MEAN +# endif +# include +#else +# include +#endif + +#include "utils.h" + +int main(int argc, char const *const *argv) +{ + char const *hostname; + int port; + amqp_socket_t *socket; + amqp_connection_state_t conn; + struct timeval *tv; + + if (argc < 3) { + fprintf(stderr, "Usage: amqp_connect_timeout host port [timeout_sec [timeout_usec=0]]\n"); + return 1; + } + + if (argc > 3) { + tv = malloc(sizeof(struct timeval)); + + tv->tv_sec = atoi(argv[3]); + + if (argc > 4 ) { + tv->tv_usec = atoi(argv[4]); + } else { + tv->tv_usec = 0; + } + + } else { + tv = NULL; + } + + + hostname = argv[1]; + port = atoi(argv[2]); + + conn = amqp_new_connection(); + + socket = amqp_tcp_socket_new(conn); + + if (!socket) { + die("creating TCP socket"); + } + + die_on_error(amqp_socket_open_noblock(socket, hostname, port, tv), "opening TCP socket"); + + die_on_amqp_error(amqp_login(conn, "/", 0, 131072, 0, AMQP_SASL_METHOD_PLAIN, "guest", "guest"), "Logging in"); + + die_on_amqp_error(amqp_connection_close(conn, AMQP_REPLY_SUCCESS), "Closing connection"); + die_on_error(amqp_destroy_connection(conn), "Ending connection"); + + printf ("Done\n"); + return 0; +} diff --git a/examples/amqps_connect_timeout.c b/examples/amqps_connect_timeout.c new file mode 100644 index 0000000..89c00ac --- /dev/null +++ b/examples/amqps_connect_timeout.c @@ -0,0 +1,132 @@ +/* vim:set ft=c ts=2 sw=2 sts=2 et cindent: */ +/* + * ***** BEGIN LICENSE BLOCK ***** + * Version: MIT + * + * Portions created by Alan Antonuk are Copyright (c) 2012-2013 + * Alan Antonuk. All Rights Reserved. + * + * Portions created by Mike Steinert are Copyright (c) 2012-2013 + * Mike Steinert. All Rights Reserved. + * + * Portions created by Bogdan Padalko are Copyright (c) 2013. + * Bogdan Padalko. All Rights Reserved. + * + * Portions created by VMware are Copyright (c) 2007-2012 VMware, Inc. + * All Rights Reserved. + * + * Portions created by Tony Garnock-Jones are Copyright (c) 2009-2010 + * VMware, Inc. and Tony Garnock-Jones. All Rights Reserved. + * + * Permission is hereby granted, free of charge, to any person + * obtaining a copy of this software and associated documentation + * files (the "Software"), to deal in the Software without + * restriction, including without limitation the rights to use, copy, + * modify, merge, publish, distribute, sublicense, and/or sell copies + * of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be + * included in all copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, + * EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF + * MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND + * NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS + * BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN + * ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN + * CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + * SOFTWARE. + * ***** END LICENSE BLOCK ***** + */ + +#include +#include +#include + +#include +#include + +#include + +#ifdef _WIN32 +# ifndef WIN32_LEAN_AND_MEAN +# define WIN32_LEAN_AND_MEAN +# endif +# include +#else +# include +#endif + +#include "utils.h" + + +int main(int argc, char const *const *argv) +{ + char const nofile[2] = "-"; + char const *hostname; + int port, status; + amqp_socket_t *socket; + amqp_connection_state_t conn; + struct timeval *tv; + + if (argc < 3) { + fprintf(stderr, "Usage: amqps_connect_timeout host port " + "[cacert.pem [key.pem cert.pem [timeout_sec [timeout_usec=0]]]]\n"); + return 1; + } + + hostname = argv[1]; + port = atoi(argv[2]); + + if (argc > 6) { + tv = malloc(sizeof(struct timeval)); + + tv->tv_sec = atoi(argv[6]); + + if (argc > 7 ) { + tv->tv_usec = atoi(argv[7]); + } else { + tv->tv_usec = 0; + } + + } else { + tv = NULL; + } + + conn = amqp_new_connection(); + + socket = amqp_ssl_socket_new(conn); + if (!socket) { + die("creating SSL/TLS socket"); + } + + if (argc > 3 && strcmp(nofile, argv[3])) { + die_on_error(amqp_ssl_socket_set_cacert(socket, argv[3]), "setting CA certificate"); + } + + if (argc > 5) { + if (!strcmp(nofile, argv[5]) && !strcmp(nofile, argv[4])) { + status = 0; + } else if (!strcmp(nofile, argv[5]) || !strcmp(nofile, argv[4])) { + status = -1; + } else { + status = amqp_ssl_socket_set_key(socket, argv[5], argv[4]); + } + + if (status) { + die("setting client key"); + } + } + + die_on_error(amqp_socket_open_noblock(socket, hostname, port, tv), "opening SSL/TLS connection"); + + die_on_amqp_error(amqp_login(conn, "/", 0, 131072, 0, AMQP_SASL_METHOD_PLAIN, "guest", "guest"), + "Logging in"); + + die_on_amqp_error(amqp_connection_close(conn, AMQP_REPLY_SUCCESS), "Closing connection"); + die_on_error(amqp_destroy_connection(conn), "Ending connection"); + + printf ("Done\n"); + return 0; +} -- cgit v1.2.1 From 33ebeede97a81c2e82ac0c3a6d88d4db0695bf29 Mon Sep 17 00:00:00 2001 From: Alan Antonuk Date: Wed, 26 Jun 2013 14:23:16 -0700 Subject: Add a high level API for consuming messages --- examples/amqp_consumer.c | 99 ++++++++++++++++++++++++++++------------------- examples/amqp_listen.c | 76 +++++++----------------------------- examples/amqp_listenq.c | 78 +++++++------------------------------ examples/amqps_consumer.c | 99 ++++++++++++++++++++++++++++------------------- examples/amqps_listen.c | 76 +++++++----------------------------- examples/amqps_listenq.c | 78 +++++++------------------------------ 6 files changed, 170 insertions(+), 336 deletions(-) (limited to 'examples') diff --git a/examples/amqp_consumer.c b/examples/amqp_consumer.c index 21a5b48..62da0ca 100644 --- a/examples/amqp_consumer.c +++ b/examples/amqp_consumer.c @@ -58,13 +58,13 @@ static void run(amqp_connection_state_t conn) 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) { + amqp_rpc_reply_t ret; + amqp_envelope_t envelope; + now = now_microseconds(); if (now > next_summary_time) { int countOverInterval = received - previous_received; @@ -78,45 +78,64 @@ static void run(amqp_connection_state_t conn) } 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(); + ret = amqp_consume_message(conn, &envelope, NULL, 0); + + if (AMQP_RESPONSE_NORMAL != ret.reply_type) { + if (AMQP_RESPONSE_LIBRARY_EXCEPTION == ret.reply_type && + AMQP_STATUS_UNEXPECTED_STATE == ret.library_error) { + if (AMQP_STATUS_OK != amqp_simple_wait_frame(conn, &frame)) { + return; + } + + if (AMQP_FRAME_METHOD == frame.frame_type) { + switch (frame.payload.method.id) { + case AMQP_BASIC_ACK_METHOD: + /* if we've turned publisher confirms on, and we've published a message + * here is a message being confirmed + */ + + break; + case AMQP_BASIC_RETURN_METHOD: + /* if a published message couldn't be routed and the mandatory flag was set + * this is what would be returned. The message then needs to be read. + */ + { + amqp_message_t message; + ret = amqp_read_message(conn, frame.channel, &message, 0); + if (AMQP_RESPONSE_NORMAL != ret.reply_type) { + return; + } + + amqp_destroy_message(&message); + } + + break; + + case AMQP_CHANNEL_CLOSE_METHOD: + /* a channel.close method happens when a channel exception occurs, this + * can happen by publishing to an exchange that doesn't exist for example + * + * In this case you would need to open another channel redeclare any queues + * that were declared auto-delete, and restart any consumers that were attached + * to the previous channel + */ + return; + + case AMQP_CONNECTION_CLOSE_METHOD: + /* a connection.close method happens when a connection exception occurs, + * this can happen by trying to use a channel that isn't open for example. + * + * In this case the whole connection must be restarted. + */ + return; + + default: + fprintf(stderr ,"An unexpected method was received %d\n", frame.payload.method.id); + return; + } + } } - body_received += frame.payload.body_fragment.len; - assert(body_received <= body_target); } received++; diff --git a/examples/amqp_listen.c b/examples/amqp_listen.c index bf5b716..ca7d538 100644 --- a/examples/amqp_listen.c +++ b/examples/amqp_listen.c @@ -104,80 +104,30 @@ int main(int argc, char const *const *argv) die_on_amqp_error(amqp_get_rpc_reply(conn), "Consuming"); { - amqp_frame_t frame; - int result; - - amqp_basic_deliver_t *d; - amqp_basic_properties_t *p; - size_t body_target; - size_t body_received; - while (1) { + amqp_rpc_reply_t res; + amqp_envelope_t envelope; + amqp_maybe_release_buffers(conn); - result = amqp_simple_wait_frame(conn, &frame); - printf("Result %d\n", result); - if (result < 0) { - break; - } - printf("Frame type %d, channel %d\n", frame.frame_type, frame.channel); - if (frame.frame_type != AMQP_FRAME_METHOD) { - continue; - } + res = amqp_consume_message(conn, &envelope, NULL, 0); - printf("Method %s\n", amqp_method_name(frame.payload.method.id)); - if (frame.payload.method.id != AMQP_BASIC_DELIVER_METHOD) { - continue; + if (AMQP_RESPONSE_NORMAL != res.reply_type) { + break; } - d = (amqp_basic_deliver_t *) frame.payload.method.decoded; printf("Delivery %u, exchange %.*s routingkey %.*s\n", - (unsigned) d->delivery_tag, - (int) d->exchange.len, (char *) d->exchange.bytes, - (int) d->routing_key.len, (char *) d->routing_key.bytes); - - result = amqp_simple_wait_frame(conn, &frame); - if (result < 0) { - break; - } + (unsigned) envelope.delivery_tag, + (int) envelope.exchange.len, (char *) envelope.exchange.bytes, + (int) envelope.routing_key.len, (char *) envelope.routing_key.bytes); - if (frame.frame_type != AMQP_FRAME_HEADER) { - fprintf(stderr, "Expected header!"); - abort(); - } - p = (amqp_basic_properties_t *) frame.payload.properties.decoded; - if (p->_flags & AMQP_BASIC_CONTENT_TYPE_FLAG) { + if (envelope.message.properties._flags & AMQP_BASIC_CONTENT_TYPE_FLAG) { printf("Content-type: %.*s\n", - (int) p->content_type.len, (char *) p->content_type.bytes); + (int) envelope.message.properties.content_type.len, + (char *) envelope.message.properties.content_type.bytes); } - printf("----\n"); - - 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) { - break; - } - - if (frame.frame_type != AMQP_FRAME_BODY) { - fprintf(stderr, "Expected body!"); - abort(); - } - - body_received += frame.payload.body_fragment.len; - assert(body_received <= body_target); - amqp_dump(frame.payload.body_fragment.bytes, - frame.payload.body_fragment.len); - } - - if (body_received != body_target) { - /* Can only happen when amqp_simple_wait_frame returns <= 0 */ - /* We break here to close the connection */ - break; - } + amqp_destroy_envelope(&envelope); } } diff --git a/examples/amqp_listenq.c b/examples/amqp_listenq.c index e76cdb1..b2e8094 100644 --- a/examples/amqp_listenq.c +++ b/examples/amqp_listenq.c @@ -85,82 +85,30 @@ int main(int argc, char const *const *argv) die_on_amqp_error(amqp_get_rpc_reply(conn), "Consuming"); { - amqp_frame_t frame; - int result; - - amqp_basic_deliver_t *d; - amqp_basic_properties_t *p; - size_t body_target; - size_t body_received; - while (1) { + amqp_rpc_reply_t res; + amqp_envelope_t envelope; + amqp_maybe_release_buffers(conn); - result = amqp_simple_wait_frame(conn, &frame); - printf("Result %d\n", result); - if (result < 0) { - break; - } - printf("Frame type %d, channel %d\n", frame.frame_type, frame.channel); - if (frame.frame_type != AMQP_FRAME_METHOD) { - continue; - } + res = amqp_consume_message(conn, &envelope, NULL, 0); - printf("Method %s\n", amqp_method_name(frame.payload.method.id)); - if (frame.payload.method.id != AMQP_BASIC_DELIVER_METHOD) { - continue; + if (AMQP_RESPONSE_NORMAL != res.reply_type) { + break; } - d = (amqp_basic_deliver_t *) frame.payload.method.decoded; printf("Delivery %u, exchange %.*s routingkey %.*s\n", - (unsigned) d->delivery_tag, - (int) d->exchange.len, (char *) d->exchange.bytes, - (int) d->routing_key.len, (char *) d->routing_key.bytes); - - result = amqp_simple_wait_frame(conn, &frame); - if (result < 0) { - break; - } + (unsigned) envelope.delivery_tag, + (int) envelope.exchange.len, (char *) envelope.exchange.bytes, + (int) envelope.routing_key.len, (char *) envelope.routing_key.bytes); - if (frame.frame_type != AMQP_FRAME_HEADER) { - fprintf(stderr, "Expected header!"); - abort(); - } - p = (amqp_basic_properties_t *) frame.payload.properties.decoded; - if (p->_flags & AMQP_BASIC_CONTENT_TYPE_FLAG) { + if (envelope.message.properties._flags & AMQP_BASIC_CONTENT_TYPE_FLAG) { printf("Content-type: %.*s\n", - (int) p->content_type.len, (char *) p->content_type.bytes); - } - printf("----\n"); - - 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) { - break; - } - - if (frame.frame_type != AMQP_FRAME_BODY) { - fprintf(stderr, "Expected body!"); - abort(); - } - - body_received += frame.payload.body_fragment.len; - assert(body_received <= body_target); - - amqp_dump(frame.payload.body_fragment.bytes, - frame.payload.body_fragment.len); - } - - if (body_received != body_target) { - /* Can only happen when amqp_simple_wait_frame returns <= 0 */ - /* We break here to close the connection */ - break; + (int) envelope.message.properties.content_type.len, + (char *) envelope.message.properties.content_type.bytes); } - amqp_basic_ack(conn, 1, d->delivery_tag, 0); + amqp_destroy_envelope(&envelope); } } diff --git a/examples/amqps_consumer.c b/examples/amqps_consumer.c index fff6677..d4cd294 100644 --- a/examples/amqps_consumer.c +++ b/examples/amqps_consumer.c @@ -60,13 +60,13 @@ static void run(amqp_connection_state_t conn) 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) { + amqp_rpc_reply_t ret; + amqp_envelope_t envelope; + now = now_microseconds(); if (now > next_summary_time) { int countOverInterval = received - previous_received; @@ -80,45 +80,64 @@ static void run(amqp_connection_state_t conn) } 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(); + ret = amqp_consume_message(conn, &envelope, NULL, 0); + + if (AMQP_RESPONSE_NORMAL != ret.reply_type) { + if (AMQP_RESPONSE_LIBRARY_EXCEPTION == ret.reply_type && + AMQP_STATUS_UNEXPECTED_STATE == ret.library_error) { + if (AMQP_STATUS_OK != amqp_simple_wait_frame(conn, &frame)) { + return; + } + + if (AMQP_FRAME_METHOD == frame.frame_type) { + switch (frame.payload.method.id) { + case AMQP_BASIC_ACK_METHOD: + /* if we've turned publisher confirms on, and we've published a message + * here is a message being confirmed + */ + + break; + case AMQP_BASIC_RETURN_METHOD: + /* if a published message couldn't be routed and the mandatory flag was set + * this is what would be returned. The message then needs to be read. + */ + { + amqp_message_t message; + ret = amqp_read_message(conn, frame.channel, &message, 0); + if (AMQP_RESPONSE_NORMAL != ret.reply_type) { + return; + } + + amqp_destroy_message(&message); + } + + break; + + case AMQP_CHANNEL_CLOSE_METHOD: + /* a channel.close method happens when a channel exception occurs, this + * can happen by publishing to an exchange that doesn't exist for example + * + * In this case you would need to open another channel redeclare any queues + * that were declared auto-delete, and restart any consumers that were attached + * to the previous channel + */ + return; + + case AMQP_CONNECTION_CLOSE_METHOD: + /* a connection.close method happens when a connection exception occurs, + * this can happen by trying to use a channel that isn't open for example. + * + * In this case the whole connection must be restarted. + */ + return; + + default: + fprintf(stderr ,"An unexpected method was received %d\n", frame.payload.method.id); + return; + } + } } - body_received += frame.payload.body_fragment.len; - assert(body_received <= body_target); } received++; diff --git a/examples/amqps_listen.c b/examples/amqps_listen.c index a5eb692..44bb88c 100644 --- a/examples/amqps_listen.c +++ b/examples/amqps_listen.c @@ -121,80 +121,30 @@ int main(int argc, char const *const *argv) die_on_amqp_error(amqp_get_rpc_reply(conn), "Consuming"); { - amqp_frame_t frame; - int result; - - amqp_basic_deliver_t *d; - amqp_basic_properties_t *p; - size_t body_target; - size_t body_received; - while (1) { + amqp_rpc_reply_t res; + amqp_envelope_t envelope; + amqp_maybe_release_buffers(conn); - result = amqp_simple_wait_frame(conn, &frame); - printf("Result %d\n", result); - if (result < 0) { - break; - } - printf("Frame type %d, channel %d\n", frame.frame_type, frame.channel); - if (frame.frame_type != AMQP_FRAME_METHOD) { - continue; - } + res = amqp_consume_message(conn, &envelope, NULL, 0); - printf("Method %s\n", amqp_method_name(frame.payload.method.id)); - if (frame.payload.method.id != AMQP_BASIC_DELIVER_METHOD) { - continue; + if (AMQP_RESPONSE_NORMAL != res.reply_type) { + break; } - d = (amqp_basic_deliver_t *) frame.payload.method.decoded; printf("Delivery %u, exchange %.*s routingkey %.*s\n", - (unsigned) d->delivery_tag, - (int) d->exchange.len, (char *) d->exchange.bytes, - (int) d->routing_key.len, (char *) d->routing_key.bytes); - - result = amqp_simple_wait_frame(conn, &frame); - if (result < 0) { - break; - } + (unsigned) envelope.delivery_tag, + (int) envelope.exchange.len, (char *) envelope.exchange.bytes, + (int) envelope.routing_key.len, (char *) envelope.routing_key.bytes); - if (frame.frame_type != AMQP_FRAME_HEADER) { - fprintf(stderr, "Expected header!"); - abort(); - } - p = (amqp_basic_properties_t *) frame.payload.properties.decoded; - if (p->_flags & AMQP_BASIC_CONTENT_TYPE_FLAG) { + if (envelope.message.properties._flags & AMQP_BASIC_CONTENT_TYPE_FLAG) { printf("Content-type: %.*s\n", - (int) p->content_type.len, (char *) p->content_type.bytes); + (int) envelope.message.properties.content_type.len, + (char *) envelope.message.properties.content_type.bytes); } - printf("----\n"); - - 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) { - break; - } - - if (frame.frame_type != AMQP_FRAME_BODY) { - fprintf(stderr, "Expected body!"); - abort(); - } - - body_received += frame.payload.body_fragment.len; - assert(body_received <= body_target); - amqp_dump(frame.payload.body_fragment.bytes, - frame.payload.body_fragment.len); - } - - if (body_received != body_target) { - /* Can only happen when amqp_simple_wait_frame returns <= 0 */ - /* We break here to close the connection */ - break; - } + amqp_destroy_envelope(&envelope); } } diff --git a/examples/amqps_listenq.c b/examples/amqps_listenq.c index 0210d88..6643500 100644 --- a/examples/amqps_listenq.c +++ b/examples/amqps_listenq.c @@ -102,82 +102,30 @@ int main(int argc, char const *const *argv) die_on_amqp_error(amqp_get_rpc_reply(conn), "Consuming"); { - amqp_frame_t frame; - int result; - - amqp_basic_deliver_t *d; - amqp_basic_properties_t *p; - size_t body_target; - size_t body_received; - while (1) { + amqp_rpc_reply_t res; + amqp_envelope_t envelope; + amqp_maybe_release_buffers(conn); - result = amqp_simple_wait_frame(conn, &frame); - printf("Result %d\n", result); - if (result < 0) { - break; - } - printf("Frame type %d, channel %d\n", frame.frame_type, frame.channel); - if (frame.frame_type != AMQP_FRAME_METHOD) { - continue; - } + res = amqp_consume_message(conn, &envelope, NULL, 0); - printf("Method %s\n", amqp_method_name(frame.payload.method.id)); - if (frame.payload.method.id != AMQP_BASIC_DELIVER_METHOD) { - continue; + if (AMQP_RESPONSE_NORMAL != res.reply_type) { + break; } - d = (amqp_basic_deliver_t *) frame.payload.method.decoded; printf("Delivery %u, exchange %.*s routingkey %.*s\n", - (unsigned) d->delivery_tag, - (int) d->exchange.len, (char *) d->exchange.bytes, - (int) d->routing_key.len, (char *) d->routing_key.bytes); - - result = amqp_simple_wait_frame(conn, &frame); - if (result < 0) { - break; - } + (unsigned) envelope.delivery_tag, + (int) envelope.exchange.len, (char *) envelope.exchange.bytes, + (int) envelope.routing_key.len, (char *) envelope.routing_key.bytes); - if (frame.frame_type != AMQP_FRAME_HEADER) { - fprintf(stderr, "Expected header!"); - abort(); - } - p = (amqp_basic_properties_t *) frame.payload.properties.decoded; - if (p->_flags & AMQP_BASIC_CONTENT_TYPE_FLAG) { + if (envelope.message.properties._flags & AMQP_BASIC_CONTENT_TYPE_FLAG) { printf("Content-type: %.*s\n", - (int) p->content_type.len, (char *) p->content_type.bytes); - } - printf("----\n"); - - 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) { - break; - } - - if (frame.frame_type != AMQP_FRAME_BODY) { - fprintf(stderr, "Expected body!"); - abort(); - } - - body_received += frame.payload.body_fragment.len; - assert(body_received <= body_target); - - amqp_dump(frame.payload.body_fragment.bytes, - frame.payload.body_fragment.len); - } - - if (body_received != body_target) { - /* Can only happen when amqp_simple_wait_frame returns <= 0 */ - /* We break here to close the connection */ - break; + (int) envelope.message.properties.content_type.len, + (char *) envelope.message.properties.content_type.bytes); } - amqp_basic_ack(conn, 1, d->delivery_tag, 0); + amqp_destroy_envelope(&envelope); } } -- cgit v1.2.1 From f8c6cee7499e970afc1ff9d05919413f884696a5 Mon Sep 17 00:00:00 2001 From: Alan Antonuk Date: Tue, 20 Aug 2013 10:51:51 -0700 Subject: FIX: destroy amqp_envelope_t in consumer example Make sure to destroy the envelope object in the amqp_consumer.c example --- examples/amqp_consumer.c | 2 ++ examples/amqps_consumer.c | 3 +++ 2 files changed, 5 insertions(+) (limited to 'examples') diff --git a/examples/amqp_consumer.c b/examples/amqp_consumer.c index 62da0ca..11e186e 100644 --- a/examples/amqp_consumer.c +++ b/examples/amqp_consumer.c @@ -136,6 +136,8 @@ static void run(amqp_connection_state_t conn) } } + } else { + amqp_destroy_envelope(&envelope); } received++; diff --git a/examples/amqps_consumer.c b/examples/amqps_consumer.c index d4cd294..affe0f6 100644 --- a/examples/amqps_consumer.c +++ b/examples/amqps_consumer.c @@ -138,8 +138,11 @@ static void run(amqp_connection_state_t conn) } } + } else { + amqp_destroy_envelope(&envelope); } + received++; } } -- cgit v1.2.1