/**\file n_network.c
 *  Network Engine
 *\author Castagnier Mickael
 *\version 1.0
 *\date 10/05/2005
 */

#include "n_common.h"
#include "n_log.h"
#include "n_str.h"
#include "n_list.h"
#include "n_network.h"



/*!\fn int handle_wsa( int mode , int v1 , int v2 )
 *\brief Do not directly use, internal api.
 *\param mode 1 for opening 0 for close 2 for status
 *\param v1 First digit of version requested
 *\param v2 Second digit of version requested
 *\return TRUE on success FALSE on error
 */
int handle_wsa( int mode , int v1 , int v2 )
{
   int compiler_warning_suppressor = 0 ;
#if !defined( LINUX ) && !defined( SOLARIS ) && !defined( AIX )
   static WSADATA WSAdata; /*WSA world*/
   static int WSA_IS_INITIALIZED = 0; /*status checking*/

   switch ( mode )
   {
      /*returning WSA status*/
      case 2 :
         return WSA_IS_INITIALIZED;
         break;
         /*loading WSA dll*/
      case 1 :
         if ( WSA_IS_INITIALIZED == 1 )
            return TRUE; /*already loaded*/
         if ( ( WSAStartup( MAKEWORD( v1 , v2 ), &WSAdata ) ) != 0 )
         {
            WSA_IS_INITIALIZED = 0;
            return FALSE;
         }
         else
         {
            WSA_IS_INITIALIZED = 1 ;
            return TRUE;
         }
         break;
         /*unloading (closing) WSA */
      case 0 :
         if ( WSA_IS_INITIALIZED == 0 )
            return TRUE; /*already CLEANED or not loaded */
         if ( WSACleanup() == 0 )
         {
            WSA_IS_INITIALIZED = 0;
            return TRUE;
         }
         break;
   } /*switch(...)*/
#endif /*#ifndef LINUX SOLARIS AIX */
   compiler_warning_suppressor = mode + v1 + v2 ;
   compiler_warning_suppressor = TRUE ;
   return compiler_warning_suppressor ;
} /*handle_wsa(...)*/



/*!\fn int netw_setsockopt( SOCKET sock , int disable_naggle , int sock_send_buf , int sock_recv_buf )
 *\brief Modify common socket options. Enable SO_REUSEADDR.  
 *\param sock The socket to configure
 *\param disable_naggle Set to positive to enable. Set to zero or negative to leave the default.
 *\param sock_send_buf If >0, then try to set the socket send buffer size in octet
 *\param sock_recv_buf If >0, then try to set the socket recv buffer size in octet
 *\return TRUE or FALSE
 */
int netw_setsockopt( SOCKET sock , int disable_naggle , int sock_send_buf , int sock_recv_buf )
{
   /* disable naggle algorithm */
   if( disable_naggle > 0 )
   {
      int tmp = 1;
      if ( setsockopt( sock , IPPROTO_TCP, TCP_NODELAY, ( const char * ) &tmp, sizeof( tmp ) ) == -1 )
      {
         n_log( LOG_ERR , "Error from setsockopt(TCP_NODELAY). errno: %s" , strerror( errno ) );
         return FALSE ;
      }
   }
   /* socket sending buffer size */
   if( sock_send_buf > 0 )
   {
      if ( setsockopt ( sock , SOL_SOCKET, SO_SNDBUF, ( const char * ) &sock_send_buf, sizeof( sock_send_buf ) ) == -1 )
      {
         n_log( LOG_ERR , "Error from setsockopt(SO_SNDBUF). errno: %s" , strerror( errno ) );
         return FALSE ;
      }
   }
   /* socket receiving buffer */
   if( sock_recv_buf > 0 )
   {
      if ( setsockopt ( sock , SOL_SOCKET, SO_RCVBUF, ( const char * ) &sock_recv_buf, sizeof( sock_recv_buf ) ) == -1 )
      {
         n_log( LOG_ERR , "Error from setsockopt(SO_RCVBUF). errno: %s" , strerror( errno ) );
         return FALSE ;
      }
   }
   /* lose the pesky "Address already in use" error message*/
   if ( setsockopt( sock , SOL_SOCKET, SO_REUSEADDR , "1", sizeof( int ) ) == -1 )
   {
      n_log( LOG_ERR , "Error from setsockopt(SO_REUSEADDR). errno: s" , strerror( errno ) );
      return FALSE ;
   }
   return TRUE ;
} /* netw_setsockopt( ... ) */



/*!\fn int netw_tcp_socket( SOCKET *sock )
 *\brief Get a new tcp socket
 *\param sock Pointer to a valid and empty SOCKET
 *\return TRUE or FALSE (in that case sock is untouched)
 */
