void protocol::load_hosts(const std::error_code& ec,
    completion_handler handle_complete)
{
    if (ec)
    {
        log_error(LOG_PROTOCOL)
            << "Could not load hosts file: " << ec.message();
        handle_complete(ec);
        return;
    }
    hosts_.fetch_count(strand_.wrap(
        &protocol::if_0_seed, this, _1, _2, handle_complete));
}

void protocol::handle_listen(const std::error_code& ec, acceptor_ptr accept)
{
    if (ec)
    {
        log_error(LOG_PROTOCOL)
            << "Error while listening: " << ec.message();
    }
    else
    {
        accept->accept(strand_.wrap(
            &protocol::handle_accept, this, _1, _2, accept));
    }
}

void stealth_fetched(const std::error_code& ec,
    const blockchain::stealth_list& stealth_results)
{
    if (ec)
    {
        log_error() << "Stealth fetch failed: " << ec.message();
        return;
    }
    for (const blockchain::stealth_row& row: stealth_results)
        log_debug() << "ephemkey: " << row.ephemkey
            << " address: " << row.address.encoded()
            << " tx_hash: " << row.transaction_hash;
}

void
state_machine_t::terminate(std::error_code ec) {
    BOOST_ASSERT(ec);

    if (closed.exchange(true)) {
        return;
    }

    COCAINE_LOG_DEBUG(log, "slave state machine is terminating: %s", ec.message());

    auto state = *this->state.synchronize();
    state->terminate(ec);
}

void protocol::handle_save(const std::error_code& ec,
    completion_handler handle_complete)
{
    if (ec)
    {
        log_error(LOG_PROTOCOL) << "Failed to save hosts '"
            << hosts_filename_ << "': " << ec.message();
        handle_complete(ec);
        return;
    }
    channel_subscribe_->relay(error::service_stopped, nullptr);
    handle_complete(std::error_code());
}

void connection_started(const std::error_code& ec, channel_ptr node,
    protocol& prot)
{
    if (ec)
    {
        log_warning() << "Couldn't start connection: " << ec.message();
        return;
    }
    log_info() << "Connection established.";
    // Resubscribe to new nodes.
    prot.subscribe_channel(
        std::bind(connection_started, _1, _2, std::ref(prot)));
}

bool GlxContext::makeNotCurrentImpl(std::error_code& ec)
{
	ec.clear();

    if(!::glXMakeCurrent(xDisplay(), 0, nullptr))
    {
		//TODO: handle error into ec
		warning("ny::GlxContext::makeNotCurrentImpl (glXMakeCurrent) failed");
        return false;
    }

    return true;
}

void recv_block(std::error_code ec, libbitcoin::message::block block)
{
    if (ec)
    {
        std::cerr << ec.message() << "\n";
        return;
    }
    BITCOIN_ASSERT(block.transactions.size() == 2);
    libbitcoin::hash_digest h1 = libbitcoin::hash_transaction(block.transactions[0]),
            h2 = libbitcoin::hash_transaction(block.transactions[1]);
    libbitcoin::hash_digest merkle = libbitcoin::generate_merkle_root(block.transactions);
    BITCOIN_ASSERT((merkle == libbitcoin::hash_digest{0x7d, 0xac, 0x2c, 0x56, 0x66, 0x81, 0x5c, 0x17, 0xa3, 0xb3, 0x64, 0x27, 0xde, 0x37, 0xbb, 0x9d, 0x2e, 0x2c, 0x5c, 0xce, 0xc3, 0xf8, 0x63, 0x3e, 0xb9, 0x1a, 0x42, 0x05, 0xcb, 0x4c, 0x10, 0xff}));
}

void protocol::inbound_channel_stopped(
    const std::error_code& ec, channel_ptr which_node)
{
    // We must always attempt a reconnection if this was an
    // outbound connection.
    if (ec) 
    {
        log_error(LOG_PROTOCOL)
            << "Channel stopped (inbound): " << ec.message();
    }
    // Remove from accepted connections.
    remove_channel(accepted_channels_, which_node);
}

bool callback_state::succeeded(const std::error_code& code, 
    const std::string& format)
{
    if (code)
    {
        // May want to change the behavior to decrement vs. zeroizing refs.
        error(boost::format(format) % code.message());
        stop(console_result::failure);
        return false;
    }

    return true;
}

bool GlxContext::makeCurrentImpl(const GlSurface& surface, std::error_code& ec)
{
	ec.clear();

	auto drawable = dynamic_cast<const GlxSurface*>(&surface)->xDrawable();
    if(!::glXMakeCurrent(xDisplay(), drawable, glxContext_))
    {
		//TODO: handle error into ec
		warning("ny::GlxContext::makeCurrentImpl (glXMakeCurrent) failed");
        return false;
    }

    return true;
}

 void net_tcp_connection_base::handle_received_pkg(const std::error_code &error, size_t bytes_transferred) {
     if (!error) {
         m_pEH->triger<tcp_recv_stream_succ>(this, bytes_transferred);
         m_oRecvBuffer.filled() += bytes_transferred;
         LOG(INFO) << "recv pkg: " << bytes_transferred << " bytes, from " << m_oRemoteEndpoint.address().to_string();
         slice_and_dispatch_pkg();
         start_recv();
     } else {
         m_iPointState = state_error;
         LOG(WARNING) << "handle_received_pkg(), Get error " << error.message() << " from " <<
                               m_oRemoteEndpoint.address().to_string();
         m_pEH->triger<tcp_recv_stream_error>(this, error);
     }
 }

