void connection::handle_read(const asio::error_code& e,
									std::size_t bytes_transferred)
{
	if (!e)
	{
		if (bytes_transferred != size)
		{
			Log("Did not receive proper amount of bytes : %d", size);
			connection_manager_.stop(shared_from_this());
			return;
		}
		//printf("uid("XI64")\n", uid);
		// read object size
		if ((size > 8192*4) || (size <= 0))
		{
			//ERROR - object too large - close connection
			connection_manager_.stop(shared_from_this());
			return;
		}

		// parse packet
		request_.size = size;
		amf3parser * cparser = new amf3parser(buffer_.data());
		request_.object = cparser->ReadNextObject();
		request_.connection = this;
		try {
			request_handler_.handle_request(request_, reply_);
		}
		catch (std::exception& e)
		{
			std::cerr << "handle_request() exception: " << e.what() << "\n";
		}
		delete cparser;
		if (reply_.objects.size() > 0)
		{
			// send reply packets
			try {
				socket_.write_some(reply_.to_buffers());
			}
			catch (std::exception& e)
			{
				std::cerr << "asio::write_some() exception: " << e.what() << "\n";
			}
			reply_.objects.clear();
// 			asio::async_write(socket_, reply_.to_buffers(),
// 				boost::bind(&connection::handle_write, shared_from_this(),
// 				asio::placeholders::error));
		}

		asio::async_read(socket_, asio::buffer(buffer_, 4), boost::bind(&connection::handle_read_header, shared_from_this(),
			asio::placeholders::error,
			asio::placeholders::bytes_transferred));
	}
	else if (e != asio::error::operation_aborted)
	{
		connection_manager_.stop(shared_from_this());
		return;
	}
}

	void URLFetcherImpl::OnError(HttpRequestJob* job, const asio::error_code& err)
	{
		status_.set_status(URLRequestStatus::FAILED);
		status_.set_error(err.value());
		if (delegate_)
			delegate_->OnURLFetchComplete(this);
	}

void Connection::onConnectCompleted(const asio::error_code& err)
{
	UInt32 result = 1; 

	if ( !err )
	{
		result = 0;
	}

	EventRawConnected* e = new EventRawConnected;

	e->idx = GetUniqueId();
	e->result = result;
	e->socketIndex = GetSessionId();
	e->key = 0;

	m_net->Notify( EventPtr( e ) );

	if ( !err )
	{
		requestRecv();
	}
	else
	{
		// 에러 날 경우 연결 종료도 통보함
		onError( err );
	}

	MU2LogSystem( 0, "Connection::Error> [Code: %d]", err.value());
}

  void handle_read(const asio::error_code& ec)
  {
    if (stopped_)
      return;

    if (!ec)
    {
      // Extract the newline-delimited message from the buffer.
      std::string line;
      std::istream is(&input_buffer_);
      std::getline(is, line);

      // Empty messages are heartbeats and so ignored.
      if (!line.empty())
      {
        std::cout << "Received: " << line << "\n";
      }

      start_read();
    }
    else
    {
      std::cout << "Error on receive: " << ec.message() << "\n";

      stop();
    }
  }

void
TCPPeer::readBodyHandler(asio::error_code const& error,
                         std::size_t bytes_transferred)
{
    // LOG(DEBUG) << "TCPPeer::readBodyHandler "
    //     << "@" << mApp.getConfig().PEER_PORT
    //     << " to " << mRemoteListeningPort
    //     << (error ? "error " : "") << " bytes:" << bytes_transferred;

    if (!error)
    {
        LoadManager::PeerContext loadCtx(mApp, mPeerID);
        mByteRead.Mark(bytes_transferred);
        recvMessage();
        startRead();
    }
    else
    {
        if (isConnected())
        {
            // Only emit a warning if we have an error while connected;
            // errors during shutdown or connection are common/expected.
            mErrorRead.Mark();
            CLOG(ERROR, "Overlay")
                << "readBodyHandler error: " << error.message() << " :"
                << toString();
        }
        drop();
    }
}

  /// Handle completion of a read operation.
  void handle_read(const asio::error_code& e)
  {
    if (!e)
    {
      // Print out the data that was received.
      for (std::size_t i = 0; i < stocks_.size(); ++i)
      {
        std::cout << "Stock number " << i << "\n";
        std::cout << "  code: " << stocks_[i].code << "\n";
        std::cout << "  name: " << stocks_[i].name << "\n";
        std::cout << "  open_price: " << stocks_[i].open_price << "\n";
        std::cout << "  high_price: " << stocks_[i].high_price << "\n";
        std::cout << "  low_price: " << stocks_[i].low_price << "\n";
        std::cout << "  last_price: " << stocks_[i].last_price << "\n";
        std::cout << "  buy_price: " << stocks_[i].buy_price << "\n";
        std::cout << "  buy_quantity: " << stocks_[i].buy_quantity << "\n";
        std::cout << "  sell_price: " << stocks_[i].sell_price << "\n";
        std::cout << "  sell_quantity: " << stocks_[i].sell_quantity << "\n";
      }
    }
    else
    {
      // An error occurred.
      std::cerr << e.message() << std::endl;
    }

    // Since we are not starting a new operation the io_context will run out of
    // work to do and the client will exit.
  }

  /// Handle completion of a accept operation.
  void handle_accept(const asio::error_code& e, connection_ptr conn)
  {
    if (!e)
    {
      // Successfully accepted a new connection. Send the list of stocks to the
      // client. The connection::async_write() function will automatically
      // serialize the data structure for us.
      conn->async_write(stocks_,
          boost::bind(&server::handle_write, this,
            asio::placeholders::error, conn));

      // Start an accept operation for a new connection.
      connection_ptr new_conn(new connection(acceptor_.io_service()));
      acceptor_.async_accept(new_conn->socket(),
          boost::bind(&server::handle_accept, this,
            asio::placeholders::error, new_conn));
    }
    else
    {
      // An error occurred. Log it and return. Since we are not starting a new
      // accept operation the io_service will run out of work to do and the
      // server will exit.
      std::cerr << e.message() << std::endl;
    }
  }