int netw_tcp_socket( SOCKET *sock )
{
   int tmpsock = 0 ;

   /* new socket */
   tmpsock = socket( AF_INET , SOCK_STREAM , 0 );
   if( tmpsock < 0 )
   {
      n_log( LOG_ERR , "Error from socket(). errno: %s" , strerror( errno ) );
      return FALSE ;
   }
   (*sock) = tmpsock ;

   return TRUE ;
} /* new_tcp_socket( ... ) */



/*!\fn int netw_connect_ex(NETWORK **netw , char *host , int port, int disable_naggle , int sock_send_buf , int sock_recv_buf , int send_list_limit , int recv_list_limit )
 *\brief Use this to connect a NETWORK to any listening one
 *\param netw a NETWORK *object
 *\param host Host or IP to connect to
 *\param port Port to use to connect
 *\param disable_naggle Disable Naggle Algorithm. Set to 1 do activate, 0 or negative to leave defaults
 *\param sock_send_buf NETW_SOCKET_SEND_BUF socket parameter , 0 or negative to leave defaults
 *\param sock_recv_buf NETW_SOCKET_RECV_BUF socket parameter , 0 or negative to leave defaults
 *\param send_list_limit Internal sending list maximum number of item. 0 or negative for unrestricted
 *\param recv_list_limit Internal receiving list maximum number of item. 0 or negative for unrestricted
 *\return TRUE or FALSE
 */
int netw_connect_ex( NETWORK **netw , char *host , int port, int disable_naggle , int sock_send_buf , int sock_recv_buf , int send_list_limit , int recv_list_limit  )
{
   struct hostent *hostinfo = NULL ;

   /*checking WSA when under windows*/
   if ( handle_wsa( 1 , 2 , 2 ) == FALSE )
   {
      n_log( LOG_ERR , "Unable to load WSA dll's" );
      return FALSE ;
   }

   /*do not work over an already used netw*/
   if( (*netw) )
   {
      n_log( LOG_ERR , "Unable to allocate (*netw), already existing. You must use empty NETWORK *structs." );
      return FALSE ;
   }

   hostinfo = gethostbyname( host );
   if( hostinfo == NULL )
   {
      n_log( LOG_ERR , "Erreur -> host : %s inconnu.", host );
      return FALSE ;
   }

   /*creating array*/
   Malloc( ( *netw ) , NETWORK , 1 );
   _assert( (*netw) , return FALSE );

   /* new socket */
   if( netw_tcp_socket( &(*netw) -> link . sock ) == FALSE )
   {
      n_log( LOG_ERR , "Error from socket, %s" , strerror( errno ) );
      Free( (*netw ) );
      return FALSE ;
   }

   if( netw_setsockopt( (*netw) -> link . sock , disable_naggle , sock_send_buf , sock_recv_buf ) != TRUE )
   {
      n_log( LOG_ERR , "Some socket options could not be set on %d" , (*netw) -> link . sock );   
   }

   /*filling host information*/
   memcpy(&( *netw ) -> link . raddr . sin_addr , hostinfo -> h_addr_list[ 0 ] , hostinfo -> h_length );
   ( *netw ) -> link . raddr . sin_family = AF_INET;
   ( *netw ) -> link . raddr . sin_port = htons( port );

   /*filling local information*/
   ( *netw ) -> link . haddr . sin_addr . s_addr = htonl( INADDR_ANY );
   ( *netw ) -> link . haddr . sin_family = AF_INET;
   ( *netw ) -> link . haddr . sin_port = htons( port );
   ( *netw ) -> link . port = port;

   /*zero the rest of the struct */
   memset( &( ( *netw ) -> link . haddr . sin_zero ), 0 , 8 );    
   memset( &( ( *netw ) -> link . raddr . sin_zero ), 0 , 8 ); 

   /*connecting*/
   if( connect( ( *netw ) -> link .sock , ( struct sockaddr * ) &( ( *netw ) -> link . raddr ) , sizeof( struct sockaddr ) ) != 0 )
   {
      n_log( LOG_ERR , "Error from connect(). errno: %s" , strerror( errno ) );
      closesocket( (*netw) -> link . sock );
      Free( (*netw) );
      return FALSE ;
   }

   /*storing connected to ip adress*/
   ( *netw ) -> link . ip = strdup( host );

   /*initialize queues */
   ( *netw ) -> recv_buf = new_generic_list( recv_list_limit );
   if( ! (*netw ) -> recv_buf )
   {
      n_log( LOG_ERR , "Error when creating receive list with %d item limit" , recv_list_limit );
      closesocket( (*netw) -> link . sock );
      Free( (*netw) );
      return FALSE ;
   }
   (*netw) -> send_buf = new_generic_list( send_list_limit );
   if( ! (*netw ) -> send_buf )
   {
      n_log( LOG_ERR , "Error when creating send list with %d item limit" , send_list_limit );
      list_destroy( &( *netw ) -> recv_buf );
      closesocket( (*netw) -> link . sock );
      Free( (*netw) );
      return FALSE ;
   }

   /*initiliaze mutexs*/
   if ( pthread_mutex_init( &( *netw ) -> sendbolt , NULL ) != 0 )
   {
      n_log( LOG_ERR , "Error initializing (*netw) -> sendbolt" ); 
      list_destroy( &( *netw ) -> recv_buf );
      list_destroy( &( *netw ) -> send_buf );
      closesocket( (*netw) -> link . sock );
      Free( (*netw) );
      return FALSE ;
   }
   if ( pthread_mutex_init( &( *netw ) -> recvbolt , NULL ) != 0 )
   { 
      n_log( LOG_ERR , "Error initializing (*netw) -> recvbolt" ); 
      pthread_mutex_destroy( &( *netw ) -> sendbolt );
      list_destroy( &( *netw ) -> recv_buf );
      list_destroy( &( *netw ) -> send_buf );
      closesocket( (*netw) -> link . sock );
      Free( (*netw) );
      return FALSE ;
   }
   if ( pthread_mutex_init( &( *netw ) -> eventbolt , NULL ) != 0 )
   {
      n_log( LOG_ERR , "Error initializing (*netw) -> eventbolt" ); 
      pthread_mutex_destroy( &( *netw ) -> sendbolt );
      pthread_mutex_destroy( &( *netw ) -> recvbolt );
      list_destroy( &( *netw ) -> recv_buf );
      list_destroy( &( *netw ) -> send_buf );
      closesocket( (*netw) -> link . sock );
      Free( (*netw) );
      return FALSE ;
   }
   (*netw) -> state = NETW_RUN ;
   (*netw) -> threaded_engine_status = NETW_THR_ENGINE_STOPPED ;

   return TRUE ;
} /* netw_connect_ex(...)*/



