ipc.c

一个功能全面的电子邮件客户端

  qStream *packet ;
  qPayload payload ;
  char buffer[MSGQ_MAXSIZE+1] ;
  

#if DEBUG > 7
  printf( "recvMsg()\n" ) ;
#endif

  /* Now, issue the read against the specified message queue. */
  ret = readMsgQ( mqd,
		  &payload,
		  buffer,
		  to,
		  wait ) ;

  /*
   * As long as it was a good read, and the packet has data,
   * let's add it to the stream list, otherwise, it was only flow
   * control or something like that.
   */
  retValue = NULL ;
  if( ret != -1 )
    {
      packet = calloc( 1, sizeof(qStream) ) ;
      packet -> payload = malloc( sizeof(qPayload) ) ;
      memcpy( packet -> payload, &payload, sizeof(qPayload) ) ;

      packet -> msg = malloc( sizeof(qMsg) ) ;
      packet -> msg -> length = packet -> payload -> tsize ;
      packet -> msg -> msgCode = packet -> payload -> msgCode ;

      /*
       * Now, figure out what to do with this data packet.
       * The callee's become the owner of the packet and it's
       * memory.
       */
      retValue = dispatchMsgPacket( packet, buffer ) ;
#if 0
#if DEBUG > 3
      if( retValue )
	{
	  printf( "**********----**********\n%s**********----**********\n\n\n", retValue ) ;
	}
#endif
#endif
    }
  
  return retValue ;
}



/*
 * Add a packet (stream fragment) to the linked list.
 */
int
addStreamPacket( qStream *packet )
{
  int retValue ;


#if DEBUG > 6
  printf( "addStreamPacket( %p ), from pid %d.\n", packet, packet -> payload -> pid ) ;
#endif
  /*
   * Tail will always point to the current packet!!!
   */
  retValue = 0 ;
  if( tailQStream ) 
    {
      tailQStream -> next = packet ;
      packet -> prev = tailQStream ;
    }
  tailQStream = packet ;

  /* If this is the first packet, anchor will point to it! */
  if( anchorQStream == NULL ) anchorQStream = packet ;

#ifdef DMALLOC
  dmalloc_verify( 0 ) ;
#endif
  return retValue ;
}



/* 
 * Extract all fragments of a given pid to recreate a data stream
 * while removing selected packets from the linked list.  The result
 * is the recreated data stream is returned.  If the stream fails
 * any of its consistancy checks, the requested stream is purged and
 * this function returns NULL.
 */
char *
buildStream( pid_t pid )
{
  long frag ;
  char *data ;
  long offset ;
  long totalSize ;
  qStream *ptr, *next ;


  /*
   * Walk through the list, finding the first node that
   * matched the pid we are looking for.  Then, allocate
   * enough memory to hold the whole stream.
   */
  frag = 0L ;
  totalSize = offset = 0L ;

  /* Walk the list of buffers */
  data = NULL ;
  ptr = anchorQStream ;
  while( ptr )
    {
      next = ptr -> next ;
      if( ptr -> payload -> pid == pid )
	{
	  /* This means we found a fragment that we need; each fragment has the total size */
	  frag++ ;
	  if( data == NULL )
	    {
	      totalSize = ptr -> payload -> tsize ;
	      data = malloc( totalSize + 1 ) ;
	    }

	  /* Now copy the data to the destination buffer */
#ifdef DMALLOC
	  dmalloc_verify( 0 ) ;
#endif
	  memcpy( &data[offset], ptr -> msg -> data.cValue, ptr -> payload -> psize ) ;
	  offset += ptr -> payload -> psize ;
	  data[offset] = (char)0x00 ;  /* Data gotta be NULL terminated!!! For sure!!! */
#ifdef DMALLOC
	  dmalloc_verify( 0 ) ;
#endif

	  /* Take this node out of the linked list */
	  next = extractStreamPacket( ptr ) ;
		  
	  /* Free all memory allocated on this stream node */
	  freeQStream( ptr ) ;
#ifdef DMALLOC
	  dmalloc_verify( 0 ) ;
#endif
	}
      ptr = next ;
    }

  /* Now, make sure we are returning a valid stream */
  if( data && (offset != totalSize) )
    {
      free( data ) ;
      data = NULL ;
      fprintf( stderr, "Internal IPC error detected.  Corrupted message is being dis-guarded.\n" ) ;
    }
#ifdef DMALLOC
  dmalloc_verify( 0 ) ;
#endif

  return data ;
}



/*
 * This function will remove a node from the linked list.
 * Returns a pointer to the next node.
 */
qStream *
extractStreamPacket( qStream *packet )
{
  qStream *next ;


  /* See if this is the first node to come off */
  next = packet -> next ;
  if( packet == anchorQStream )
    {
#if 0
      fprintf( stderr, "anchor node is being removed.\n" ) ;
#endif
      if( tailQStream == packet ) 
	{
#if 0
	  fprintf( stderr, "anchor is HEAD and TAIL node\n" ) ;
#endif
	  tailQStream = anchorQStream = NULL ;
	}
      else
	{
#if 0
	  fprintf( stderr, "anchor is now what was anchor -> next\n" ) ;
#endif
	  anchorQStream = anchorQStream -> next ;
	  anchorQStream -> prev = NULL ;
	}
    }
  else
    {
      /* See if this is the last node to come off */
      if( packet == tailQStream )
	{
#if 0
	  fprintf( stderr, "tail node is being removed.\n" ) ;
#endif
	  if( tailQStream -> prev ) tailQStream -> prev -> next = NULL ;
	  tailQStream = tailQStream -> prev ;
	}
      else
	{
	  /* Now, we know this is a middle node to remove! */
#if 0
	  fprintf( stderr, "middle node is being removed.\n" ) ;
#endif
	  packet -> prev -> next = packet -> next ;
	  packet -> next -> prev = packet -> prev ;
	}
    }
#ifdef DMALLOC
  dmalloc_verify( 0 ) ;
#endif

  return next ;
}



