iochannel.c

来自「GNet是一个简单的网络库。它是目标定向的」· C语言 代码 · 共 837 行 · 第 1/2 页

  g_return_if_fail (id != NULL);

  state = (GNetIOChannelWriteAsyncState*) id;

  if (delete_buffer)
    g_free(state->buffer);

  /* Don't delete state if we're in an upcall.  We'll do it after the
     upcall. */
  if (state->in_upcall)
    return;

  while (g_source_remove_by_user_data(state))
    ;

  g_free(state);
}



/* called internally only */
static void
write_async_cancel (GNetIOChannelWriteAsyncState* state)
{
  while (g_source_remove_by_user_data(state))
    ;

  g_free(state);
}


static gboolean 
write_async_cb (GIOChannel* iochannel, GIOCondition condition, gpointer data)
{
  GNetIOChannelWriteAsyncState* state;

  state = (GNetIOChannelWriteAsyncState*) data;

  g_return_val_if_fail (iochannel, FALSE);
  g_return_val_if_fail (state, FALSE);
  g_return_val_if_fail (iochannel == state->iochannel, FALSE);

  /* Error */
  if (condition & (G_IO_ERR | G_IO_HUP | G_IO_NVAL))
    {

      state->in_upcall = TRUE;
      (state->func)(iochannel, state->buffer, state->length, state->n,
		    GNET_IOCHANNEL_WRITE_ASYNC_STATUS_ERROR, state->user_data);
      state->in_upcall = FALSE;
      write_async_cancel (state);

      return FALSE;
    }

  /* Write */
  else if (condition & G_IO_OUT)
    {
      guint bytes_written;

      if (g_io_channel_write(iochannel, 
			     &state->buffer[state->n], 
			     state->length - state->n,
			     &bytes_written) != G_IO_ERROR_NONE)
	{
	  state->in_upcall = TRUE;
	  state->func(iochannel, state->buffer, state->length, state->n,
		      GNET_IOCHANNEL_WRITE_ASYNC_STATUS_ERROR, state->user_data);
	  state->in_upcall = FALSE;
	  write_async_cancel (state);
	  return FALSE;
	}

      state->n += bytes_written;

      if (state->n == state->length)
	{
	  state->in_upcall = TRUE; /* delay user cancelation */
	  state->func(iochannel, state->buffer, state->length, state->n,
		      GNET_IOCHANNEL_WRITE_ASYNC_STATUS_OK, state->user_data);
	  state->in_upcall = FALSE;
	  write_async_cancel (state);

	  return FALSE;
	}

      return TRUE;
    }

  return FALSE;
}


static gboolean 
write_async_timeout_cb (gpointer data)
{
  GNetIOChannelWriteAsyncState* state;

  state = (GNetIOChannelWriteAsyncState*) data;

  g_return_val_if_fail (state != NULL, FALSE);

  state->in_upcall = TRUE;
  state->func(state->iochannel, state->buffer, state->length, state->n,
	      GNET_IOCHANNEL_WRITE_ASYNC_STATUS_TIMEOUT, state->user_data);
  state->in_upcall = FALSE;
  write_async_cancel(state);

  return FALSE;
}


/* ************************************************************ */

typedef struct _GNetIOChannelReadAsyncState
{
  GIOChannel* iochannel;

  gboolean read_one;
  gboolean my_buffer;
  gchar* buffer;
  guint max_len;
  guint length;
  guint offset;
  guint timeout;

  guint watch;
  guint timer;

  GNetIOChannelReadAsyncCheckFunc check_func;
  gpointer check_user_data;

  GNetIOChannelReadAsyncFunc func;
  gpointer user_data;

  gboolean in_upcall;

} GNetIOChannelReadAsyncState;

static gboolean read_async_cb (GIOChannel* iochannel, GIOCondition condition, gpointer data);
static gboolean read_async_timeout_cb (gpointer data);



GNetIOChannelReadAsyncID
gnet_io_channel_read_async (GIOChannel* iochannel, 
			   gchar* buffer, guint length, guint timeout, 
			   gboolean read_one_byte_at_a_time, 
			   GNetIOChannelReadAsyncCheckFunc check_func, 
			   gpointer check_user_data,
			   GNetIOChannelReadAsyncFunc func, 	    
			   gpointer user_data)
{
  GNetIOChannelReadAsyncState* state;

  g_return_val_if_fail (iochannel, NULL);
  g_return_val_if_fail (check_func, NULL);
  g_return_val_if_fail (func, NULL);
  g_return_val_if_fail (buffer || (!buffer && length), NULL);

  state = g_new0 (GNetIOChannelReadAsyncState, 1);

  state->iochannel = iochannel;
  state->read_one = read_one_byte_at_a_time;

  if (buffer)
    {
      state->my_buffer = FALSE;
      state->buffer = buffer;
      state->max_len = length;
      state->length = length;
      state->offset = 0;
    }
  else
    {
      state->my_buffer = TRUE;
      state->buffer = NULL;
      state->max_len = length;
      state->length = 0;
      state->offset = 0;
    }

  state->check_func = 		check_func;
  state->check_user_data = 	check_user_data;
  state->func = 		func;
  state->user_data = 		user_data;

  state->watch = g_io_add_watch (iochannel, 
				 G_IO_IN | G_IO_ERR | G_IO_HUP | G_IO_NVAL, 
				 read_async_cb, state);

  state->timeout = timeout;
  if (timeout > 0)
    state->timer = g_timeout_add (timeout, read_async_timeout_cb, state);

  return state;
}


