text: #include <muduo/net/TcpConnection.h>
  
  #include <muduo/base/Logging.h>
  #include <muduo/base/WeakCallback.h>
  #include <muduo/net/Channel.h>
  #include <muduo/net/EventLoop.h>
  #include <muduo/net/Socket.h>
  #include <muduo/net/SocketsOps.h>
  
  #include <boost/bind.hpp>
  
  #include <errno.h>
  
  using namespace muduo;
  using namespace muduo::net;
  
  void muduo::net::defaultConnectionCallback(const TcpConnectionPtr &conn)
  {
  	LOG_TRACE << conn->localAddress().toIpPort() << " -> "
  		<< conn->peerAddress().toIpPort() << " is " 
  		<< (conn->connected() ? "UP" : "DOWN");
  }
  
  void muduo::net::defaultMessageCallback(const TcpConnectionPtr &conn, Buffer *buf, Timestamp)
  {
  	buf->retrieveAll();
  }
  
  TcpConnection::TcpConnection(EventLoop *loop, 
  		const string &nameArg, 
  		int sockfd, 
  		const InetAddress &localAddr, 
  		const InetAddress &peerAddr)
  	:loop_(CHECK_NOTNULL(loop)),
  	name_(nameArg), 
  	state_(kConnecting),
  	reading_(true), 
  	socket_(new Socket(sockfd)), 
  	channel_(new Channel(loop, sockfd)), 
  	localAddr_(localAddr), 
  	peerAddr_(peerAddr), 
  	highWaterMark_(64*1024*1024) // 64M
  {
  	channel_->setReadCallback(boost::bind(&TcpConnection::handleRead, this, _1));
  	channel_->setWriteCallback(boost::bind(&TcpConnection::handleWrite, this));
  	channel_->setCloseCallback(boost::bind(&TcpConnection::handleClose, this));
  	channel_->setErrorCallback(boost::bind(&TcpConnection::handleError, this));
  	LOG_DEBUG << "TcpConnection::ctor[" << name_ << "] at " << this  << " fd=" << sockfd;
  	socket_->setKeepAlive(true);
  }
  
  TcpConnection::~TcpConnection()
  {
  	LOG_DEBUG << "TcpConnection::dtor[" << name_ << "] at " << this << " fd=" << channel_->fd() << " state=" << stateToString();
  	assert(state_ == kDisconnected);
  }
  
  bool TcpConnection::getTcpInfo(struct tcp_info *tcpi) const
  {
  	return socket_->getTcpInfo(tcpi);
  }
  
  string TcpConnection::getTcpInfoString() const
  {
  	char buf[1024];
  	buf[0] = '\0';
  	socket_->getTcpInfoString(buf, sizeof(buf));
  	return buf;
  }
  
  void TcpConnection::send(const void *data, int len)
  {
  	send(StringPiece(static_cast<const char *>(data), len));
  }
  
  void TcpConnection::send(const StringPiece &message)
  {
  	if (state_ == kConnected)
  	{
  		if (loop_->isInLoopThread())
  		{
  			sendInLoop(message);
  		}
  		else
  		{
  			loop_->runInLoop(boost::bind(&TcpConnection::sendInLoop, this, message.as_string()));
  		}
  	}
  }
  
  void TcpConnection::send(Buffer *buf)
  {
  	if (state_ == kConnected)
  	{
  		if (loop_->inInLoopThread())
  		{
  			sendInLoop(buf->peek(), buf->readableBytes());
  			buf->retrieveAll();
  		}
  		else
  		{
  			loop_->runInLoop(boost::bind(&TcpConnection::sendInLoop, this, buf->retrieveAllAsString()));
  		}
  	}
  }
  
  void TcpConnection::sendInLoop(const StringPiece &message)
  {
  	sendInLoop(message.data(), message.size());
  }
  
  void TcpConnection::sendInLoop(const void *data, size_t len)
  {
  	loop_->assertInLoopThread();
  	ssize_t nwrote = 0;
  	size_t remaining = len;
  	bool faultError = false;
  	if (state_ == kDisconnected)
  	{
  		LOG_WARN << "disconnected, give up writing";
  		return ;
  	}
  
  	// if no thing in output queue, try writing directly
  	if (!channel_->isWriting() && outputBuffer_.readableBytes() == 0)
  	{
  		nwrote = sockets::write(channel_->fd(), data, len);
  		if (nwrote >= 0)
  		{
  			remaining = len - nwrote;
  			if (remaining == 0 && writeCompleteCallback_)
  			{
  				loop_->queueInLoop(boost::bind(writeCompleteCallback_, shared_from_this()));
  			}
  		}
  		else
  		{
  			nwrote = 0;
  			if (errno != EWOULDBLOCK)
  			{
  				LOG_SYSERR << "TcpConnection::sendInLoop";
  				if (errno == EPIPE || errno == ECONNRESET)
  				{
  					faultError = true;
  				}
  			}
  		}
  
  		assert(remaining <= len);
  		if (!faultError && remaining > 0)
  		{
  			size_t oldLen = outputBuffer_.readableBytes();
  			if (oldLen + remaining >= highWaterMark_ && 
  					oldLen < highWaterMark_ && 
  					highWaterMarkCallback_)
  			{
  				loop_->ququeInLoop(boost::bind(highWaterMarkCallback_, shared_from_this(), oldLen + remaining));
  			}
  			outputBuffer_.append(static_cast<const char *>(data) + nwrote, remaining);
  			if (!channel_->isWriteing())
  			{
  				channel_->enableWriting();
  			}
  		}
  }
  
  void TcpConnection::shudown()
  {
  	if (state_ == kConnected)
  	{
  		setState(kDisconnecting);
  		loop_->runInLoop(boost::bind(&TcpConnection::shutdownInLoop, this));
  	}
  }
  
  void TcpConnection::shutDownInLoop()
  {
  	loop_->assertInLoopThread();
  	if (!Channle_->isWriting())
  	{
  		// we are not writing 
  		socket_->shutdownWrite();
  	}
  }
  
  void TcpConnection::forceClose()
  {
  	if (state_ == kConnected || state_ == kDisconnecting)
  	{
  		setState(kDisconnecting);
  		loop_->queueInLoop(boost::bind(&TcpConnection::forceCloseInLoop, shared_from_this()));
  	}
  }
  
  void TcpConnection::forceCloseWithDelay(double seconds)
  {
  	if (state_ == kConnected || state_ == kDisconnecting)
  	{
  		setState(kDisconnecting);
  		loop_->runAfter(seconds, makeWeakCallBack(shared_from_this(), &TcpConnection::forceClose));
  	}
  }
  
  void TcpConnection::forceCloseInLoop()
  {
  	loop_->assertInLoopThread();
  	if (state_ == kConnected || state_ == kDisconnecting)
  	{
  		handleClose();
  	}
  }
  
  const char* TcpConnection::stateToString() const
  {
  	switch(state_)
  	{
  		case kDisconnected:
  			return "kDisconnected";
  		case kConnecting:
  			return "kConnecting";
  		case kConnected:
  			return "kConnected";
  		case kDisconnecting:
  			return "kDisconnecting";
  		defalut:
  			return "unknown state";
  	}
  }
  
  void TcpConnection::setTcpNoDelay(bool on)
  {
  	socket_->setTcpNoDelay(on);
  }
  
  void TcpConnection::startRead()
  {
  	loop_->runInLoop(boost::bind(&TcpConnection::startReadInLoop, this));
  }
  
  void TcpConnection::startReadInLoop()
  {
  	loop_->assertInLoopThread();
  	if (!reading_ || !channel_->isReading())
  	{
  		channel_->enableReading();
  		reading_ = true;
  	}
  }
  
  void TcpConnection::stopRead()
  {
  	loop_->runInLoop(boost::bind(&TcpConnection::stopReadInLoop, this));
  }
  
  void TcpConnection::stopReadInLoop()
  {
  	loop_->assertInLoopThread();
  	if (reading_ || channel_->isReading())
  	{
  		channel_->disableReading();
  		reading_ = false;
  	}
  }
  
  void TcpConnection::connectEstablished()
  {
  	loop_->assertInLoopThread();
  	assert(state_ == kConnecting);
  	setState(kConnected);
  	channel_->tie(shared_from_this());
  	channel_->enableReading();
  
  	connectionCallback_(shared_from_this());
  }
  
  void TcpConnection::connetionDestroyed()
  {
  	loop_->assertInLoopThread();
  	if (state_ == kConnected)
  	{
  		setState(kDisconnected);
  		channel_->disableAdd();
  		connectionCallback_(shared_from_this());
  	}
  	channel_->remove();
  }
  
  void TcpConnection::handleRead(Timestamp receiveTime)
  {
  	loop_->assertInLoopThread();
  	int saveErrno = 0;
  	ssize_t n = inputBuffer_.readFd(channel_->fd(), &savedErrno);
  	if (n > 0)
  	{
  		messageCallback_(shared_from_this(), &inputBuffer_, receiveTime);
  	}
  	else if (n == 0)
  	{
  		handleClose();
  	}
  	else
  	{
  		errno = savedErrno;
  		LOG_SYSERR << "TcpConnection::handleRead";
  		handleError();
  	}
  }
  
  void TcpConnection::handleWrite()
  {
  	loop_->assertInLoopThread();
  	if (channel_->isWriting())
  	{
  		ssize_t n = sockets::write(channel_->fd(), 
  				outputBuffer_.peek(), 
  				outputBuffer_.readableBytes());
  
  		if (n > 0)
  		{
  			outputBuffer_retrieve(n);
  			if (outputBuffer_.readableBytes() == 0)
  			{
  				channel_->disableWriting();
  				if (writeCompleteCallback_)
  				{
  					loop_->queueInLoop(boost::bind(writeCompleteCallback_, shared_from_this()));
  				}
  				if (state_ == kDisconnecting)
  				{
  					shutdownInLoop();
  				}
  			}
  		}
  		else
  		{
  			LOG_SYSERR << "TcpConnection::handleWrite";
  		}
  	}
  	else
  	{
  		LOG_TRACE << "Connection fd = " << channel_->fd() 
  			<< " is down, no more writing";
  	}
  }
  
  void TcpConnection::handleClose()
  {
  	loop_->assertInLoopThread();
  	LOG_TRACE << "fd = " << channel_->fd() << " state = " << stateToString();
  	assert(state_ == kConnected || state_ == kDisconnecting);
  
  	// we don't close fd, leave it to dtor, so we can find leaks easyily
  	setState(kDisconnected);
  	channel_->disableAll();
  
  	TcpConnectionPtr guardThis(shared_from_this());
  	connectionCallback_(guardThis);
  
  	// must be the last line
  	closeCallback_(guardThis);
  }
  
  void TcpConnection::handleError()
  {
  	int err = sockets::getSocketError(channel_->fd());
  	LOG_ERROR << "TcpConnection::handleError [" << name_ 
  		<< "] - SO_ERROR = " << err << " " << strerror_tl(err);
  }
  