void handshake::receive_version(
			const std::error_code& ec, const version_type&,
			channel_ptr node, handshake::handshake_handler completion_callback)
		{
			if (ec)
				completion_callback(ec);
			else
				node->send(verack_type(),
				strand_.wrap(std::bind(&handshake::handle_message_sent,
				this, _1, completion_callback)));
		}

void protocol::seeds::request_addresses(
    const std::error_code& ec, channel_ptr dns_seed_node)
{
    if (ec)
    {
        log_error(LOG_PROTOCOL)
            << "Failed to connect to seed node: " << ec.message();
        error_case(ec);
    }
    else
    {
        dns_seed_node->send(get_address_type(),
            strand_.wrap(std::bind(&protocol::seeds::handle_send_get_address,
                shared_from_this(), _1)));
        dns_seed_node->subscribe_address(
            strand_.wrap(std::bind(&protocol::seeds::save_addresses,
                shared_from_this(), _1, _2, dns_seed_node)));
    }

}

void recv_transaction(const std::error_code& ec,
    const transaction_type& tx, channel_ptr node)
{
    if (ec)
    {
        log_error() << "transaction: " << ec.message();
        return;
    }
    p->transaction_pool_.store(tx, handle_confirm,
        std::bind(&handle_mempool_store, _1, _2, tx, node));
    node->subscribe_transaction(std::bind(recv_transaction, _1, _2, node));
}

void node_impl::monitor_tx(const std::error_code& ec, channel_ptr node)
{
    if (ec)
    {
        log_warning() << "Couldn't start connection: " << ec.message();
        return;
    }
    node->subscribe_transaction(
        std::bind(&node_impl::recv_transaction, this, _1, _2, node));
    protocol_.subscribe_channel(
        std::bind(&node_impl::monitor_tx, this, _1, _2));
}

void session::get_data(const std::error_code& ec,
    const get_data_type& packet, channel_ptr node)
{
    if (ec)
    {
        log_error(LOG_SESSION) << "get_data: " << ec.message();
        return;
    }
    // simple stuff
    node->subscribe_get_data(
        std::bind(&session::get_data, this, _1, _2, node));
}

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 getx_responder::monitor(channel_ptr node)
{
    // Use this wrapper to add shared state to our node for inv
    // requests to trigger getblocks requests so a node can continue
    // downloading blocks.
    channel_with_state special;
    special.node = node;

    // Serve tx and blocks to nodes.
    BITCOIN_ASSERT(node);
    node->subscribe_get_data(
        std::bind(&getx_responder::receive_get_data,
            this, _1, _2, special));
}

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 poller::receive_block(const std::error_code& ec,
    const block_type& block, channel_ptr node)
{
    if (!node)
        return;

    if (ec)
    {
        log_warning(LOG_POLLER) << "Received bad block: "
            << ec.message();
        node->stop(/* ec */);
        return;
    }

    blockchain_.store(block,
        strand_.wrap(&poller::handle_store_block,
            this, _1, _2, hash_block_header(block.header), node));

    // Resubscribe.
    node->subscribe_block(
        std::bind(&poller::receive_block,
            this, _1, _2, node));
}

void getx_responder::pool_tx(const std::error_code& code,
    const transaction_type& tx, const hash_digest& tx_hash, channel_ptr node)
{
    if (code)
    {
        chain_.fetch_transaction(tx_hash,
            std::bind(&getx_responder::chain_tx,
                this, _1, _2, node));
    }
    else
    {
        BITCOIN_ASSERT(node);
        node->send(tx, [](const std::error_code&) {});
    }
}

void poller::receive_block(const std::error_code& ec,
                           const block_type& blk, channel_ptr node)
{
    if (ec)
    {
        log_error(LOG_POLLER) << "Received bad block: " << ec.message();
        return;
    }
    chain_.store(blk,
                 std::bind(&poller::handle_store,
                           this, _1, _2, hash_block_header(blk.header), node));
    node->subscribe_block(
        std::bind(&poller::receive_block,
                  this, _1, _2, node));
}

void protocol::receive_address_message(const std::error_code& ec,
                                       const address_type& packet, channel_ptr node)
{
    if (ec)
    {
        log_error(LOG_PROTOCOL)
                << "Problem receiving addresses: " << ec.message();
        return;
    }
    log_debug(LOG_PROTOCOL) << "Storing addresses.";
    for (const network_address_type& net_address: packet.addresses)
        hosts_.store(net_address, strand_.wrap(
                         &protocol::handle_store_address, this, _1));
    node->subscribe_address(strand_.wrap(
                                &protocol::receive_address_message, this, _1, _2, node));
}

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

void send_tx(const std::error_code& ec, channel_ptr node, transaction_type& tx)
{
    check_error(ec);
    std::cout << "sendtx: Sending " << hash_transaction(tx) << std::endl;
    auto handle_send =
        [node](const std::error_code& ec)
        {
            if (ec)
                log_warning() << "Send failed: " << ec.message();
            else
                std::cout << "sendtx: Sent "
                    << time(nullptr) << std::endl;
            stopped = true;
        };
    node->send(tx, handle_send);
}

