bgp_read (struct thread *thread)
{
  int ret;
  u_char type = 0;
  struct peer *peer;
  bgp_size_t size;
  char notify_data_length[2];

  /* Yes first of all get peer pointer. */
  peer = THREAD_ARG (thread);
  peer->t_read = NULL;

  /* For non-blocking IO check. */
  if (peer->status == Connect)
    {
      bgp_connect_check (peer);
      goto done;
    }
  else
    {
      if (peer->fd < 0)
	{
	  zlog_err ("bgp_read peer's fd is negative value %d", peer->fd);
	  return -1;
	}
      BGP_READ_ON (peer->t_read, bgp_read, peer->fd);
    }

  /* Read packet header to determine type of the packet */
  if (peer->packet_size == 0)
    peer->packet_size = BGP_HEADER_SIZE;

  if (stream_get_endp (peer->ibuf) < BGP_HEADER_SIZE)
    {
      ret = bgp_read_packet (peer);

      /* Header read error or partial read packet. */
      if (ret < 0) 
	goto done;

      /* Get size and type. */
      stream_forward_getp (peer->ibuf, BGP_MARKER_SIZE);
      memcpy (notify_data_length, stream_pnt (peer->ibuf), 2);
      size = stream_getw (peer->ibuf);
      type = stream_getc (peer->ibuf);

      if (BGP_DEBUG (normal, NORMAL) && type != 2 && type != 0)
	zlog_debug ("%s rcv message type %d, length (excl. header) %d",
		   peer->host, type, size - BGP_HEADER_SIZE);

      /* Marker check */
      if (((type == BGP_MSG_OPEN) || (type == BGP_MSG_KEEPALIVE))
	  && ! bgp_marker_all_one (peer->ibuf, BGP_MARKER_SIZE))
	{
	  bgp_notify_send (peer,
			   BGP_NOTIFY_HEADER_ERR, 
			   BGP_NOTIFY_HEADER_NOT_SYNC);
	  goto done;
	}

      /* BGP type check. */
      if (type != BGP_MSG_OPEN && type != BGP_MSG_UPDATE 
	  && type != BGP_MSG_NOTIFY && type != BGP_MSG_KEEPALIVE 
	  && type != BGP_MSG_ROUTE_REFRESH_NEW
	  && type != BGP_MSG_ROUTE_REFRESH_OLD
	  && type != BGP_MSG_CAPABILITY)
	{
	  if (BGP_DEBUG (normal, NORMAL))
	    plog_debug (peer->log,
		      "%s unknown message type 0x%02x",
		      peer->host, type);
	  bgp_notify_send_with_data (peer,
				     BGP_NOTIFY_HEADER_ERR,
			 	     BGP_NOTIFY_HEADER_BAD_MESTYPE,
				     &type, 1);
	  goto done;
	}
      /* Mimimum packet length check. */
      if ((size < BGP_HEADER_SIZE)
	  || (size > BGP_MAX_PACKET_SIZE)
	  || (type == BGP_MSG_OPEN && size < BGP_MSG_OPEN_MIN_SIZE)
	  || (type == BGP_MSG_UPDATE && size < BGP_MSG_UPDATE_MIN_SIZE)
	  || (type == BGP_MSG_NOTIFY && size < BGP_MSG_NOTIFY_MIN_SIZE)
	  || (type == BGP_MSG_KEEPALIVE && size != BGP_MSG_KEEPALIVE_MIN_SIZE)
	  || (type == BGP_MSG_ROUTE_REFRESH_NEW && size < BGP_MSG_ROUTE_REFRESH_MIN_SIZE)
	  || (type == BGP_MSG_ROUTE_REFRESH_OLD && size < BGP_MSG_ROUTE_REFRESH_MIN_SIZE)
	  || (type == BGP_MSG_CAPABILITY && size < BGP_MSG_CAPABILITY_MIN_SIZE))
	{
	  if (BGP_DEBUG (normal, NORMAL))
	    plog_debug (peer->log,
		      "%s bad message length - %d for %s",
		      peer->host, size, 
		      type == 128 ? "ROUTE-REFRESH" :
		      bgp_type_str[(int) type]);
	  bgp_notify_send_with_data (peer,
				     BGP_NOTIFY_HEADER_ERR,
			  	     BGP_NOTIFY_HEADER_BAD_MESLEN,
				     (u_char *) notify_data_length, 2);
	  goto done;
	}

      /* Adjust size to message length. */
      peer->packet_size = size;
    }

  ret = bgp_read_packet (peer);
  if (ret < 0) 
    goto done;

  /* Get size and type again. */
  size = stream_getw_from (peer->ibuf, BGP_MARKER_SIZE);
  type = stream_getc_from (peer->ibuf, BGP_MARKER_SIZE + 2);

  /* BGP packet dump function. */
  bgp_dump_packet (peer, type, peer->ibuf);
  
  size = (peer->packet_size - BGP_HEADER_SIZE);

  /* Read rest of the packet and call each sort of packet routine */
  switch (type) 
    {
    case BGP_MSG_OPEN:
      peer->open_in++;
      bgp_open_receive (peer, size); /* XXX return value ignored! */
      break;
    case BGP_MSG_UPDATE:
      peer->readtime = time(NULL);    /* Last read timer reset */
      bgp_update_receive (peer, size);
      break;
    case BGP_MSG_NOTIFY:
      bgp_notify_receive (peer, size);
      break;
    case BGP_MSG_KEEPALIVE:
      peer->readtime = time(NULL);    /* Last read timer reset */
      bgp_keepalive_receive (peer, size);
      break;
    case BGP_MSG_ROUTE_REFRESH_NEW:
    case BGP_MSG_ROUTE_REFRESH_OLD:
      peer->refresh_in++;
      bgp_route_refresh_receive (peer, size);
      break;
    case BGP_MSG_CAPABILITY:
      peer->dynamic_cap_in++;
      bgp_capability_receive (peer, size);
      break;
    }

  /* Clear input buffer. */
  peer->packet_size = 0;
  if (peer->ibuf)
    stream_reset (peer->ibuf);

 done:
  if (CHECK_FLAG (peer->sflags, PEER_STATUS_ACCEPT_PEER))
    {
      if (BGP_DEBUG (events, EVENTS))
	zlog_debug ("%s [Event] Accepting BGP peer delete", peer->host);
      peer_delete (peer);
    }
  return 0;
}