/* called by user, possibly during upcalls */
void
gnet_io_channel_read_async_cancel (GNetIOChannelReadAsyncID id)
{
  GNetIOChannelReadAsyncState* state;

  g_return_if_fail (id != NULL);

  state = (GNetIOChannelReadAsyncState*) id;

  if (state->in_upcall)
    return;

  g_source_remove (state->watch);
  if (state->timer)
    g_source_remove (state->timer);

  if (state->my_buffer)
    g_free(state->buffer);

  g_free(state);
}


/* used internally */
static void
read_async_cancel (GNetIOChannelReadAsyncState* state)
{
  g_source_remove (state->watch);
  if (state->timer)
    g_source_remove (state->timer);

  if (state->my_buffer)
    g_free(state->buffer);

  g_free(state);
}


static gboolean 
read_async_cb (GIOChannel* iochannel, GIOCondition condition, gpointer data)
{
  GNetIOChannelReadAsyncState* state;
  GIOError error;
  guint bytes_to_read;
  guint bytes_read;
  gint bytes_processed;

  state = (GNetIOChannelReadAsyncState*) data;

  g_return_val_if_fail (iochannel, FALSE);
  g_return_val_if_fail (state, FALSE);
  g_return_val_if_fail (iochannel == state->iochannel, FALSE);


  if (condition & (G_IO_ERR | G_IO_HUP | G_IO_NVAL))
    {
      state->in_upcall = TRUE;
      (state->func)(iochannel, GNET_IOCHANNEL_READ_ASYNC_STATUS_ERROR, 
		    NULL, 0, state->user_data);
      state->in_upcall = FALSE;
      read_async_cancel(state);

      return FALSE;
    }

  else if (condition & G_IO_IN)
    {
      /* Check if we should make the buffer larger */
      if (state->my_buffer && state->length == state->offset)
	{
	  if (state->length)
	    {
	      state->length *= 2;
	      state->buffer = g_realloc(state->buffer, state->length);
	    }
	  else
	    {
	      state->length = MIN (128, state->max_len);
	      state->buffer = g_malloc(state->length);
	    }
	}


      if (state->read_one)
	bytes_to_read = 1;
      else
	bytes_to_read = state->length - state->offset;

      /* Read in some stuff */
      error = g_io_channel_read (iochannel, &state->buffer[state->offset], 
				 bytes_to_read, &bytes_read);
      state->offset += bytes_read;

      if (error == G_IO_ERROR_AGAIN)
	return TRUE;

      else if (error != G_IO_ERROR_NONE)
	{
	  state->in_upcall = TRUE;
	  state->func (iochannel, GNET_IOCHANNEL_READ_ASYNC_STATUS_ERROR, 
		       NULL, 0, state->user_data);
	  state->in_upcall = FALSE;
	  read_async_cancel (state);
	  return FALSE;
	}

      /* If we read nothing, that means EOF and we're done.  Note we do
	 not send anything that might be in the buffer */
      else if (bytes_read == 0)
	{
	  state->in_upcall = TRUE;
	  state->func (iochannel, GNET_IOCHANNEL_READ_ASYNC_STATUS_OK, 
		       NULL, 0, state->user_data);
	  state->in_upcall = FALSE;
	  read_async_cancel (state);
	  return FALSE;
	}

      /* Check if we read something */
    again:
      bytes_processed = (state->check_func)(state->buffer, state->offset, state->check_user_data);
      if (bytes_processed)
	{
	  state->in_upcall = TRUE;
	  if (!state->func (iochannel, GNET_IOCHANNEL_READ_ASYNC_STATUS_OK, 
			    state->buffer, bytes_processed, state->user_data))
	    {
	      state->in_upcall = FALSE;
	      read_async_cancel (state);
	      return FALSE;
	    }
	  state->in_upcall = FALSE;

	  /* It is a programmer error if they called
	     g_io_channel_read_async_cancel() in the upcall and
	     returned TRUE (read more). */

	  /* Move it over */
	  g_memmove (state->buffer, &state->buffer[bytes_processed], 
		     state->offset - bytes_processed);
	  state->offset -= bytes_processed;

	  goto again;
	}

      /* Check if we hit the max length.  If so, it's an error. */
      if (state->offset >= state->max_len)
	{
	  state->in_upcall = TRUE;
	  state->func(iochannel, GNET_IOCHANNEL_READ_ASYNC_STATUS_ERROR, 
		      state->buffer, state->offset, state->user_data);
	  state->in_upcall = FALSE;
	  read_async_cancel (state);
	  return FALSE;
	}

      /* Reset the timer */
      if (state->timeout)
	{
	  if (state->timer)
	    g_source_remove (state->timer);

	  state->timer = g_timeout_add (state->timeout, 
					read_async_timeout_cb, 
					state);
	}

      return TRUE;
    }

  return FALSE;
}


static gboolean 
read_async_timeout_cb (gpointer data)
{
  GNetIOChannelReadAsyncState* state;

  state = (GNetIOChannelReadAsyncState*) data;

  g_return_val_if_fail (state, FALSE);

  state->in_upcall = TRUE;
  (state->func)(state->iochannel, GNET_IOCHANNEL_READ_ASYNC_STATUS_TIMEOUT, 
		NULL, 0, state->user_data);
  state->in_upcall = FALSE;
  read_async_cancel (state);

  return FALSE;
}


gint
gnet_io_channel_readany_check_func (gchar* buffer, guint length, 
				   gpointer user_data)
{
  return length;
}


gint
gnet_io_channel_readline_check_func (gchar* buffer, guint length, 
				     gpointer user_data)
{
  guint i;

  for (i = 0; i < length; ++i)
    {
      if (buffer[i] == '\n')
	return i + 1;
    }

  return 0;
}