/*!\fn int netw_connect( NETWORK **netw , char *host , int port )
 *\brief Use this to connect a NETWORK to any listening one, with following defaults: disable_naggle = 0 , untouched sockbufsendsize and sockbufrecvsize and unrestricted sned/recv lists
 *\param netw a NETWORK *object
 *\param host Host or IP to connect to
 *\param port Port to use to connect
 *\return TRUE or FALSE
 */
int netw_connect( NETWORK **netw , char *host , int port )
{
   return netw_connect_ex( &(*netw) , host , port , 0 , 0 , 0 , 0 , 0 );
} /* netw_connect() */



/*!\fn int netw_set( NETWORK *netw , int flag )
 *\brief Restart or reset the specified network ability
 *\param netw The NETWORK *connection to modify
 *\param flag NETW_EMPTY_SENDBUF, NETW_EMPTY_RECVBUF, NETW_PAUSE , NETW_RUN
 *\return TRUE or FALSE
 */
int netw_set( NETWORK *netw , int flag )
{
   /* empty send queue */
   if( flag & NETW_EMPTY_SENDBUF )
   {
      pthread_mutex_lock( &netw -> sendbolt );
      list_empty( netw -> send_buf );
      pthread_mutex_unlock( &netw -> sendbolt );
   };
   /* empty recv queue */
   if( flag & NETW_EMPTY_RECVBUF )
   {
      pthread_mutex_lock( &netw -> recvbolt );
      list_empty( netw -> recv_buf );
      pthread_mutex_unlock( &netw -> recvbolt );
   }
   /* pause network */
   if( flag & NETW_PAUSE )
   {
      pthread_mutex_lock( &netw -> eventbolt );
      netw -> state = NETW_PAUSE ;
      pthread_mutex_unlock( &netw -> eventbolt );
   }
   /* unpause network */
   if( flag & NETW_RUN )
   {
      pthread_mutex_lock( &netw -> eventbolt );
      netw -> state = NETW_RUN ;
      pthread_mutex_unlock( &netw -> eventbolt );
   }
   return TRUE;
} /* netw_set(...) */



/*!\fn netw_close( NETWORK **netw )
 *\brief Closing a specified Network, destroy queues, free the structure
 *\param netw A NETWORK *network to close
 *\return TRUE on success , FALSE on failure
 */
int netw_close( NETWORK **netw )
{
   _assert( (*netw) , return FALSE );

   /*closing connection*/
   closesocket( (*netw) -> link . sock );

   if( (*netw) -> link . ip )
   {
      Free( (*netw) -> link . ip );
   }
   
   /*list freeing*/
   list_destroy( &(*netw) -> send_buf );
   list_destroy( &(*netw) -> recv_buf );


   pthread_mutex_destroy( &(*netw) -> recvbolt );
   pthread_mutex_destroy( &(*netw) -> sendbolt );
   pthread_mutex_destroy( &(*netw) -> eventbolt );

   Free( (*netw) );

   return TRUE;
} /* netw_close(...)*/