// 更新处理结果
void DatabaseProxy::update_handle_result(asio::error_code error_code)
{
    if (error_code)
    {
        logger()->error("{}:{} {}", __FUNCTION__, __LINE__, error_code.message());
        return;
    }

    // 获取已完成任务
    assert(completion_lists_.empty());
    action_queue_.get_completed_tasks(completion_lists_);

    // 分发已完成任务
    network::NetMessage buffer;
    for (size_t i = 0; i < completion_lists_.size(); ++i)
    {
        buffer.clear();
        auto found = requests_.find(completion_lists_[i].get_sequence());
        assert(found != requests_.end());
        if (found != requests_.end())
        {
            send_handle_result(found->second.session_id, found->second.sequence, completion_lists_[i], buffer);
        }
        generator_.put(completion_lists_[i].get_sequence());
        requests_.erase(found);
    }

    completion_lists_.clear();
    timer_.async_wait(wait_handler_);
}

void Binlog_tcp_driver::handle_net_packet_header(const asio::error_code& err, std::size_t bytes_transferred)
{
  if (err)
  {
    Binary_log_event * ev= create_incident_event(175, err.message().c_str(), m_binlog_offset);
    std::cout << "3:" << err.message() << std::endl;
    m_event_queue->push_front(ev);
    return;
  }

  if (bytes_transferred != 4)
  {
    std::ostringstream os;
    os << "Expected byte size to be between 0 and "
       << MAX_PACKAGE_SIZE
       << " number of bytes; got "
       << bytes_transferred
       << " instead.";
    Binary_log_event * ev= create_incident_event(175, os.str().c_str(), m_binlog_offset);
    std::cout << "4:" << os.str() << std::endl;
    m_event_queue->push_front(ev);
    return;
  }

  int packet_length=(unsigned long) (m_net_header[0] &0xFF);
  packet_length+=(unsigned long) ((m_net_header[1] &0xFF) << 8);
  packet_length+=(unsigned long) ((m_net_header[2] &0xFF) << 16);

  // TODO validate packet sequence numbers
  //int packet_no=(unsigned char) m_net_header[3];

  if (m_waiting_event == 0)
  {
    //std::cerr << "event_stream_buffer.size= " << m_event_stream_buffer.size() << std::endl;
    m_waiting_event= new Log_event_header();
    m_event_packet=  asio::buffer_cast<char *>(m_event_stream_buffer.prepare(packet_length));
    //assert(m_event_stream_buffer.size() == 0);
  }


  Read_handler read_handler;
  read_handler.method     = &Binlog_tcp_driver::handle_net_packet;
  read_handler.tcp_driver = this;
  asio::async_read(*m_socket,
                          asio::buffer(m_event_packet, packet_length),
                          read_handler);
}

void GstVideoServer::socketError(  const asio::error_code &error, uint64_t identifier )
{
    auto clientEntry = mConnectedClients.find( identifier );
    if(clientEntry != mConnectedClients.end() ){
        auto& client = clientEntry->second;
		auto address = client.getRemoteEndpoint().address().to_string();
		CI_LOG_E("Socket Error for Client with ip address " << address <<". Error:  " << error.message() );
		client.shutdown();
        client.close();
		mConnectedAdressedClients.erase(address);
        mConnectedClients.erase(clientEntry);
        mOscReceiver->closeConnection(identifier);
    }else{
		CI_LOG_E( "Couldn't find client for id: " << identifier << " and errror: " << error.message() );
    }

}

