query_service_handler::query_service_handler(
    config_map_type& config, node_impl& node)
  : stop_secret_(config["stop-secret"].c_str()),
    chain_(node.blockchain()),
    txpool_(node.transaction_pool()),
    protocol_(node.protocol())
{
}

void fullnode_fetch_history(node_impl& node,
    const incoming_message& request, zmq_socket_ptr socket)
{
    payment_address payaddr;
    uint32_t from_height;
    if (!unwrap_fetch_history_args(payaddr, from_height, request))
        return;
    log_debug(LOG_WORKER) << "fetch_history("
        << payaddr.encoded() << ", from_height=" << from_height << ")";
    fetch_history(node.blockchain(), node.transaction_indexer(),
        payaddr,
        std::bind(send_history_result, _1, _2, request, socket),
        from_height);
}

void register_with_node(subscribe_manager& manager, node_impl& node)
{
    auto recv_blk = [&manager](size_t height, const block_type& blk)
    {
        const hash_digest blk_hash = hash_block_header(blk.header);
        for (const transaction_type& tx: blk.transactions)
            manager.submit(height, blk_hash, tx);
    };
    auto recv_tx = [&manager](const transaction_type& tx)
    {
        manager.submit(0, null_hash, tx);
    };
    node.subscribe_blocks(recv_blk);
    node.subscribe_transactions(recv_tx);
}

void blockchain_fetch_stealth(node_impl& node,
    const incoming_message& request, queue_send_callback queue_send)
{
    const data_chunk& data = request.data();
    if (data.empty())
    {
        log_error(LOG_WORKER)
            << "Incorrect data size (empty) for blockchain.fetch_stealth";
        return;
    }
    auto deserial = make_deserializer(data.begin(), data.end());
    // number_bits
    uint8_t number_bits = deserial.read_byte();
    stealth_prefix prefix(number_bits);
    log_info(LOG_WORKER) << data.size() << " " << prefix.num_blocks();
    if (data.size() != 1 + prefix.num_blocks() + 4)
    {
        log_error(LOG_WORKER)
            << "Incorrect data size (" << data.size()
	    << ") for blockchain.fetch_stealth";
        return;
    }
    // actual bitfield data
    data_chunk bitfield = deserial.read_data(prefix.num_blocks());
    boost::from_block_range(bitfield.begin(), bitfield.end(), prefix);
    // from_height
    size_t from_height = deserial.read_4_bytes();
    node.blockchain().fetch_stealth(prefix,
        std::bind(stealth_fetched, _1, _2, request, queue_send), from_height);
}

void fetch_block_header_by_hash(node_impl& node,
    const incoming_message& request, queue_send_callback queue_send)
{
    const data_chunk& data = request.data();
    BITCOIN_ASSERT(data.size() == 32);
    auto deserial = make_deserializer(data.begin(), data.end());
    const hash_digest blk_hash = deserial.read_hash();
    node.blockchain().fetch_block_header(blk_hash,
        std::bind(block_header_fetched, _1, _2, request, queue_send));
}

void fetch_block_header_by_height(node_impl& node,
    const incoming_message& request, queue_send_callback queue_send)
{
    const data_chunk& data = request.data();
    BITCOIN_ASSERT(data.size() == 4);
    auto deserial = make_deserializer(data.begin(), data.end());
    size_t height = deserial.read_4_bytes();
    node.blockchain().fetch_block_header(height,
        std::bind(block_header_fetched, _1, _2, request, queue_send));
}

void blockchain_fetch_transaction(node_impl& node,
    const incoming_message& request, queue_send_callback queue_send)
{
    hash_digest tx_hash;
    if (!unwrap_fetch_transaction_args(tx_hash, request))
        return;
    log_debug(LOG_REQUEST) << "blockchain.fetch_transaction("
        << tx_hash << ")";
    node.blockchain().fetch_transaction(tx_hash,
        std::bind(transaction_fetched, _1, _2, request, queue_send));
}

void blockchain_fetch_last_height(node_impl& node,
    const incoming_message& request, queue_send_callback queue_send)
{
    const data_chunk& data = request.data();
    if (!data.empty())
    {
        log_error(LOG_WORKER)
            << "Incorrect data size for blockchain.fetch_last_height";
        return;
    }
    node.blockchain().fetch_last_height(
        std::bind(last_height_fetched, _1, _2, request, queue_send));
}

void blockchain_fetch_history(node_impl& node,
    const incoming_message& request, queue_send_callback queue_send)
{
    payment_address payaddr;
    uint32_t from_height;
    if (!unwrap_fetch_history_args(payaddr, from_height, request))
        return;
    log_debug(LOG_REQUEST) << "blockchain.fetch_history("
        << payaddr.encoded() << ", from_height=" << from_height << ")";
    node.blockchain().fetch_history(payaddr,
        std::bind(send_history_result,
            _1, _2, request, queue_send), from_height);
}