/*!\fn int netw_make_listening( NETWORK **netw , int PORT , int nbpending )
 *\brief Make a NETWORK be a Listening network
 *\param netw A NETWORK **network to make listening
 *\param PORT For choosing a PORT to listen to
 *\param nbpending Number of pending connection when listening
 *\return TRUE on success, FALSE on error
 */
int netw_make_listening( NETWORK **netw , int PORT , int nbpending )
{
   /*checking WSA when under windows*/
   if ( handle_wsa( 1 , 2 , 2 ) == FALSE )
   {
      n_log( LOG_ERR , "Unable to load WSA dll's" );
      return FALSE ;
   }

   if( *netw )
   {
      n_log( LOG_ERR , "Cannot use an allocated network. Please pass a NULL network to modify" );
      return FALSE; 
   }

   /*creating array*/
   Malloc( *netw , NETWORK , 1 );

   if( netw_tcp_socket( &(*netw) -> link . sock ) == FALSE )
   {
      n_log( LOG_ERR , "Error from socket, %s" , strerror( errno ) );
      Free( (*netw ) );
      return FALSE ;
   }

   if( netw_setsockopt( (*netw) -> link . sock , 0 , 0 , 0 ) != TRUE )
   {
      n_log( LOG_ERR , "Some socket options could not be set on %d" , (*netw) -> link . sock );   
   }

   /*filling local information*/
   ( *netw ) -> link . haddr . sin_addr.s_addr = INADDR_ANY;
   ( *netw ) -> link . haddr . sin_family = AF_INET;
   ( *netw ) -> link . haddr . sin_port = htons( PORT );
   ( *netw ) -> nb_pending = nbpending;

   memset( &( ( *netw ) -> link . haddr . sin_zero ), 0 , 8 ); /* zero the rest of the struct*/
   memset( &( ( *netw ) -> link . raddr . sin_zero ), 0 , 8 ); /* zero the rest of the struct*/

   /*binding information to socket*/
   if ( bind( ( *netw ) -> link . sock , ( struct sockaddr * ) &( ( *netw ) -> link . haddr ) , sizeof( struct sockaddr ) ) == -1 )
   {
      n_log( LOG_ERR , "Error binding socket %d" , ( *netw ) -> link . sock );
      return FALSE;
   }

   /* nb_pending connections*/
   listen( ( *netw ) -> link . sock, ( *netw ) -> nb_pending );

   (*netw) -> mode = NETW_SERVER ;
   (*netw) -> state = NETW_RUN ;
   (*netw) -> threaded_engine_status = NETW_THR_ENGINE_STOPPED ;

   return TRUE;
} /* netw_make_listening(...)*/



/*!\fn NETWORK *netw_accept_from_ex( NETWORK *from , int disable_naggle , int sock_send_buf , int sock_recv_buf , int send_list_limit , int recv_list_limit )
 *\brief make a normal 'accept' . Network 'from' must be allocated with netw_make_listening.
 *\param from the network from where we accept
 *\param disable_naggle Disable Naggle Algorithm. Set to 1 do activate, 0 or negative to leave defaults
 *\param sock_send_buf NETW_SOCKET_SEND_BUF socket parameter , 0 or negative to leave defaults
 *\param sock_recv_buf NETW_SOCKET_RECV_BUF socket parameter , 0 or negative to leave defaults
 *\param send_list_limit Internal sending list maximum number of item. 0 or negative for unrestricted
 *\param recv_list_limit Internal receiving list maximum number of item. 0 or negative for unrestricted
 *\return NULL on failure, if not a pointer to the connected network
 */
