Function main

cpp/examples/multithreaded_client_flow_control.cpp:263–316

}

int main(int argc, const char **argv) {
    try {
        if (argc != 5) {
            std::cerr <<
                "Usage: " << argv[0] << " CONNECTION-URL AMQP-ADDRESS MESSAGE-COUNT THREAD-COUNT\n"
                "CONNECTION-URL: connection address, e.g.'amqp://127.0.0.1'\n"
                "AMQP-ADDRESS: AMQP node address, e.g. 'examples'\n"
                "MESSAGE-COUNT: number of messages to send\n"
                "THREAD-COUNT: number of sender/receiver thread pairs\n";
            return 1;
        }

        const char *url = argv[1];
        const char *address = argv[2];
        int n_messages = atoi(argv[3]);
        int n_threads = atoi(argv[4]);
        int count = n_messages * n_threads;

        // Total messages to be received, multiple receiver threads will decrement this.
        std::atomic_int remaining;
        remaining.store(count);

        // Run the proton container
        proton::container container;
        auto container_thread = std::thread([&]() { container.run(); });

        // A single sender and receiver to be shared by all the threads
        sender send(container, url, address);
        receiver recv(container, url, address);

        // Start receiver threads, then sender threads.
        // Starting receivers first gives all receivers a chance to compete for messages.
        std::vector<std::thread> threads;
        threads.reserve(n_threads*2); // Avoid re-allocation once threads are started
        for (int i = 0; i < n_threads; ++i)
            threads.push_back(std::thread([&]() { receive_thread(recv, remaining); }));
        for (int i = 0; i < n_threads; ++i)
            threads.push_back(std::thread([&]() { send_thread(send, n_messages); }));

        // Wait for threads to finish
        for (auto& t : threads) t.join();
        send.close();
        recv.close();
        container_thread.join();
        if (remaining > 0)
            throw std::runtime_error("not all messages were received");
        std::cout << count << " messages sent and received" << std::endl;

        return 0;
    } catch (const std::exception& e) {
        std::cerr << e.what() << std::endl;
    }
    return 1;
}