void InnerTcpConnection::handleSendMessage(const asio::error_code& error) {
  if(unlikely(error)) {
    LOG(ERROR) << "Failed to send message, error: " << error << '(' << error.message() << ')';
    terminate();
  } {
    DVLOG(3) << "Send message successfully";
    state.store(READY);
    realSendMessage();
  }
}

void Connection::onError(const asio::error_code& err)
{
	MU2LogSystem( 0, "Connection::onError> %s", err.message().c_str() );

	EventDisconnected* e = new EventDisconnected;
	e->idx = GetUniqueId();
	e->socketIndex = GetSessionId();
	e->sessionType = 0;

	m_net->Notify( EventPtr( e ) );
}

 void handle_receive_from(const asio::error_code& err, size_t length)
 {
   if (err)
   {
     std::cout << "Receive error: " << err.message() << "\n";
   }
   else
   {
     std::cout << "Successful receive\n";
   }
 }

void TcpSession::onClose( const asio::error_code& err )
{
	if ( err ) {
		if ( mErrorEventHandler != nullptr ) {
			mErrorEventHandler( err.message(), 0 );
		}
	} else {
		if ( mCloseEventHandler != nullptr ) {
			mCloseEventHandler();
		}
	}
}

void RealmConnection::_complete(const asio::error_code& e, std::size_t bytes_transferred, PacketPtr packet_ptr)
{
	UT_DEBUGMSG(("RealmConnection::_complete()\n"));
	if (e)
	{
		UT_DEBUGMSG(("Error reading message: %s\n", e.message().c_str()));
		_disconnect();		
		return;
	}	
	m_buf.commit(bytes_transferred);
	_complete_packet(packet_ptr);
}

void TcpClient::onConnect( TcpSessionRef session, const asio::error_code& err )
{
	if ( err ) {
		if ( mErrorEventHandler != nullptr ) {
			mErrorEventHandler( err.message(), 0 );
		}
	} else {
		if ( mConnectEventHandler != nullptr ) {
			mConnectEventHandler( session );
		}
	}
}

 void handle_connect(const asio::error_code& error)
 {
   if (!error)
   {
     socket_.async_handshake(asio::ssl::stream_base::client,
         boost::bind(&client::handle_handshake, this,
           asio::placeholders::error));
   }
   else
   {
     std::cout << "Connect failed: " << error.message() << "\n";
   }
 }

void TcpServer::onAccept( TcpSessionRef session, const asio::error_code& err )
{
	if ( err ) {
		if ( mErrorEventHandler != nullptr ) {
			mErrorEventHandler( err.message(), 0 );
		}
	} else {
		if ( mAcceptEventHandler != nullptr ) {
			mAcceptEventHandler( session );
		}
		listen();
	}
}

void UdpClient::onConnect( UdpSessionRef session, const asio::error_code& err )
{
	if ( err ) {
		if ( mErrorEventHandler != nullptr ) {
			mErrorEventHandler( err.message(), 0 );
		}
	} else {
		if ( mConnectEventHandler != nullptr ) {
			session->mSocket->set_option( asio::socket_base::reuse_address( true ) );
			mConnectEventHandler( session );
		}
	}
}

// 更新定时器
void LinkerManager::OnUpdateTimer(asio::error_code error_code)
{
	if (error_code)
	{
		logger()->error("{}:{} {}", __FUNCTION__, __LINE__, error_code.message());
		return;
	}

	// 删除超时Token
	for (auto iter = user_auth_.begin(); iter != user_auth_.end();)
	{
		if (std::chrono::steady_clock::now() - iter->second.time >= std::chrono::seconds(60))
		{
			logger()->debug("删除超时的Token {}", iter->first);
			iter = user_auth_.erase(iter);
		}
		{
			++iter;
		}
	}

	// 关闭长时间未验证的连接
	for (auto iter = unauth_user_session_.begin(); iter != unauth_user_session_.end();)
	{
		auto session = static_cast<SessionHandle*>(threads_.SessionHandler(*iter).get());
		if (session != nullptr && session->IsAuthTimeout())
		{
			logger()->debug("关闭长时间未验证连接，{}:{}", session->RemoteEndpoint().address().to_string(), session->RemoteEndpoint().port());
			session->Close();
			iter = unauth_user_session_.erase(iter);
		}
		else
		{
			++iter;
		}
	}

	// 上报Linker在线人数
	if (counter_ > 0 && --counter_ == 0)
	{
		svr::ReportLinkerReq request;
		request.set_load(user_session_.size());
		GlobalLoginConnector()->Send(&request);
		counter_ = ServerConfig::GetInstance()->GetReportInterval();
		logger()->info("上报当前在线人数: {}", user_session_.size());
	}

	timer_.expires_from_now(std::chrono::seconds(1));
	timer_.async_wait(wait_handler_);
}