NETWORK *netw_accept_from_ex( NETWORK *from , int disable_naggle , int sock_send_buf , int sock_recv_buf , int send_list_limit , int recv_list_limit )
{
   int tmp = 0 ;

#if defined( LINUX ) || defined( SOLARIS ) || defined( AIX )
   socklen_t sin_size = 0 ;
#else
   int sin_size = 0 ;
#endif

   NETWORK *netw = NULL ;

   /*checking WSA when under windows*/
   if( handle_wsa( 1 , 2 , 2 ) == FALSE )
   {
      n_log( LOG_ERR , "Unable to load WSA dll's" );
      return FALSE ;
   }

   _assert( from , return FALSE );

   Malloc( netw , NETWORK , 1 );

   sin_size = sizeof( struct sockaddr_in );

   tmp = accept( from -> link . sock, ( struct sockaddr * ) & ( netw -> link . raddr ) , &sin_size );
   if ( tmp < 0 )
   {
      n_log( LOG_ERR , "error accepting on %d" , netw -> link . sock );
      Free( netw );
      return NULL;
   }

   netw -> link . sock = tmp ;
   netw -> link . port = from -> link . port;

   if( netw_setsockopt( netw -> link . sock , disable_naggle , sock_send_buf , sock_recv_buf ) != TRUE )
   {
      n_log( LOG_ERR , "Some socket options could not be set on %d" , netw-> link . sock );   
   }
   /*initialize queues */
   netw -> recv_buf = new_generic_list( recv_list_limit );
   if( !netw -> recv_buf )
   {
      n_log( LOG_ERR , "Error when creating receive list with %d item limit" , recv_list_limit );
      closesocket( netw -> link . sock );
      Free( netw );
      return FALSE ;
   }
   netw -> send_buf = new_generic_list( send_list_limit );
   if( ! netw -> send_buf )
   {
      n_log( LOG_ERR , "Error when creating send list with %d item limit" , send_list_limit );
      list_destroy( &netw -> recv_buf );
      closesocket( netw -> link . sock );
      Free( netw );
      return FALSE ;
   }

   /*initiliaze mutexs*/
   if( pthread_mutex_init( &netw -> sendbolt , NULL ) != 0 )
   {
      n_log( LOG_ERR , "Error initializing netw -> sendbolt" ); 
      list_destroy( &netw -> recv_buf );
      list_destroy( &netw -> send_buf );
      closesocket( netw -> link . sock );
      Free( netw );
      return FALSE ;
   }
   if( pthread_mutex_init( &netw -> recvbolt , NULL ) != 0 )
   { 
      n_log( LOG_ERR , "Error initializing netw -> recvbolt" ); 
      pthread_mutex_destroy( &netw -> sendbolt );
      list_destroy( &netw -> recv_buf );
      list_destroy( &netw -> send_buf );
      closesocket( netw -> link . sock );
      Free( netw );
      return FALSE ;
   }
   if ( pthread_mutex_init( &netw -> eventbolt , NULL ) != 0 )
   {
      n_log( LOG_ERR , "Error initializing (*netw) -> eventbolt" ); 
      pthread_mutex_destroy( &netw -> sendbolt );
      pthread_mutex_destroy( &netw -> recvbolt );
      list_destroy( &netw -> recv_buf );
      list_destroy( &netw -> send_buf );
      closesocket( netw -> link . sock );
      Free( netw );
      return FALSE ;
   }

   netw -> mode = NETW_SERVER ;
   netw -> state = NETW_RUN ;
   netw -> threaded_engine_status = NETW_THR_ENGINE_STOPPED ;
   netw -> link . ip = NULL ;

   return netw;
} /* netw_accept_from_ex(...) */



/*!\fn NETWORK *netw_accept_from( NETWORK *from )
 *\brief make a normal 'accept' . Network 'from' must be allocated with netw_make_listening.
 *\param from The network from which to obtaion the connection
 *\return NULL 
 */
NETWORK *netw_accept_from( NETWORK *from )
{
   return netw_accept_from_ex( from , 0 , 0 , 0 , 0 , 0 );
} /* network_accept_from( ... ) */



/*!\fn int netw_add_msg( NETWORK *netw , N_STR *msg )
 *\brief Add a message to send in aimed NETWORK
 *\param netw NETWORK where add the message
 *\param msg  the message to add
 *\return TRUE if success FALSE on error
 */
int netw_add_msg( NETWORK *netw , N_STR *msg )
{
   _assert( netw , return FALSE );
   _assert( msg , return FALSE );
   _assert( msg -> data , return FALSE );

   if( msg -> length <= 0 )
   {
      n_log( LOG_ERR , "Empty messages are not supported. msg(%p)->lenght=%d" , msg , msg -> length );
      return FALSE;
   }

   pthread_mutex_lock( &netw -> sendbolt );

   if( list_push( netw -> send_buf , msg , free_nstr_ptr ) == FALSE )
   {
      pthread_mutex_unlock( &netw -> sendbolt );
      return FALSE;
   }

   pthread_mutex_unlock( &netw -> sendbolt );

   return TRUE;
} /* netw_add_msg(...) */



/*!\fn N_STR *netw_get_msg( NETWORK *netw )
 *\brief Get a message from aimed NETWORK
 *\param netw NETWORK where get the msg
 *\return NULL or a filled N_STR
 */
N_STR *netw_get_msg( NETWORK *netw )
{
   N_STR * ptr = NULL ;

   _assert( netw , return NULL );
   
   pthread_mutex_lock( &netw -> recvbolt );

   ptr = list_shift( netw -> recv_buf , N_STR );

   pthread_mutex_unlock( &netw -> recvbolt );

   return ptr ;
} /* netw_get_msg(...)*/



/*!\fn N_STR *netw_wait_msg( NETWORK *netw , int refresh , int timeout )
 *\brief Wait a message from aimed NETWORK. Recheck each usec until a valid
 *\param netw The link on which we wait a message
 *\param refresh The time in usec between each check until there is a message
 *\param timeout Maximum amount of time to wait before returnin. Pass a value <= 0 to disable.
 *\return NULL or a filled N_STR 
 */