void inventory_received(const std::error_code& ec, const inventory_type& inv,
    channel_ptr node, tx_watch& watch)
{
    check_error(ec);
    // Loop through inventory hashes.
    for (const inventory_vector_type& ivec: inv.inventories)
    {
        // We're only interested in transactions. Discard everything else.
        if (ivec.type != inventory_type_id::transaction)
            continue;
        watch.push(ivec.hash);
    }
    // Resubscribe to inventory packets.
    node->subscribe_inventory(
        std::bind(inventory_received, _1, _2, node, std::ref(watch)));
}

void poller::handle_store_block(const std::error_code& ec, block_info info,
    const hash_digest& block_hash, channel_ptr node)
{
    if (ec == error::duplicate)
    {
        // This is common, we get blocks we already have.
        log_debug(LOG_POLLER) << "Redundant block ["
            << encode_hash(block_hash) << "]";
        return;
    }

    if (ec)
    {
        log_error(LOG_POLLER) << "Error storing block ["
            << encode_hash(block_hash) << "] " << ec.message();
        node->stop(/* ec */);
        return;
    }
    
    switch (info.status)
    {
        // The block has been accepted as an orphan (ec not set).
        case block_status::orphan:
            log_debug(LOG_POLLER) << "Potential block ["
                << encode_hash(block_hash) << "]";

            // This is how we get other nodes to send us the blocks we are
            // missing from the top of our chain to the orphan.
            request_blocks(block_hash, node);
            break;

        // The block has been rejected from the store (redundant?).
        // This case may be redundant with error::duplicate.
        case block_status::rejected:
            log_debug(LOG_POLLER) << "Rejected block ["
                << encode_hash(block_hash) << "]";
            break;

        // The block has been accepted into the long chain (ec not set).
        case block_status::confirmed:
            log_info(LOG_POLLER) << "Block #"
                << info.height << " " << encode_hash(block_hash);
            break;
    }
}

void send_tx(const std::error_code& ec, channel_ptr node,
    protocol& prot, transaction_type& tx)
{
    check_error(ec);
    std::cout << "sendtx-p2p: Sending " << hash_transaction(tx) << std::endl;
    auto handle_send =
        [](const std::error_code& ec)
        {
            if (ec)
                log_warning() << "Send failed: " << ec.message();
            else
                std::cout << "sendtx-p2p: Sent "
                    << time(nullptr) << std::endl;
        };
    node->send(tx, handle_send);
    prot.subscribe_channel(
        std::bind(send_tx, _1, _2, std::ref(prot), std::ref(tx)));
}

void protocol::handle_manual_connect(
    const std::error_code& ec, channel_ptr node,
    const std::string& hostname, uint16_t port)
{
    if (ec)
    {
        log_warning(LOG_PROTOCOL) << "Manual connection to "
                                  << hostname << ":" << port << " failed with " << ec.message();
        maintain_connection(hostname, port);
        return;
    }
    manual_connections_.push_back(node);
    // Connected!
    log_info(LOG_PROTOCOL) << "Manual connection established: "
                           << hostname << ":" << port;
    // Remove channel from list of connections
    node->subscribe_stop(strand_.wrap(
                             &protocol::manual_channel_stopped, this, _1, node, hostname, port));
    setup_new_channel(node);
}

// Start monitoring this channel.
void poller::monitor(channel_ptr node)
{
    if (!node)
    {
        log_error(LOG_POLLER)
            << "The node is not initialized and cannot be monitored.";
        return;
    }

    //////node->subscribe_inventory(
    //////    std::bind(&poller::receive_inv,
    //////        this, _1, _2, node));

    node->subscribe_block(
        std::bind(&poller::receive_block,
            this, _1, _2, node));

    // Issue the initial ask for blocks.
    request_blocks(null_hash, node);
}

void recv_blk(const std::error_code& ec, const message::block& packet,
    channel_ptr node, blockchain_ptr chain)
{
    static bool stop_inserts = false;
    if (ec)
        log_error() << ec.message();
    node->subscribe_block(std::bind(recv_blk, _1, _2, node, chain));
    // store block in bdb
    //if (hash_block_header(packet) ==
    //    hash_digest{0x00, 0x00, 0x00, 0x00, 0xd1, 0x14, 0x57, 0x90,
    //                0xa8, 0x69, 0x44, 0x03, 0xd4, 0x06, 0x3f, 0x32,
    //                0x3d, 0x49, 0x9e, 0x65, 0x5c, 0x83, 0x42, 0x68,
    //                0x34, 0xd4, 0xce, 0x2f, 0x8d, 0xd4, 0xa2, 0xee})
    //{
    //    test_mem_pool(packet);
    //    stop_inserts = true;
    //}
    //else if (!stop_inserts)
        chain->store(packet, std::bind(handle_store, _1, _2, node, chain, hash_block_header(packet)));
}