/* 
 * Dispatch protocol and data packets alike
 * This function calls the proper message handler.
 */
char *
dispatchMsgPacket( qStream *packet, char *buffer )
{
  int ret ;
  char *retValue ;


  /* Determine if this is a DATA or PROTOCOL packet */
  retValue = NULL ;
  if( packet -> payload -> msgCode & MSG_DATA ) 
    {
      packet -> msg -> datType = cValue ;
      packet -> msg -> data.cValue = strdup( buffer ) ;
      retValue = processDataPacket( packet ) ;
    }
  else
    {
      /*
       * If it's not a data packet, it must be a protocol packet.
       * Switch on the msgCode to figure out how to process the protocol message.
       * If it's a logging message, we'll just take care of it, otherwise, pass it
       * to the correct function and let it handle it.
       */
      ret = 0 ;
      switch( packet -> payload -> msgCode )
	{
	case MSG_TEXT_ERROR:
	  insert_error( buffer ) ;
	  break ;

	case MSG_TEXT_WARN:
	  insert_warning( buffer ) ;
	  break ;

	case MSG_TEXT_INFO:
	  insert_message( buffer ) ;
	  break ;

	default:
	  /* This means it is an IPC Protocol Message */
	  packet -> msg -> datType = iValue ;
	  packet -> msg -> data.iValue = *(int *)buffer ;
#if DEBUG > 5
	  printf( "dispatchMsgPacket() just got a packet of type iValue (%d:%d).\n",
		  packet -> msg -> msgCode, *(int *)buffer ) ;
#endif
	  ret = processProtocolPacket( packet ) ;
	  break ;
	}

      /* Now, process what ret is */
    }
  
  return retValue ;
}



/* Process protcol packets */
int 
processProtocolPacket( qStream *packet )
{
  int wo ;
  pid_t pid ;
  int status ;
  int account ;
  int numMsgs ;
  int retValue ;

  retValue = 0 ;
#if DEBUG > 3
  printf( "processProtocolPacket() has %d.\n", packet -> payload -> msgCode ) ;
#endif
  switch( packet -> payload -> msgCode )
    {
    case MSG_MANAGE_PID:
      break ;

    case MSG_MANAGE_DONE:
    case MSG_WAIT :
      wo = WUNTRACED ;
      pid = packet -> msg -> data.iValue ;
#if DEBUG > 3
      printf( "Waiting on pid %d to complete.\n", pid ) ;
#endif
      waitpid( pid, &status, wo ) ;
#if DEBUG > 3
      printf( "Child completed with status of %d.\n", status ) ;
#endif
      break ;

    case MSG_ACCOUNT_LOCK:
#if DEBUG > 3
      printf( "Locking account %d\n", packet -> msg -> data.iValue ) ;
#endif
      account = packet -> msg -> data.iValue ;
      proxy_account_lock( account, 1 ) ;
      proxyNewAccountPidNodeInsert( account, packet -> payload -> pid ) ;
      break ;

    case MSG_ACCOUNT_UNLOCK:
#if DEBUG > 3
      printf( "Unlocking account %d\n", packet -> msg -> data.iValue ) ;
#endif
      account = packet -> msg -> data.iValue ;
      proxy_account_lock( account, 0 ) ;
      proxyAccountPidNodeDestroy( packet -> payload -> pid ) ;
      break ;

    case MSG_TOTAL_MESSAGES:
      numMsgs = packet -> msg -> data.iValue ;
      account = proxy_seek_account_by_pid( packet -> payload -> pid ) ;
      proxySetAccountTotalMessages( account, numMsgs ) ;
#if DEBUG > 3
      printf( "Account %d has %d total messages\n", account, numMsgs ) ;
#endif
      break ;

    case MSG_NEW_MESSAGES:
      numMsgs = packet -> msg -> data.iValue ;
      account = proxy_seek_account_by_pid( packet -> payload -> pid ) ;
#if DEBUG > 3
      printf( "Account %d has %d new messages\n", account, numMsgs ) ;
#endif
      break ;
    }

  return retValue ;
}



/* Process data packets */
char * 
processDataPacket( qStream *packet )
{
  char *retValue ;


  /* Now, if this is a DATA (cValue) packet, add it to the linked list */
#if DEBUG > 6
  printf( "processDataPacket() has %d.\n", packet -> payload -> msgCode ) ;
#endif
  retValue = NULL ;
  if( packet -> msg -> datType == cValue ) 
    {
      /* Save the stream fragment */
      addStreamPacket( packet ) ;

      /*
       * Now, we should check to see if this message completed packet
       * delivery of a stream.
       */
#if 1
      if( packet -> payload -> msgCode == MSG_MSG_LAST_BLOCK )
	retValue = buildStream( packet -> payload -> pid ) ;
#else
      printf( "msgCode = %X, MSG_MSG_LAST_BLOCK bit is turned '%s', ON has value of %X, OFF has value of %X.\n", 
	      packet -> payload -> msgCode,
	      (packet -> payload -> msgCode == MSG_MSG_LAST_BLOCK)?"ON":"OFF",
	      MSG_MSG_LAST_BLOCK, MSG_MSG_BLOCK ) ;
      if( packet -> payload -> msgCode == MSG_MSG_LAST_BLOCK )
	printf( "-----> Calling buildStream() to rebuild the data stream.\n" ) ;
#endif
	}

  return retValue ;
}