N_STR *netw_wait_msg( NETWORK *netw , int refresh , int timeout )
{
   N_STR *nstrptr = NULL ;
   int DONE = 0 , timed = 0 ;

   _assert( netw , return NULL );

   if( timeout > 0 )
      timed = 1 ;

   while( !DONE )
   {
      nstrptr = netw_get_msg( netw );
      if( nstrptr )
         return nstrptr ;

      usleep( refresh );
      if( timed == 1 )
      {
         timeout -= refresh ;
         if( timeout < 0 )
            break ;
      }
   }
   return NULL ;
} /* netw_wait_msg(...) */



/*!\fn netw_start_thr_engine( NETWORK *netw )
 *\brief Start the NETWORK netw Threaded Engine. Create a sending & receiving thread.
 *\param netw The aimed NETWORK connection to start receiving data
 *\return TRUE on success FALSE on failure
 */
int netw_start_thr_engine( NETWORK *netw )
{
   _assert( netw , return FALSE );

   pthread_mutex_lock( &netw -> eventbolt );
   if( netw -> threaded_engine_status == NETW_THR_ENGINE_STARTED )
   {
      n_log( LOG_ERR , "THR Engine already started for network %p" , netw );
      pthread_mutex_unlock( &netw -> eventbolt );
      return FALSE ;
   }

   if( pthread_create(  &netw -> recv_thr ,   NULL ,  netw_recv_func ,  (void *) netw ) != 0 )
   {
      n_log( LOG_ERR , "Unable to create recv_thread for network %p" , netw );
      pthread_mutex_unlock( &netw -> eventbolt );
      return FALSE ;
   }
   if( pthread_create(  &netw -> send_thr ,   NULL ,  netw_send_func ,  (void *) netw ) != 0 )
   {
      n_log( LOG_ERR , "Unable to create send_thread for network %p" , netw );
      pthread_mutex_unlock( &netw -> eventbolt );
      return FALSE ;
   }

   netw -> threaded_engine_status = NETW_THR_ENGINE_STARTED ;

   pthread_mutex_unlock( &netw -> eventbolt );

   return TRUE ;
}  /* netw_create_recv_thread(....) */



/*!\fn void *netw_send_func( void *NET )
 *\brief Thread send function
 *\param NET the NETWORK connection to use, casted into (void*)
 */
void *netw_send_func( void *NET )
{
   int DONE = 0 ,
       state = 0 ;

   NSTRBYTE nboctet = 0 ;
       
   char nboct[4];

   N_STR *ptr = NULL ;

   NETWORK *netw = (NETWORK *) NET ;

   _assert( netw , return NULL );

   do
   {
      pthread_mutex_lock( &netw -> eventbolt );
      state =  netw -> state ;
      pthread_mutex_unlock( &netw -> eventbolt );

      if( state == NETW_QUIT )
      {
         DONE = 3;

         /* preparing the number of octet */
         nboctet = NETW_QUIT ;
         nboctet = htonl( nboctet );

         memcpy( nboct , &nboctet , sizeof( NSTRBYTE ) );
         n_log( LOG_DEBUG , "%d Sending Quit !" , netw -> link . sock );

         /* sending the number of octet to receive on next message */
         if ( send_data( netw -> link . sock , nboct , sizeof( NSTRBYTE ) ) == -1 )
            DONE = 4 ;
      }
      else
      {
         if( state != NETW_PAUSE )
         {
            pthread_mutex_lock( &netw -> sendbolt );

            ptr = list_shift( netw -> send_buf , N_STR ) ;
            if( ptr && ptr -> length > 0 && ptr -> data )
            {
               n_log( LOG_DEBUG , "Sending ptr size %ld written %ld" , ptr -> length , ptr -> written );

               /* preparing the number of octet */
               nboctet = ptr -> length;
               nboctet = htonl( nboctet );

               memcpy( nboct , &nboctet , sizeof( NSTRBYTE ) );

               /* sending the number of octet to receive on next message */
               if ( !DONE && send_data( netw -> link . sock , nboct , sizeof( NSTRBYTE ) ) == -1 )
                  DONE = 1 ;

               /* sending the data itself */
               if ( !DONE && send_data( netw -> link . sock , ptr->data , ptr -> length ) == -1 )
                  DONE = 2 ;

               free_nstr( &ptr );
               pthread_mutex_unlock( &netw -> sendbolt );
#if !defined( LINUX ) && !defined( SOLARIS ) && !defined( AIX )
               rest( 0 );
#else
               usleep( 0 ); /* we will have to make a sleep thing for that */
#endif
            }
            else
            {
               pthread_mutex_unlock( &netw -> sendbolt );
#if !defined( LINUX ) && !defined( SOLARIS ) && !defined( AIX ) 
               rest( 1 );
#else
               usleep( 1000 );
#endif
            }
         }
         else
         {
#if !defined( LINUX ) && !defined( SOLARIS ) && !defined( AIX ) 
            rest( 10 );
#else
            usleep( 10000 );
#endif
         }
      }
   }while( !DONE );

   if( DONE == 1 )
      n_log( LOG_ERR ,  "Error when sending number of octet on socket %d" , netw -> link . sock );
   else if( DONE == 2 )
      n_log( LOG_ERR ,  "Error when sending data from socket %d" , netw -> link . sock );
   else if( DONE == 3 )
      n_log( LOG_DEBUG ,  "Quit asked, closing sending thread on socket %d" , netw -> link . sock );
   else if( DONE == 4 )
      n_log( LOG_ERR ,  "Error when sending number of QUIT on socket %d" , netw -> link . sock );
   else
      n_log( LOG_ERR , "Closing send thread for socket %d\nReason unknown\nDONE = %d" , netw -> link . sock , DONE );

   pthread_mutex_lock( &netw -> eventbolt );
   if( DONE == 4 )
      netw -> state = NETW_QUIT ;
   else
      netw -> state = NETW_ERROR ;
   pthread_mutex_unlock( &netw -> eventbolt );

   n_log( LOG_DEBUG , "Socket %d: Sending thread exited !" , netw -> link . sock );

   pthread_exit( 0 );

   /* suppress compiler warning */
#if !defined( LINUX ) && !defined( SOLARIS )
   return NULL ;
#endif
} /* netw_send_func(...) */



