examples/engine/c/psend.c - qpid-proton-j - Git at Google

 /*
  *
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
  * distributed with this work for additional information
  * regarding copyright ownership.  The ASF licenses this file
  * to you under the Apache License, Version 2.0 (the
  * "License"); you may not use this file except in compliance
  * with the License.  You may obtain a copy of the License at
  *
  *   http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing,
  * software distributed under the License is distributed on an
  * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
  * KIND, either express or implied.  See the License for the
  * specific language governing permissions and limitations
  * under the License.
  *
  */


 /*################################################################
   This program is half of a pair.  Precv and Psend are meant to
   be simple-as-possible examples of how to use the proton-c
   engine interface to send and receive messages over a single
   connection and a single session.

   In addition to being examples, these programs or their
   descendants will be used in performance regression testing
   for both throughput and latency, and long-term soak testing.

   This program, psend, is highly similar to its peer precv.
   I put all the good comments in precv.
 *################################################################*/


 #include <unistd.h>
 #include <stdlib.h>
 #include <stdio.h>
 #include <string.h>
 #include <ctype.h>
 #include <time.h>
 #include <sys/time.h>
 #define __STDC_FORMAT_MACROS
 #include <inttypes.h>


 #include <proton/connection.h>
 #include <proton/delivery.h>
 #include <proton/driver.h>
 #include <proton/event.h>
 #include <proton/terminus.h>
 #include <proton/link.h>
 #include <proton/message.h>
 #include <proton/session.h>


 #define MY_BUF_SIZE 1000


 void
 print_timestamp ( FILE * fp, char const * label )
 {
   struct timeval tv;
   struct tm      * timeinfo;

   gettimeofday ( & tv, 0 );
   timeinfo = localtime ( & tv.tv_sec );

   int seconds_today = 3600 * timeinfo->tm_hour +
                         60 * timeinfo->tm_min  +
                              timeinfo->tm_sec;

   fprintf ( fp, "time : %d.%.6ld : %s\n", seconds_today, tv.tv_usec, label );
 }


 static
 double
 get_time ( )
 {
   struct timeval tv;
   struct tm      * timeinfo;

   gettimeofday ( & tv, 0 );
   timeinfo = localtime ( & tv.tv_sec );

   double time_now = 3600 * timeinfo->tm_hour +
                       60 * timeinfo->tm_min  +
                            timeinfo->tm_sec;

   time_now += ((double)(tv.tv_usec) / 1000000.0);
   return time_now;
 }


 int
 main ( int argc, char ** argv )
 {
   char addr [ 1000 ];
   char host [ 1000 ];
   char port [ 1000 ];
   char output_file_name[1000];


   uint64_t messages       = 2000000,
            delivery_count = 0;

   int message_length = 100;

   bool done           = false;
   int  sent_count     = 0;
   int  n_links        = 5;
   int const max_links = 100;

   strcpy ( addr, "queue" );
   strcpy ( host, "0.0.0.0" );
   strcpy ( port, "5672" );

   FILE * output_fp;


   for ( int i = 1; i < argc; ++ i )
   {
     if ( ! strcmp ( "--host", argv[i] ) )
     {
       strcpy ( host, argv[i+1] );
       ++ i;
     }
     else
     if ( ! strcmp ( "--port", argv[i] ) )
     {
       strcpy ( port, argv[i+1] );
       ++ i;
     }
     else
     if ( ! strcmp ( "--messages", argv[i] ) )
     {
       sscanf ( argv [ i+1 ], "%" SCNd64 , & messages );
       ++ i;
     }
     else
     if ( ! strcmp ( "--message_length", argv[i] ) )
     {
       sscanf ( argv [ i+1 ], "%d", & message_length );
       ++ i;
     }
     else
     if ( ! strcmp ( "--n_links", argv[i] ) )
     {
       sscanf ( argv [ i+1 ], "%d", & n_links );
       ++ i;
     }
     if ( ! strcmp ( "--output", argv[i] ) )
     {
       if ( ! strcmp ( "stderr", argv[i+1] ) )
       {
         output_fp = stderr;
         strcpy ( output_file_name, "stderr");
       }
       else
       if ( ! strcmp ( "stdout", argv[i+1] ) )
       {
         output_fp = stdout;
         strcpy ( output_file_name, "stdout");
       }
       else
       {
         output_fp = fopen ( argv[i+1], "w" );
         strcpy ( output_file_name, argv[i+1] );
         if ( ! output_fp )
         {
           fprintf ( stderr, "Can't open |%s| for writing.\n", argv[i+1] );
           exit ( 1 );
         }
       }
       ++ i;
     }
     else
     {
       fprintf ( output_fp, "unknown arg %s", argv[i] );
     }
   }


   fprintf ( output_fp, "host            %s\n",          host );
   fprintf ( output_fp, "port            %s\n",          port );
   fprintf ( output_fp, "messages        %" PRId64 "\n", messages );
   fprintf ( output_fp, "message_length  %d\n",          message_length );
   fprintf ( output_fp, "n_links         %d\n",          n_links );
   fprintf ( output_fp, "output          %s\n",          output_file_name );


   if ( n_links > max_links )
   {
     fprintf ( output_fp, "You can't have more than %d links.\n", max_links );
     exit ( 1 );
   }

   pn_driver_t     * driver;
   pn_connector_t  * connector;
   pn_connector_t  * driver_connector;
   pn_connection_t * connection;
   pn_collector_t  * collector;
   pn_link_t       * links [ max_links ];
   pn_session_t    * session;
   pn_event_t      * event;
   pn_delivery_t   * delivery;


   char * message = (char *) malloc(message_length);
   memset ( message, 13, message_length );

   /*----------------------------------------------------
     Get everything set up.
     We will have a single connector, a single
     connection, a single session, and a single link.
   ----------------------------------------------------*/
   driver = pn_driver ( );
   connector = pn_connector ( driver, host, port, 0 );

   connection = pn_connection();
   collector  = pn_collector  ( );
   pn_connection_collect ( connection, collector );
   pn_connector_set_connection ( connector, connection );

   session = pn_session ( connection );
   pn_connection_open ( connection );
   pn_session_open ( session );

   for ( int i = 0; i < n_links; ++ i )
   {
     char name[100];
     sprintf ( name, "tvc_15_%d", i );
     links[i] = pn_sender ( session, name );
     pn_terminus_set_address ( pn_link_target(links[i]), addr );
     pn_link_open ( links[i] );
   }

   /*-----------------------------------------------------------
     For my speed tests, I do not want to count setup time.
     Start timing here.  The receiver will print out a similar
     timestamp when he receives the final message.
   -----------------------------------------------------------*/
   fprintf ( output_fp, "psend: sending %llu messages.\n", messages );

   // Just before we start sending, print the start timestamp.
   fprintf ( output_fp, "psend_start %.3lf\n", get_time() );

   while ( 1 )
   {
     pn_driver_wait ( driver, -1 );

     int event_count = 1;
     while ( event_count > 0 )
     {
       event_count = 0;
       pn_connector_process ( connector );

       event = pn_collector_peek(collector);
       while ( event )
       {
         ++ event_count;
         pn_event_type_t event_type = pn_event_type ( event );
         //fprintf ( output_fp, "event: %s\n", pn_event_type_name ( event_type ) );

         switch ( event_type )
         {
           case PN_LINK_FLOW:
           {
             if ( delivery_count < messages )
             {
               /*---------------------------------------------------
                 We may have opened multiple links.
                 The event will tell us which one this flow-event
                 happened on.  If the flow event gave us some
                 credit, we will greedily send messages until it
                 is all used up.
               ---------------------------------------------------*/
               pn_link_t * link = pn_event_link ( event );
               int credit = pn_link_credit ( link );

               while ( credit > 0 )
               {
                 // Every delivery we create needs a unique tag.
                 char str [ 100 ];
                 sprintf ( str, "%x", delivery_count ++ );
                 delivery = pn_delivery ( link, pn_dtag(str, strlen(str)) );

                 // If you settle the delivery before sending it,
                 // you will spend some time wondering why your
                 // messages don't have any content when they arrive
                 // at the receiver.
                 pn_link_send ( link, message, message_length );
                 pn_delivery_settle ( delivery );
                 pn_link_advance ( link );
                 credit = pn_link_credit ( link );
               }

               if ( delivery_count >= messages )
               {
                 fprintf ( output_fp,
                           "I have sent all %d messages.\n" ,
                           delivery_count
                         );
                 /*
                   I'm still kind of hazy on how to shut down the
                   psend / precv system properly ....
                   I can't go to all_done here, or precv will never
                   get all its messages and terminate.
                   So I let precv terminate properly ... which means
                   that this program, psend, dies with an error.
                   Hmm.
                 */
                 // goto all_done;
               }
             }
           }
           break;


           case PN_TRANSPORT:
             // I don't know what this means here, either.
           break;


           case PN_TRANSPORT_TAIL_CLOSED:
             goto all_done;
           break;


           default:
             /*
             fprintf ( output_fp,
                       "precv unhandled event: %s\n",
                       pn_event_type_name(event_type)
                     );
             */
           break;

         }

         pn_collector_pop ( collector );
         event = pn_collector_peek(collector);
       }
     }
   }

   all_done:

   for ( int i = 0; i < n_links; ++ i )
   {
     pn_link_close ( links[i] );
   }

   pn_session_close ( session );
   pn_connection_close ( connection );

   return 0;
 }
	/*
	*
	* Licensed to the Apache Software Foundation (ASF) under one
	* or more contributor license agreements. See the NOTICE file
	* distributed with this work for additional information
	* regarding copyright ownership. The ASF licenses this file
	* to you under the Apache License, Version 2.0 (the
	* "License"); you may not use this file except in compliance
	* with the License. You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing,
	* software distributed under the License is distributed on an
	* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
	* KIND, either express or implied. See the License for the
	* specific language governing permissions and limitations
	* under the License.
	*
	*/


	/*################################################################
	This program is half of a pair. Precv and Psend are meant to
	be simple-as-possible examples of how to use the proton-c
	engine interface to send and receive messages over a single
	connection and a single session.

	In addition to being examples, these programs or their
	descendants will be used in performance regression testing
	for both throughput and latency, and long-term soak testing.

	This program, psend, is highly similar to its peer precv.
	I put all the good comments in precv.
	################################################################/


	#include <unistd.h>
	#include <stdlib.h>
	#include <stdio.h>
	#include <string.h>
	#include <ctype.h>
	#include <time.h>
	#include <sys/time.h>
	#define __STDC_FORMAT_MACROS
	#include <inttypes.h>


	#include <proton/connection.h>
	#include <proton/delivery.h>
	#include <proton/driver.h>
	#include <proton/event.h>
	#include <proton/terminus.h>
	#include <proton/link.h>
	#include <proton/message.h>
	#include <proton/session.h>


	#define MY_BUF_SIZE 1000



	void
	print_timestamp ( FILE * fp, char const * label )
	{
	struct timeval tv;
	struct tm * timeinfo;

	gettimeofday ( & tv, 0 );
	timeinfo = localtime ( & tv.tv_sec );

	int seconds_today = 3600 * timeinfo->tm_hour +
	60 * timeinfo->tm_min +
	timeinfo->tm_sec;

	fprintf ( fp, "time : %d.%.6ld : %s\n", seconds_today, tv.tv_usec, label );
	}





	static
	double
	get_time ( )
	{
	struct timeval tv;
	struct tm * timeinfo;

	gettimeofday ( & tv, 0 );
	timeinfo = localtime ( & tv.tv_sec );

	double time_now = 3600 * timeinfo->tm_hour +
	60 * timeinfo->tm_min +
	timeinfo->tm_sec;

	time_now += ((double)(tv.tv_usec) / 1000000.0);
	return time_now;
	}





	int
	main ( int argc, char ** argv )
	{
	char addr [ 1000 ];
	char host [ 1000 ];
	char port [ 1000 ];
	char output_file_name[1000];


	uint64_t messages = 2000000,
	delivery_count = 0;

	int message_length = 100;

	bool done = false;
	int sent_count = 0;
	int n_links = 5;
	int const max_links = 100;

	strcpy ( addr, "queue" );
	strcpy ( host, "0.0.0.0" );
	strcpy ( port, "5672" );

	FILE * output_fp;


	for ( int i = 1; i < argc; ++ i )
	{
	if ( ! strcmp ( "--host", argv[i] ) )
	{
	strcpy ( host, argv[i+1] );
	++ i;
	}
	else
	if ( ! strcmp ( "--port", argv[i] ) )
	{
	strcpy ( port, argv[i+1] );
	++ i;
	}
	else
	if ( ! strcmp ( "--messages", argv[i] ) )
	{
	sscanf ( argv [ i+1 ], "%" SCNd64 , & messages );
	++ i;
	}
	else
	if ( ! strcmp ( "--message_length", argv[i] ) )
	{
	sscanf ( argv [ i+1 ], "%d", & message_length );
	++ i;
	}
	else
	if ( ! strcmp ( "--n_links", argv[i] ) )
	{
	sscanf ( argv [ i+1 ], "%d", & n_links );
	++ i;
	}
	if ( ! strcmp ( "--output", argv[i] ) )
	{
	if ( ! strcmp ( "stderr", argv[i+1] ) )
	{
	output_fp = stderr;
	strcpy ( output_file_name, "stderr");
	}
	else
	if ( ! strcmp ( "stdout", argv[i+1] ) )
	{
	output_fp = stdout;
	strcpy ( output_file_name, "stdout");
	}
	else
	{
	output_fp = fopen ( argv[i+1], "w" );
	strcpy ( output_file_name, argv[i+1] );
	if ( ! output_fp )
	{
	fprintf ( stderr, "Can't open \|%s\| for writing.\n", argv[i+1] );
	exit ( 1 );
	}
	}
	++ i;
	}
	else
	{
	fprintf ( output_fp, "unknown arg %s", argv[i] );
	}
	}


	fprintf ( output_fp, "host %s\n", host );
	fprintf ( output_fp, "port %s\n", port );
	fprintf ( output_fp, "messages %" PRId64 "\n", messages );
	fprintf ( output_fp, "message_length %d\n", message_length );
	fprintf ( output_fp, "n_links %d\n", n_links );
	fprintf ( output_fp, "output %s\n", output_file_name );


	if ( n_links > max_links )
	{
	fprintf ( output_fp, "You can't have more than %d links.\n", max_links );
	exit ( 1 );
	}

	pn_driver_t * driver;
	pn_connector_t * connector;
	pn_connector_t * driver_connector;
	pn_connection_t * connection;
	pn_collector_t * collector;
	pn_link_t * links [ max_links ];
	pn_session_t * session;
	pn_event_t * event;
	pn_delivery_t * delivery;


	char * message = (char *) malloc(message_length);
	memset ( message, 13, message_length );

	/*----------------------------------------------------
	Get everything set up.
	We will have a single connector, a single
	connection, a single session, and a single link.
	----------------------------------------------------*/
	driver = pn_driver ( );
	connector = pn_connector ( driver, host, port, 0 );

	connection = pn_connection();
	collector = pn_collector ( );
	pn_connection_collect ( connection, collector );
	pn_connector_set_connection ( connector, connection );

	session = pn_session ( connection );
	pn_connection_open ( connection );
	pn_session_open ( session );

	for ( int i = 0; i < n_links; ++ i )
	{
	char name[100];
	sprintf ( name, "tvc_15_%d", i );
	links[i] = pn_sender ( session, name );
	pn_terminus_set_address ( pn_link_target(links[i]), addr );
	pn_link_open ( links[i] );
	}

	/*-----------------------------------------------------------
	For my speed tests, I do not want to count setup time.
	Start timing here. The receiver will print out a similar
	timestamp when he receives the final message.
	-----------------------------------------------------------*/
	fprintf ( output_fp, "psend: sending %llu messages.\n", messages );

	// Just before we start sending, print the start timestamp.
	fprintf ( output_fp, "psend_start %.3lf\n", get_time() );

	while ( 1 )
	{
	pn_driver_wait ( driver, -1 );

	int event_count = 1;
	while ( event_count > 0 )
	{
	event_count = 0;
	pn_connector_process ( connector );

	event = pn_collector_peek(collector);
	while ( event )
	{
	++ event_count;
	pn_event_type_t event_type = pn_event_type ( event );
	//fprintf ( output_fp, "event: %s\n", pn_event_type_name ( event_type ) );

	switch ( event_type )
	{
	case PN_LINK_FLOW:
	{
	if ( delivery_count < messages )
	{
	/*---------------------------------------------------
	We may have opened multiple links.
	The event will tell us which one this flow-event
	happened on. If the flow event gave us some
	credit, we will greedily send messages until it
	is all used up.
	---------------------------------------------------*/
	pn_link_t * link = pn_event_link ( event );
	int credit = pn_link_credit ( link );

	while ( credit > 0 )
	{
	// Every delivery we create needs a unique tag.
	char str [ 100 ];
	sprintf ( str, "%x", delivery_count ++ );
	delivery = pn_delivery ( link, pn_dtag(str, strlen(str)) );

	// If you settle the delivery before sending it,
	// you will spend some time wondering why your
	// messages don't have any content when they arrive
	// at the receiver.
	pn_link_send ( link, message, message_length );
	pn_delivery_settle ( delivery );
	pn_link_advance ( link );
	credit = pn_link_credit ( link );
	}

	if ( delivery_count >= messages )
	{
	fprintf ( output_fp,
	"I have sent all %d messages.\n" ,
	delivery_count
	);
	/*
	I'm still kind of hazy on how to shut down the
	psend / precv system properly ....
	I can't go to all_done here, or precv will never
	get all its messages and terminate.
	So I let precv terminate properly ... which means
	that this program, psend, dies with an error.
	Hmm.
	*/
	// goto all_done;
	}
	}
	}
	break;


	case PN_TRANSPORT:
	// I don't know what this means here, either.
	break;


	case PN_TRANSPORT_TAIL_CLOSED:
	goto all_done;
	break;


	default:
	/*
	fprintf ( output_fp,
	"precv unhandled event: %s\n",
	pn_event_type_name(event_type)
	);
	*/
	break;

	}

	pn_collector_pop ( collector );
	event = pn_collector_peek(collector);
	}
	}
	}

	all_done:

	for ( int i = 0; i < n_links; ++ i )
	{
	pn_link_close ( links[i] );
	}

	pn_session_close ( session );
	pn_connection_close ( connection );

	return 0;
	}