void fullnode::connection_started(const std::error_code& ec, channel_ptr node)
{
    if (ec)
    {
        log_warning() << "Couldn't start connection: " << ec.message();
        return;
    }
    // Subscribe to transaction messages from this node.
    node->subscribe_transaction(
        std::bind(&fullnode::recv_tx, this, _1, _2, node));
    // Stay subscribed to new connections.
    protocol_.subscribe_channel(
        std::bind(&fullnode::connection_started, this, _1, _2));
}

void json_history_fetched(const payment_address& payaddr,
    const std::error_code& ec, const blockchain::history_list& history)
{
    if (ec)
    {
        std::cerr << "history: Failed to fetch history: "
            << ec.message() << std::endl;
        return;
    }
    bool is_first = true;
    for (const auto& row: history)
    {
        // Put commas between each array item in json output.
        if (is_first)
            is_first = false;
        else
            std::cout << "," << std::endl;
        // Actual row data.
        std::cout << "{" << std::endl;
        std::cout << "  \"address\": \"" << payaddr.encoded()
            << "\"," << std::endl;
        std::cout << "  \"output\": \"" << row.output
            << "\"," << std::endl;
        std::cout << "  \"output_height\": ";
        if (!row.output_height)
            std::cout << "\"Pending\"";
        else
            std::cout << row.output_height;
        std::cout << "," << std::endl;
        std::cout << "  \"value\":  \"" << row.value << "\"," << std::endl;
        if (row.spend.hash == null_hash)
        {
            std::cout << "  \"spend\": \"Unspent\"," << std::endl;
            std::cout << "  \"spend_height\": \"Unspent\"" << std::endl;
        }
        else
        {
            std::cout << "  \"spend\": \"" << row.spend << "\"," << std::endl;
            std::cout << "  \"spend_height\": ";
            if (!row.spend_height)
                std::cout << "\"Pending\"";
            else
                std::cout << "\"" << row.spend_height << "\"";
        }
        std::cout << "}";
    }
    if (--remaining_count > 0)
        std::cout << ",";
    std::cout << std::endl;
}

void server_node::monitor_tx(const std::error_code& ec, network::channel_ptr node)
{
    if (ec)
    {
        log_warning(LOG_NODE) << "Couldn't start connection: " << ec.message();
        return;
    }
    // Subscribe to transaction messages from this node.
    node->subscribe_transaction(
        std::bind(&server_node::recv_transaction, this, _1, _2, node));
    // Stay subscribed to new connections.
    protocol_.subscribe_channel(
        std::bind(&server_node::monitor_tx, this, _1, _2));
}

void block_header_fetched(const std::error_code& ec,
    const block_header_type& blk_header)
{
    if (ec)
    {
        std::cerr << "fetch-block-header: " << ec.message() << std::endl;
        stopped = true;
        return;
    }
    data_chunk raw_blk_header(satoshi_raw_size(blk_header));
    satoshi_save(blk_header, raw_blk_header.begin());
    std::cout << raw_blk_header << std::endl;
    stopped = true;
}

void ask_blocks(const std::error_code& ec, channel_ptr node,
    const message::block_locator& loc, blockchain_ptr chain,
    const hash_digest& hash_stop)
{
    if (ec)
        log_error() << ec.message();
    node->subscribe_inventory(std::bind(recv_inv, _1, _2, node));
    node->subscribe_block(std::bind(recv_blk, _1, _2, node, chain));

    message::get_blocks packet;
    packet.start_hashes = loc;
    packet.hash_stop = hash_stop;
    node->send(packet, std::bind(&handle_send_packet, _1));
}

void resume_copy(const std::error_code& ec, size_t last_depth,
    blockchain* chain_1, blockchain* chain_2)
{
    size_t resume_depth = last_depth + 1;
    if (ec == error::not_found)
        resume_depth = 0;
    else if (ec)
    {
        log_error() << "Error fetch last depth from DEST: " << ec.message();
        return;
    }
    fetch_block(*chain_1, resume_depth,
        std::bind(copy_block, _1, _2, resume_depth, chain_1, chain_2));
}

void session::get_blocks(const std::error_code& ec,
    const get_blocks_type& packet, channel_ptr node)
{
    if (ec)
    {
        log_error(LOG_SESSION) << "get_blocks: " << ec.message();
        return;
    }
    // send 500 invs from last fork point
    // have memory of last inv, ready to trigger send next 500 once
    // getdata done for it.
    node->subscribe_get_blocks(
        std::bind(&session::get_blocks, this, _1, _2, node));
}

    void
    finalize(const std::error_code& ec) {
        // Prepare the internal socket object for consequential operations by moving its contents
        // to a heap-allocated object, which in turn might be attached to an engine.
        auto ptr = std::make_unique<socket_type>(std::move(socket));

        switch(ec.value()) {
        case 0:
            COCAINE_LOG_DEBUG(parent.m_log, "accepted connection on fd {:d}", ptr->native_handle());
            parent.metrics->connections_accepted->fetch_add(1);

            try {
                parent.m_context.engine().attach(std::move(ptr), parent.m_prototype);
            } catch(const std::system_error& e) {
                COCAINE_LOG_ERROR(parent.m_log, "unable to attach connection to engine: {}",
                    error::to_string(e));
                ptr = nullptr;
            }

            break;

        case asio::error::operation_aborted:
            return;

        default:
            COCAINE_LOG_ERROR(parent.m_log, "unable to accept connection: [{:d}] {}", ec.value(),
                ec.message());
            parent.metrics->connections_rejected->fetch_add(1);
            break;
        }

        // TODO: Find out if it's always a good idea to continue accepting connections no matter
        // what.
        // For example, destroying a socket from outside this thread will trigger weird stuff on
        // Linux.
        operator()();
    }