.. _examples_txrad:

examples/txrad.cpp
#####################

::

    #include <bitcoin/bitcoin.hpp>
    using namespace bc;

    using std::placeholders::_1;
    using std::placeholders::_2;

    // Watches transactions. Keeps a view count per seen tx hash, and
    // cleans up old expired tx hashes.
    class tx_watch
      : public async_strand
    {
    public:
        tx_watch(threadpool& pool, time_t timeout=200);

        // Push a seen tx hash. If this entry exists then the count
        // will be incremented.
        // Else create a new entry in our list.
        void push(const hash_digest& tx_hash);
        // Cleans up expired items. We could make this implicitly called
        // by push() or display(), but single use methods with no side-effects
        // is better code design.
        void cleanup();
        // Display transactions and their count. A better design would be to
        // separate the view from the model and instead provide a method which
        // fetches a copy of our list, but we keep it simple here.
        void display();

    private:
        struct entry_count
        {
            hash_digest tx_hash;
            size_t count;
            // Timestamp of when transaction hash was first seen.
            time_t timest;
        };

        typedef std::vector<entry_count> entry_list;

        // The public methods push these methods to the threadpool to be
        // executed and then return immediately.
        // async_strand::queue() is a helper method which posts the work
        // to the threadpool and serializes access.
        // No 2 operations posted through the same async_strand using queue()
        // will execute at the same time.
        void do_push(const hash_digest& tx_hash);
        void do_cleanup();
        void do_display();

        entry_list entries_;
        // Time until an entry is ready to be removed.
        time_t timeout_;
    };

    tx_watch::tx_watch(threadpool& pool, time_t timeout)
      : async_strand(pool), timeout_(timeout)
    {
    }

    void tx_watch::push(const hash_digest& tx_hash)
    {
        queue(std::bind(&tx_watch::do_push, this, tx_hash));
        // Returns immediately.
    }
    void tx_watch::do_push(const hash_digest& tx_hash)
    {
        // If tx_hash is found then increment count...
        bool is_found = false;
        for (entry_count& entry: entries_)
            if (entry.tx_hash == tx_hash)
            {
                ++entry.count;
                is_found = true;
            }
        // Else create a new entry with a count of 1.
        if (!is_found)
            entries_.push_back({tx_hash, 1, time(nullptr)});
    }

    void tx_watch::cleanup()
    {
        queue(std::bind(&tx_watch::do_cleanup, this));
    }
    void tx_watch::do_cleanup()
    {
        // Erase entries where timest is older than (now - timeout_) seconds.
        time_t current_time = time(nullptr);
        auto erase_pred =
            [&](const entry_count& entry)
            {
                return (current_time - entry.timest) > timeout_;
            };
        auto erase_begin =
            std::remove_if(entries_.begin(), entries_.end(), erase_pred);
        // If we have old entries to delete then erase them.
        if (erase_begin != entries_.end())
            entries_.erase(erase_begin);
    }

    void tx_watch::display()
    {
        queue(std::bind(&tx_watch::do_display, this));
    }
    void tx_watch::do_display()
    {
        // Sort entries by count. Highest numbers at the top.
        std::sort(entries_.begin(), entries_.end(),
            [](const entry_count& entry_a, const entry_count& entry_b)
            {
                return entry_a.count > entry_b.count;
            });
        // Display the first 20 entries.
        for (size_t i = 0; i < 20 && i < entries_.size(); ++i)
        {
            const entry_count& entry = entries_[i];
            log_info() << entry.tx_hash << " " << entry.count;
        }
    }

    // We don't have a database open, and aren't doing any critical file
    // operations so we aren't worried about exiting suddenly.
    void check_error(const std::error_code& ec)
    {
        if (!ec)
            return;
        log_fatal() << ec.message();
        exit(-1);
    }

    // Needed for the C callback capturing the signals.
    bool stopped = false;
    void signal_handler(int sig)
    {
        log_info() << "Caught signal: " << sig;
        stopped = true;
    }

    // Started protocol. Node discovery complete.
    void handle_start(const std::error_code& ec)
    {
        check_error(ec);
        log_debug() << "Started.";
    }

    void connection_started(const std::error_code& ec, channel_ptr node,
        protocol& prot, tx_watch& watch);
    void inventory_received(const std::error_code& ec, const inventory_type& inv,
        channel_ptr node, tx_watch& watch);

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

    int main()
    {
        threadpool pool(4);
        // Create dependencies for our protocol object.
        hosts hst(pool);
        handshake hs(pool);
        network net(pool);
        // protocol service.
        protocol prot(pool, hst, hs, net);
        // Perform node discovery if needed and then creating connections.
        prot.start(handle_start);
        // Our table tracking transaction counts.
        tx_watch watch(pool, 200);
        // Notify us of new connections.
        prot.subscribe_channel(
            std::bind(connection_started, _1, _2, std::ref(prot), std::ref(watch)));
        // Catch C signals for stopping the program.
        signal(SIGABRT, signal_handler);
        signal(SIGTERM, signal_handler);
        signal(SIGINT, signal_handler);
        while (!stopped)
        {
            watch.cleanup();
            watch.display();
            sleep(10);
        }
        // Safely close down.
        pool.stop();
        pool.join();
        return 0;
    }