/*
 *
 * 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.
*################################################################*/

#include <stdio.h>
#include <string.h>
#include <ctype.h>
#include <sys/time.h>
#include <unistd.h>
#include <stdlib.h>
#include <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


/*---------------------------------------------------------
  These high-resolution times are used both for
  interim timing reports -- i.e. every 'report_frequency'
  messages -- and for the final timestamp, after all 
  expected messages have been received.
---------------------------------------------------------*/
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;
}


/*----------------------------------------------------
  These absolute timestamps are useful in soak tests,
  where I want to align the program's output with 
  output from top to look at CPU and memory use..
----------------------------------------------------*/
void
print_timestamp_like_a_normal_person ( FILE * fp )
{
  char const * month_abbrevs[] = { "jan", 
                                   "feb", 
                                   "mar", 
                                   "apr", 
                                   "may", 
                                   "jun", 
                                   "jul", 
                                   "aug", 
                                   "sep", 
                                   "oct", 
                                   "nov", 
                                   "dec" 
                                 };
  time_t rawtime;
  struct tm * timeinfo;

  time ( & rawtime );
  timeinfo = localtime ( &rawtime );

  char time_string[100];
  sprintf ( time_string,
            "%d-%s-%02d %02d:%02d:%02d",
            1900 + timeinfo->tm_year,
            month_abbrevs[timeinfo->tm_mon],
            timeinfo->tm_mday,
            timeinfo->tm_hour,
            timeinfo->tm_min,
            timeinfo->tm_sec
          );

  fprintf ( fp, "timestamp %s\n", time_string );
}