/*!\fn void *netw_recv_func( void *NET )
 *\brief To Thread Receiving function
 *\param NET the NETWORK connection to use
 */
void *netw_recv_func( void *NET )
{
   int DONE = 0 ,
       state = 0 ,
       tmpstate = 0 ;

   NSTRBYTE nboctet = 0 ;

   char nboct[4];

   N_STR *recvdmsg = NULL ;

   NETWORK *netw = (NETWORK *) NET ;

   _assert( netw , return NULL );

   do
   {
      pthread_mutex_lock( &netw -> eventbolt );
      state = netw -> state ;
      pthread_mutex_unlock( &netw -> eventbolt );

      if( state == NETW_QUIT )
      {
         DONE = 4;
      }
      else
      {
         /* receiving the number of octet to receive on next message */
         if( !DONE && state != NETW_PAUSE )
         {
            if( recv_data( netw -> link . sock , nboct , sizeof( NSTRBYTE ) ) == -1 )
               DONE = 1 ;
            else
            {
               memcpy( &nboctet , nboct , sizeof( NSTRBYTE ) );
               tmpstate = ntohl( nboctet );
               nboctet = tmpstate ;
               if( tmpstate == NETW_QUIT )
               {
                  n_log( LOG_DEBUG ,  "%d receiving order to QUIT, nboctet %d NETW_QUIT %d !" , netw -> link . sock , nboctet , NETW_QUIT );

                  DONE = 4;
                  pthread_mutex_lock( &netw -> eventbolt );
                  netw -> state = NETW_QUIT ;
                  pthread_mutex_unlock( &netw -> eventbolt );
               }
               else
               {
                  Malloc( recvdmsg , N_STR , 1 );
                  if( !recvdmsg )
                     DONE = 3 ;
                  else
                  {
                     n_log( LOG_DEBUG ,  "%d octets to receive..." , nboctet );
                     Malloc( recvdmsg -> data , char , nboctet + 1 );
                     recvdmsg -> length  = nboctet + 1 ;
                     recvdmsg -> written = nboctet ;

                     /* receiving the data itself */
                     if ( recv_data( netw -> link . sock , recvdmsg -> data , nboctet ) == -1 )
                        DONE = 2 ;

                     pthread_mutex_lock( &netw -> recvbolt );

                     if( !DONE && list_push( netw -> recv_buf , recvdmsg , free_nstr_ptr ) == FALSE )
                        DONE = 5 ;

                     pthread_mutex_unlock( &netw -> recvbolt );
                  }
               }
            }
         }
         else
         {
#if !defined( LINUX ) && !defined( SOLARIS ) && !defined( AIX ) 
            rest( 10 );
#else
            usleep( 10000 );
#endif
         }
      }

   }while( !DONE );

   if( DONE == 1 )
      n_log( LOG_ERR ,   "Error when receiving number of octet from socket %d" , netw -> link . sock );
   else if( DONE == 2 )
      n_log( LOG_ERR ,  "Error when receiving data from socket %d" , netw -> link . sock );
   else if( DONE == 3 )
      n_log( LOG_ERR ,  "Error allocating receved message from socket %d" , netw -> link . sock );
   else if( DONE == 4 )
      n_log( LOG_DEBUG ,  "Quit asked, closing receiving thread on socket %d" , netw -> link . sock );
   else if( DONE == 5 )
      n_log( LOG_ERR ,  "Error adding receved message from socket %d" , netw -> link . sock );
   else
      n_log( LOG_ERR , "Closing recv thread for socket %d\nReason unknown\nDONE = %d" , netw -> link . sock , DONE );


   pthread_mutex_lock( &netw -> eventbolt );
   if( DONE == 4 )
      netw -> state = NETW_QUIT ;
   else
      netw -> state = NETW_ERROR ;
   pthread_mutex_unlock( &netw -> eventbolt );

   n_log( LOG_DEBUG , "Socket %d: Receive thread exited !" , netw -> link . sock );

   pthread_exit( 0 );

   /* suppress compiler warning */
#if !defined( LINUX ) && !defined( SOLARIS )
   return NULL ;
#endif
} /* netw_recv_func(...) */



