void subscribe_manager::do_renew(
    const incoming_message& request, queue_send_callback queue_send)
{
    payment_address addr_key;
    if (!deserialize_address(addr_key, request.data()))
    {
        log_warning(LOG_SUBSCRIBER) << "Incorrect format for subscribe renew.";
        return;
    }
    const posix_time::ptime now = second_clock::universal_time();
    // Find entry and update expiry_time.
    auto range = subs_.equal_range(addr_key);
    for (auto it = range.first; it != range.second; ++it)
    {
        subscription& sub = it->second;
        // Only update subscriptions which were created by
        // the same client as this request originated from.
        if (sub.client_origin != request.origin())
            continue;
        // Future expiry time.
        sub.expiry_time = now + sub_expiry;
    }
    // Send response.
    data_chunk result(4);
    auto serial = make_serializer(result.begin());
    write_error_code(serial, std::error_code());
    outgoing_message response(request, result);
    queue_send(response);
}

bool backend_cluster::process_filters(const incoming_message& response)
{
    auto filter_it = filters_.find(response.command());
    if (filter_it == filters_.end())
        return false;
    filter_it->second(response.data(), response.origin());
    return true;
}

bool backend_cluster::process_as_reply(const incoming_message& response)
{
    auto handle_it = handlers_.find(response.id());
    // Unknown response. Not in our map.
    if (handle_it == handlers_.end())
        return false;
    handle_it->second(response.data(), response.origin());
    handlers_.erase(handle_it);
    size_t n_erased = retry_queue_.erase(response.id());
    BITCOIN_ASSERT(n_erased == 1);
    return true;
}

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 protocol_broadcast_transaction(server_node& node,
    const incoming_message& request, queue_send_callback queue_send)
{
    const data_chunk& raw_tx = request.data();
    chain::transaction tx;
    data_chunk result(4);
    auto serial = make_serializer(result.begin());

    if (!tx.from_data(raw_tx))
    {
        // error
        write_error_code(serial, error::bad_stream);
        outgoing_message response(request, result);
        queue_send(response);
        return;
    }

    auto ignore_send = [](const std::error_code&, size_t) {};

    // Send and hope for the best!
    node.protocol().broadcast(tx, ignore_send);

    // Response back to user saying everything is fine.
    write_error_code(serial, std::error_code());

    log_debug(LOG_SERVICE)
        << "protocol.broadcast_transaction() finished. Sending response.";

    outgoing_message response(request, result);
    queue_send(response);
}

std::error_code subscribe_manager::add_subscription(
    const incoming_message& request, queue_send_callback queue_send)
{
    payment_address addr_key;
    if (!deserialize_address(addr_key, request.data()))
    {
        log_warning(LOG_SUBSCRIBER) << "Incorrect format for subscribe data.";
        return error::bad_stream;
    }
    // Now create subscription.
    const posix_time::ptime now = second_clock::universal_time();
    // Limit absolute number of subscriptions to prevent exhaustion attacks.
    if (subs_.size() >= subscribe_limit_)
        return error::pool_filled;
    subs_.emplace(addr_key, subscription{
        now + sub_expiry, request.origin(), queue_send});
    return std::error_code();
}

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 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 subscribe_manager::do_subscribe(
    const incoming_message& request, zmq_socket_ptr socket)
{
    payment_address addr_key;
    if (!deserialize_address(addr_key, request.data()))
    {
        log_warning(LOG_SUBSCRIBER) << "Incorrect format for subscribe data.";
        return;
    }
    // Now create subscription.
    const posix_time::ptime now = second_clock::universal_time();
    subs_.emplace(addr_key, subscription{
        now + sub_expiry, request.origin(), socket});
    // Send response.
    data_chunk result(4);
    auto serial = make_serializer(result.begin());
    write_error_code(serial, std::error_code());
    outgoing_message response(request, result);
    response.send(*socket);
}

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));
}

bool unwrap_fetch_transaction_args(
    hash_digest& tx_hash, const incoming_message& request)
{
    const data_chunk& data = request.data();
    if (data.size() != 32)
    {
        log_error(LOG_WORKER)
            << "Incorrect data size for *.fetch_transaction";
        return false;
    }
    auto deserial = make_deserializer(data.begin(), data.end());
    tx_hash = deserial.read_hash();
    return true;
}

void blockchain_fetch_block_header(node_impl& node,
    const incoming_message& request, queue_send_callback queue_send)
{
    const data_chunk& data = request.data();
    if (data.size() == 32)
        fetch_block_header_by_hash(node, request, queue_send);
    else if (data.size() == 4)
        fetch_block_header_by_height(node, request, queue_send);
    else
    {
        log_error(LOG_WORKER)
            << "Incorrect data size for blockchain.fetch_block_header";
        return;
    }
}

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));
}

bool unwrap_fetch_history_args(
    payment_address& payaddr, uint32_t& from_height,
    const incoming_message& request)
{
    const data_chunk& data = request.data();
    if (data.size() != 1 + short_hash_size + 4)
    {
        log_error(LOG_WORKER)
            << "Incorrect data size for .fetch_history";
        return false;
    }
    auto deserial = make_deserializer(data.begin(), data.end());
    uint8_t version_byte = deserial.read_byte();
    short_hash hash = deserial.read_short_hash();
    from_height = deserial.read_4_bytes();
    BITCOIN_ASSERT(deserial.iterator() == data.end());
    payaddr.set(version_byte, hash);
    return true;
}

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));
}

outgoing_message::outgoing_message(
    const incoming_message& request, const data_chunk& data)
  : dest_(request.origin()), command_(request.command()),
    id_(request.id()), data_(data)
{
}