void blockchain_fetch_block_height(node_impl& node,
    const incoming_message& request, queue_send_callback queue_send)
{
    const data_chunk& data = request.data();
    if (data.size() != 32)
    {
        log_error(LOG_WORKER)
            << "Incorrect data size for blockchain.fetch_block_height";
        return;
    }
    auto deserial = make_deserializer(data.begin(), data.end());
    const hash_digest blk_hash = deserial.read_hash();
    node.blockchain().fetch_block_height(blk_hash,
        std::bind(block_height_fetched, _1, _2, request, queue_send));
}

void blockchain_fetch_spend(node_impl& node,
    const incoming_message& request, queue_send_callback queue_send)
{
    const data_chunk& data = request.data();
    if (data.size() != 36)
    {
        log_error(LOG_WORKER)
            << "Incorrect data size for blockchain.fetch_spend";
        return;
    }
    auto deserial = make_deserializer(data.begin(), data.end());
    output_point outpoint;
    outpoint.hash = deserial.read_hash();
    outpoint.index = deserial.read_4_bytes();
    node.blockchain().fetch_spend(outpoint,
        std::bind(spend_fetched, _1, _2, request, queue_send));
}

find_data::find_data(
	node_impl& node
	, node_id target
	, data_callback const& dcallback
	, nodes_callback const& ncallback
    , uint8_t search_type)
	: traversal_algorithm(node, target)
	, m_data_callback(dcallback)
	, m_nodes_callback(ncallback)
	, m_target(target)
    , m_id(node.nid())
	, m_done(false)
	, m_got_peers(false)
    , m_search_type(search_type)
{
	node.m_table.for_each_node(&add_entry_fun, 0, (traversal_algorithm*)this);
}

void transaction_pool_validate(node_impl& node,
                               const incoming_message& request, queue_send_callback queue_send)
{
    const data_chunk& raw_tx = request.data();
    transaction_type tx;
    try
    {
        satoshi_load(raw_tx.begin(), raw_tx.end(), tx);
    }
    catch (end_of_stream)
    {
        // error
        transaction_validated(error::bad_stream, index_list(),
                              request, queue_send);
        return;
    }
    node.transaction_pool().validate(tx,
                                     std::bind(transaction_validated, _1, _2, request, queue_send));
}

subscribe_manager::subscribe_manager(node_impl& node)
  : strand_(node.memory_related_threadpool())
{
    // subscribe to blocks and txs -> submit
    register_with_node(*this, node);
}

void send_dht_msg(node_impl& node, char const* msg, udp::endpoint const& ep
	, lazy_entry* reply, char const* t = "10", char const* info_hash = 0
	, char const* name = 0, std::string const token = std::string(), int port = 0
	, char const* target = 0, entry const* value = 0
	, bool scrape = false, bool seed = false
	, std::string const key = std::string(), std::string const sig = std::string()
	, int seq = -1)
{
	// we're about to clear out the backing buffer
	// for this lazy_entry, so we better clear it now
	reply->clear();
	entry e;
	e["q"] = msg;
	e["t"] = t;
	e["h"] = "q";
	entry::dictionary_type& a = e["g"].dict();
	//a["id"] = generate_next().to_string();
	a["id"] = generate_id(ep.address()).to_string();
	if (info_hash) a["infoHash"] = std::string(info_hash, 20);
	if (name) a["n"] = name;
	if (!token.empty()) a["token"] = token;
	if (port) a["port"] = port;
	if (target) a["target"] = std::string(target, 20);
	if (value) a["v"] = *value;
	if (!sig.empty()) a["sig"] = sig;
	if (!key.empty()) a["k"] = key;
	if (scrape) a["scrape"] = 1;
	if (seed) a["seed"] = 1;
	if (seq >= 0) a["seq"] = seq;
	char msg_buf[1500];
	int size = bencode(msg_buf, e);
#if defined TORRENT_DEBUG && TORRENT_USE_IOSTREAM
// this yields a lot of output. too much
//	std::cerr << "sending: " <<  e << "\n";
#endif
	//fprintf(stderr, "sending: %s\n", msg_buf);

	lazy_entry decoded;
	error_code ec;
	lazy_bdecode(msg_buf, msg_buf + size, decoded, ec);
	if (ec) fprintf(stderr, "lazy_bdecode failed: %s\n", ec.message().c_str());

	dht::msg m(decoded, ep);
	node.incoming(m);

	// by now the node should have invoked the send function and put the
	// response in g_responses

	std::list<std::pair<udp::endpoint, entry> >::iterator i
		= std::find_if(g_responses.begin(), g_responses.end()
			, boost::bind(&std::pair<udp::endpoint, entry>::first, _1) == ep);
	if (i == g_responses.end())
	{
		TEST_ERROR("not response from DHT node");
		return;
	}

	static char inbuf[1500];
	int len = bencode(inbuf, i->second);
	g_responses.erase(i);
	int ret = lazy_bdecode(inbuf, inbuf + len, *reply, ec);
	TEST_CHECK(ret == 0);
}