int 
main ( int argc, char ** argv  ) 
{
  char info[1000];

  uint64_t received = 0;

  char host [1000];
  char port [1000];
  char output_file_name[1000];

  int initial_flow   = 400;
  int flow_increment = 200;

  int       report_frequency  = 200000;
  int64_t   messages          = 2000000,
            delivery_count    = 0;

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

  pn_driver_t     * driver;
  pn_listener_t   * listener;
  pn_connector_t  * connector;
  pn_connection_t * connection;
  pn_collector_t  * collector;
  pn_transport_t  * transport;
  pn_sasl_t       * sasl;
  pn_session_t    * session;
  pn_event_t      * event;
  pn_link_t       * link;
  pn_delivery_t   * delivery;


  double last_time,
         this_time,
         time_diff;

  char * message_data          = (char *) malloc ( MY_BUF_SIZE );
  int    message_data_capacity = MY_BUF_SIZE;

  FILE * output_fp;


  /*-----------------------------------------------------------
    Read the command lines args and override initialization.
  -----------------------------------------------------------*/
  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 ( "--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
    if ( ! strcmp ( "--report_frequency", argv[i] ) )
    {
      report_frequency = atoi ( argv[i+1] );
      ++ i;
    }
    else
    if ( ! strcmp ( "--initial_flow", argv[i] ) )
    {
      sscanf ( argv [ i+1 ], "%d", & initial_flow );
      ++ i;
    }
    else
    if ( ! strcmp ( "--flow_increment", argv[i] ) )
    {
      sscanf ( argv [ i+1 ], "%d", & flow_increment );
      ++ i;
    }
    else
    {
      fprintf ( output_fp, "unknown arg |%s|\n", argv[i] );
      exit ( 1 );
    }
  }

  /*-----------------------------------------------
    Show what we ended up with for all the args.
  -----------------------------------------------*/
  fprintf ( output_fp, "host                %s\n",          host );
  fprintf ( output_fp, "port                %s\n",          port );
  fprintf ( output_fp, "messages            %" PRId64 "\n", messages );
  fprintf ( output_fp, "report_frequency    %d\n",          report_frequency );
  fprintf ( output_fp, "initial_flow        %d\n",          initial_flow );
  fprintf ( output_fp, "flow_increment      %d\n",          flow_increment );
  fprintf ( output_fp, "output              %s\n",          output_file_name );


  /*--------------------------------------------
    Create a standard driver and listen for the 
    initial connector.
  --------------------------------------------*/
  driver = pn_driver ( );

  if ( ! pn_listener(driver, host, port, 0) ) 
  {
    fprintf ( output_fp, "precv listener creation failed.\n" );
    exit ( 1 );
  }

  fprintf ( output_fp, "\nprecv ready...\n\n" );

  while ( 1 )
  {
    pn_driver_wait ( driver, -1 );
    if ( listener = pn_driver_listener(driver) )
    {
      if ( connector = pn_listener_accept(listener) )
        break;
    }
  }

  /*--------------------------------------------------------
    Now make all the other structure around the connector,
    and tell it that skipping sasl is OK.
  --------------------------------------------------------*/
  connection = pn_connection ( );
  collector  = pn_collector  ( );
  pn_connection_collect ( connection, collector );
  pn_connector_set_connection ( connector, connection );

  transport = pn_connector_transport ( connector );
  sasl = pn_sasl ( transport );
  pn_sasl_mechanisms ( sasl, "ANONYMOUS" );
  pn_sasl_server ( sasl );
  pn_sasl_allow_skip ( sasl, true );
  pn_sasl_done ( sasl, PN_SASL_OK );

  /*----------------------------------------------------------
    If report_frequency is not set to zero, we will 
    produce a timing report every report_frequency messages.
    The timing reported will be the delta from the last_time
    to the current time.  
    This is useful in soak testing, where you basically 
    never stop, but still need to see how the system is doing
    every so often.
  ----------------------------------------------------------*/
  last_time = get_time();


  /*------------------------------------------------------------
    A triply-nested loop.
    In the outermost one, we just wait for activity to come in 
    from the driver.
  ------------------------------------------------------------*/
  while ( 1 )
  {
    pn_driver_wait ( driver, -1 );

    /*---------------------------------------------------------------
      In the next loop, we keep going as long as we processed
      some events.  This is because our own processing of events
      may have caused more to be generated that we need to handle.
      If we go back to the outermost loop and its pn_driver_wait()
      without handling these events, we will end up with the sender
      and receiver programs just staring at each other with blank
      expressions on their faces.
    ---------------------------------------------------------------*/
    int event_count = 1;
    while ( event_count > 0 )
    {
      event_count = 0;
      pn_connector_process ( connector );

      /*-------------------------------------------------------
        After we process the connector, it may have generated
        a batch of events for us to handle.  As we go through 
        this batch of events, our handling may generate other 
        events which we must handle before going back to 
        pn_driver_wait().
      --------------------------------------------------------*/
      event = pn_collector_peek(collector);
      while ( event )
      {
        ++ event_count;
        pn_event_type_t event_type = pn_event_type ( event );
        //fprintf ( output_fp, "precv event: %s\n", pn_event_type_name(event_type));


        switch ( event_type )
        {
          case PN_CONNECTION_REMOTE_OPEN:
          break;


          case PN_SESSION_REMOTE_OPEN:
            session = pn_event_session(event);
            if ( pn_session_state(session) & PN_LOCAL_UNINIT ) 
            {
              // big number because it's in bytes.
              pn_session_set_incoming_capacity ( session, 1000000 );
              pn_session_open ( session );
            }
          break;


          case PN_LINK_REMOTE_OPEN:
            /*----------------------------------------------------
              When we first open the link, we give it an initial 
              amount of credit in units of messages.
              We will later increment its credit whenever credit
              falls below some threshold.
            ----------------------------------------------------*/
            link = pn_event_link(event);
            if (pn_link_state(link) & PN_LOCAL_UNINIT )
            {
              pn_link_open ( link );
              pn_link_flow ( link, initial_flow );
            }
          break;


          case PN_CONNECTION_BOUND:
            if ( pn_connection_state(connection) & PN_LOCAL_UNINIT )
            {
              pn_connection_open ( connection );
            }
          break;


          // And now the event that you've all been waiting for.....

          case PN_DELIVERY:
            link = pn_event_link ( event );
            delivery = pn_event_delivery ( event );

            /*------------------------------------------------
              Since I want this program to be a receiver,
              I am not interested in deliver-related events 
              unless they are incoming, 'readable' events.
            ------------------------------------------------*/
            if ( pn_delivery_readable ( delivery ) )
            {
              // If the delivery is partial I am just going to ignore
              // it until it becomes complete.
              if ( ! pn_delivery_partial ( delivery ) )
              {
                ++ delivery_count; 
                /*
                if ( ! (delivery_count % report_frequency) )
                {
                  pn_link_t * delivery_link = pn_delivery_link ( delivery );
                  int received_bytes = pn_delivery_pending ( delivery );
                  pn_link_recv ( delivery_link, incoming, 1000 );
                  fprintf ( output_fp, "received bytes: %d\n", received_bytes );
                }
                */

                // don't bother updating.  they're pre-settled.
                // pn_delivery_update ( delivery, PN_ACCEPTED );
                pn_delivery_settle ( delivery );

                int credit = pn_link_credit ( link );

                if ( delivery_count >= messages )
                {
                  fprintf ( output_fp, "precv_stop %.3lf\n", get_time() );
                  goto all_done;
                }

                // Make a report frequency of zero shut down interim reporting.
                if ( report_frequency
                     &&
                     (! (delivery_count % report_frequency))
                   )
                {
                  this_time = get_time();
                  time_diff = this_time - last_time;

                  print_timestamp_like_a_normal_person ( output_fp );
                  fprintf ( output_fp, 
                            "deliveries: %" PRIu64 "  time: %.3lf\n", 
                            delivery_count,
                            time_diff
                          );
                  fflush ( output_fp );
                  last_time = this_time;
                }

                /*----------------------------------------------------------
                  I replenish the credit whevere it falls below this limit
                  because I this psend / precv pair of programs to run as
                  fast as possible.  A real application might want to do
                  something fancier here.
                ----------------------------------------------------------*/
                if ( credit <= flow_increment )
                {
                  pn_link_flow ( link, flow_increment );
                }
              }
            }
          break;


          case PN_TRANSPORT:
            // not sure why I would care...
          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:
  pn_session_close ( session );
  pn_connection_close ( connection );
  fclose ( output_fp );
  return 0;
}