// // client.c - Sends arbitrary request and waits for answer // // Copyright (c) 1996-2007 iMatix Corporation // All rights reserved. // // This file is licensed under the BSD license as follows: // // Redistribution and use in source and binary forms, with or without // modification, are permitted provided that the following conditions // are met: // // * Redistributions of source code must retain the above copyright // notice, this list of conditions and the following disclaimer. // * Redistributions in binary form must reproduce the above copyright // notice, this list of conditions and the following disclaimer in // the documentation andor other materials provided with the // distribution. // * Neither the name of iMatix Corporation nor the names of its // contributors may be used to endorse or promote products derived // from this software without specific prior written permission. // // THIS SOFTWARE IS PROVIDED BY IMATIX CORPORATION "AS IS" AND ANY // EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE // IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR // PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL IMATIX CORPORATION BE // LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR // CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF // SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR // BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF // LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING // NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS // SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. // // Name: client // Usage: client // Example: client 127.0.0.1 client1 // Function: Sends arbitrary request and waits for answer #include "base.h" #include "amq_client_connection.h" #include "amq_client_session.h" int main (int argc, char *argv []) { if (argc != 3) { printf ("Usage: client \n"); return 0; } amq_client_connection_t *connection = NULL; amq_client_session_t *session = NULL; icl_longstr_t *auth_data; amq_content_basic_t *content = NULL; char message_text [1024]; size_t message_size; char routing_key [512]; icl_system_initialise (2, argv); auth_data = amq_client_connection_auth_plain ("guest", "guest"); connection = amq_client_connection_new (argv [1], "/", auth_data, "test", 0, 30000); icl_longstr_destroy (&auth_data); assert (connection); // Print server info to stdout printf ("%s: ", argv [1]); printf ("%s/%s - %s - %s\n", connection->server_product, connection->server_version, connection->server_platform, connection->server_information); // Taken as a second argument from command line printf ("My ID: %s\n", argv [2]); session = amq_client_session_new (connection); assert (session); // Make sure that 'services' exchange exists, if it does not, create it amq_client_session_exchange_declare ( session, // session 0, // ticket "services", // exchange name "direct", // exchange type FALSE, // shoudn't the exchange be created? FALSE, // durable FALSE, // auto-delete when unused FALSE, // create internal exchange NULL); // arguments for declaration // Make sure that 'responses' exchange exists, if it does not, create it amq_client_session_exchange_declare ( session, // session 0, // ticket "responses", // exchange name "direct", // exchange type FALSE, // shoudn't the exchange be created? FALSE, // durable FALSE, // auto-delete when unused FALSE, // create internal exchange NULL); // arguments for declaration // Create a response queue with queue_name = "" // (name will be asigned by broker) amq_client_session_queue_declare ( session, // session 0, // ticket "", // queue name FALSE, // passive FALSE, // durable TRUE, // exclusive TRUE, // auto-delete NULL); // arguments // Bind the response queue to the exchange amq_client_session_queue_bind ( session, // session 0, // ticket session->queue, // queue "responses", // exchange session->queue, // routing-key NULL); // arguments // Consume from the response queue amq_client_session_basic_consume ( session, // session 0, // ticket session->queue, // queue NULL, // consumer-tag TRUE, // no-local TRUE, // no-ack TRUE, // exclusive NULL); // arguments while (1) { if (!connection->alive) break; printf ("Enter routing key for \'GET\' message [Q for exit]: "); char *ch_ptr = fgets (routing_key, sizeof (routing_key), stdin); assert (ch_ptr == routing_key); ch_ptr = strchr (routing_key, '\n'); assert (ch_ptr); *ch_ptr = '\0'; if ((strlen (routing_key) == 1) & (routing_key [0] == 'q' || routing_key [0] == 'Q')) break; sprintf (message_text, "GET: It is me %s.", argv [2]); printf ("Send : rk \'%s\', text \'%s\', reply_to \'%s\'\n", routing_key, message_text, session->queue); content = amq_content_basic_new (); amq_content_basic_set_body (content, message_text, strlen (message_text), NULL); amq_content_basic_set_message_id (content, "ID001"); // Set reply_to message property amq_content_basic_set_reply_to (content, session->queue); // Publish the message amq_client_session_basic_publish ( session, // session content, // content to send 0, // ticket "services", // exchange to send message to routing_key, // routing-key TRUE, // mandatory //true TRUE); // immediate //true amq_content_basic_destroy (&content); // Wait for incomming message amq_client_session_wait (session, 0); // Check if it is "returned" message content = amq_client_session_basic_returned (session); if (content) { printf ("No server connected. MSG: %s\n", session->reply_text); continue; } content = amq_client_session_basic_arrived (session); if (!content) break; // Get the message body message_size = amq_content_basic_get_body (content, (byte*) message_text, sizeof (message_text)); if (message_size) { message_text [message_size] = 0; } amq_content_basic_destroy (&content); printf ("Reply: \'%s\'\n", message_text); } // Close the session and connection amq_client_session_destroy (&session); amq_client_connection_destroy (&connection); icl_system_terminate (); return 0; }