/*!\fn netw_stop_thr_engine( NETWORK *netw )
 *\brief Stop a NETWORK connection sending and receing thread
 *\param netw The aimed NETWORK conection to stop
 *\return TRUE on success FALSE on failure
 */
int netw_stop_thr_engine( NETWORK *netw )
{
   _assert( netw ,  return FALSE );

   n_log( LOG_DEBUG ,  "Network %d stop threads event" , netw -> link . sock );

   pthread_mutex_lock( &netw -> eventbolt );
   if( netw -> threaded_engine_status != NETW_THR_ENGINE_STARTED )
   {
      n_log( LOG_ERR , "Couldn not stop: network thread engine not started for network %p" , netw );
      pthread_mutex_unlock( &netw -> eventbolt );
      return FALSE ;
   }

   netw -> state = NETW_QUIT ;

   pthread_mutex_unlock( &netw -> eventbolt );

   n_log( LOG_DEBUG ,  "Network %d waits for send threads to stop" , netw -> link . sock );
   pthread_join( netw -> send_thr , NULL );
   n_log( LOG_DEBUG ,  "Network %d waits for recv threads to stop" , netw -> link . sock );
   pthread_join( netw -> recv_thr , NULL );

   pthread_mutex_lock( &netw -> eventbolt );
   netw -> state = NETW_STOP ;
   netw -> threaded_engine_status = NETW_THR_ENGINE_STOPPED ;
   pthread_mutex_unlock( &netw -> eventbolt );

   n_log( LOG_DEBUG ,  "Network %d threads Stopped !" , netw -> link . sock );

   return TRUE;
} /* Stop_Network( ... ) */



/*!\fn send_data( SOCKET s , char *buf, NSTRBYTE n )
 *\brief send data onto the socket
 *\param s connected socket
 *\param buf pointer to buffer
 *\param n number of characters we want to send
 *\return -1 on error, n on success
 */
int send_data( SOCKET s, char *buf, NSTRBYTE n )
{
   int bcount = 0 ; /* counts bytes read */
   int br = 0 ;     /* bytes read this pass */

   _assert( buf , return -1 );

   if( n == 0 )
   {
      n_log( LOG_ERR , "Send of 0 is unsupported." );
      return -1 ;
   }

   while( (NSTRBYTE)bcount < n )                 /* loop until full buffer */
   {
      if ( ( br = send( s, buf, n - bcount, NETFLAGS ) ) > 0 )
      {
         bcount += br;                /* increment byte counter */
         buf += br;                   /* move buffer ptr for next read */
      }
      else
         if ( br <= 0 )
         {                            /* signal an error to the caller */
            n_log( LOG_ERR ,  "Socket %d sending Error" , s );
            return -1 ;
         }
   }

   return bcount;
} /*send_data(...)*/



/*!\fn recv_data( SOCKET s , char *buf, NSTRBYTE n )
 *\brief recv data from the socket
 *\param s connected socket
 *\param buf pointer to buffer
 *\param n number of characters we want
 *\return -1 on error, n on success
 */
int recv_data( SOCKET s, char *buf , NSTRBYTE n )
{
   int bcount = 0 ; /* counts bytes read */
   int br = 0 ;     /* bytes read this pass */

   _assert( buf , return -1 );

   if( n == 0 )
   {
      n_log( LOG_ERR , "Recv of 0 is unsupported." );
      return -1 ;
   }

   while( (NSTRBYTE)bcount < n )
   {                                        /* loop until full buffer */

      if ( ( br = recv( s, buf, n - bcount, NETFLAGS ) ) > 0 )
      {
         bcount += br;                    /* increment byte counter */
         buf += br;                       /* move buffer ptr for next read */
      }
      else
         if ( br <= 0 )
         {                                /* signal an error to the caller */
            n_log( LOG_ERR ,  "Socket %d receive Error" , s );
            return -1 ;
         }
   }

   return bcount;
} /*recv_data(...)*/
