Search  

---

C:/temp/src/j2k/nto/Server.cpp

Go to the documentation of this file.

#ifndef __J2K__Server_CPP__
#define __J2K__Server_CPP__

#include <j2k/nto/Server.hpp>

  Server::Server( NameSpace* n0 )
    : Basic_PThread(), 
      MsgContainer( n0, server ), 
      nbProducer( 0 ),
      nbConsumer( 0 )
  { 
  }

  Server::~Server() 
  { 
    printf( "Killing server... \n" );
    register size_t  i   = 0;
    register size_t  max = timeout.size();

    iter = timeout.begin();
    for( i = 0; i < max; i++ )
    {
      if ( timeout[i] != NULL )
      {
        deleteEntry( i );
      }
      iter++;
    }
  }

  void Server::deleteEntry( register int index )
  {
//     printf( "Delete server entry [%d] \n", index );
     fflush( stdout );

     // Delete that entry
     timeout[ index ]->stop();
     timeout[ index ]->reset();

//     Timeout* t = timeout[ index ];
     timeout[ index ] = NULL;
//     timeout.erase( iter );
//     delete t;

//    printf( "New server container size [%d] \n", timeout.size() );
    fflush( stdout );
  }

  void Server::createEntry( register int rcvid, 
                            register Message* message, 
                            register int startTimer = 0 )
  {

//    printf( "Create server entry for [%d] \n", message->getNumber() );
//    fflush( stdout );

    Timeout* t = new Timeout( MsgContainer::n, 0, 1 ); // 5 sec
    t->copy( message );

    if ( startTimer > 0 )
    {
      t->start();
    }

    t->setReceiver( rcvid );
    timeout.push_back( t );

//    printf( "New server container size [%d] \n", timeout.size() );
  }

  /*** Server Side of the code ***/
  void Server::run()
  {
    printf( "Server run \n" );
    fflush( stdout );

    // Create a namespace channel for Producer-Server-Consumer
    connect();

    printf( "Server attached \n" );
    fflush( stdout );

    /* Do your MsgReceive's here now with the chid */
    while( 1 ) 
    {
      printf( "Wait for receive...\n" );
      fflush( stdout );

      int rcvid = message->receive( n->getChannel() ); 

      printf( "Message received from ID#%d \n", rcvid );
      fflush( stdout );

      // Check for local Message
      if ( rcvid == 0 ) continue;
       
      // Check for Kernel Message
      if ( message->checkHeader( rcvid ) < 0 ) break;

      printf("Server receive %d \n", message->getNumber() );
      fflush( stdout );

      message->print();
      message->setStatus( message->getRequest() );
   
      // Handling Consumer Message       
      if ( message->isConsumer() )  
      {
        printf( "Consumer message \n" );
        fflush( stdout );

        // Feed the Consumer
        if ( nbProducer > 0 ) 
        {
     // There is some Producer waiting to feed consumer so FEED IT!
          register size_t  i   = 0;
          register size_t  max = timeout.size();
          iter = timeout.begin();
          for( i = 0; i < max; i++, iter++ )
          {
           if ( timeout[i] != NULL )
           {
            if ( timeout[ i ]->getNumber() != -1  &&  timeout[ i ]->isProducer() )
            {
              printf( "Consumer fed by entry %d, producer %d \n", i, timeout[ i ]->getNumber() );
              fflush( stdout );

              timeout[ i ]->stop(); 
              timeout[ i ]->reply( rcvid );                   // Reply Consumer
              message->reply( timeout[ i ]->getReceiver() );  // Reply Producer

              // Delete that entry
              deleteEntry( i );

              nbProducer--;
              i = max;

              printf( "Done with case 1 ! \n" );
              fflush( stdout );
            }
           }
          }
        }
        else 
        {
     // Store the consumer request with timeout
          printf( "Storing Consumer %d \n", message->getNumber() );
          fflush( stdout );

          createEntry( rcvid, message, 1 );
          nbConsumer++;

          printf( "Done with case 2 ! \n" );
          fflush( stdout );
        }
      }
      else if (  message->isProducer() )   // Handling Producer
      {
        printf( "Producer message \n" );
        fflush( stdout );

        if ( nbConsumer > 0 )
        {
          // Feed the Consumer from the current Producer
          register size_t  i   = 0;
          register size_t  max = timeout.size();
          iter = timeout.begin();
          for( i = 0; i < max; i++, iter++ )
          {
           if ( timeout[ i ] != NULL )
           {
            if ( timeout[ i ]->getNumber() != -1  &&  timeout[ i ]->isConsumer() )
            {
              printf( "Consumer fed by entry %d, producer %d \n", i, timeout[ i ]->getNumber() );
              fflush( stdout );

              timeout[ i ]->stop();
              timeout[ i ]->reply( rcvid );                   // Reply Consumer
              message->reply( timeout[ i ]->getReceiver() );  // Reply Producer

              // Delete that entry
              deleteEntry( i );

              nbConsumer--;
              i = max;

              printf( "Done with case 3 ! \n" );
              fflush( stdout ); 
            }
           }
          }
        }
        else
        {
          // Storing Producer, No consumer Waiting.
          printf( "Storing Producer %d \n", message->getNumber() );
          fflush( stdout );

          createEntry( rcvid, message, 0 );
          nbProducer++;

          printf( "Done with case 4 ! \n" );
          fflush( stdout );
        }
      }
      else  // Generic reply
      {
        printf( "Generic reply \n" );
        fflush( stdout );

          int data = 100;
          message->reply( rcvid, data );

        printf( "Done with case 5 ! \n" );
        fflush( stdout );

     }

      printf( "End of message response. \n" );
      printf( "****************************************************************\n" );
      fflush( stdout );

    } // End of while loop

    printf( "Server Disconnect \n" );
    fflush( stdout );

    disconnect();
  }

#endif // End of Server.cpp

---