callee_new.cpp

#include <iostream>
#include <memory>
#include <string>
#include <thread>
#include <tuple>
#include <sstream>
#include <cstdlib>

#include <boost/asio.hpp>
#include <boost/version.hpp>
#include <boost/date_time/posix_time/posix_time.hpp>

#include <autobahn/autobahn.hpp>


void log (const std::string& msg) {
    std::cerr << msg << " [thread " << boost::this_thread::get_id() << "]" << std::endl;
}


void add2(autobahn::wamp_invocation invocation)
{
    auto a = invocation->argument<uint64_t>(0);
    auto b = invocation->argument<uint64_t>(1);

    std::ostringstream oss;
    oss << "Procedure com.examples.calculator.add2 invoked: " << a << ", " << b << std::endl;
    log(oss.str());

    invocation->result(std::make_tuple(a + b));
}


void longop(autobahn::wamp_invocation invocation)
{
    auto a = invocation->argument<uint64_t>(0);

    std::ostringstream oss;
    oss << "Procedure com.myapp.longop invoked: " << a << std::endl;
    log(oss.str());

    uint64_t i = 0;
    for (; i < a; i++)
    {
        boost::this_thread::sleep(boost::posix_time::milliseconds(3000));
        if (i < a)
        {
            invocation->progress(std::make_tuple(i));
        }
    }
    invocation->result(std::make_tuple(i));
}


int main(int argc, char** argv)
{
    if (argc != 3) {
        std::cerr << "Usage: callee <RawSocket IP> <RawSocket PORT>" << std::endl;
        return -1;
    }

    try {
        log("starting program ..");
        log(argv[1]);
        log(argv[1]);

        boost::asio::io_service io;

        bool debug = true;

        // Make sure the continuation futures we use do not run out of scope prematurely.
        // Since we are only using one thread here this can cause the io service to block
        // as a future generated by a continuation will block waiting for its promise to be
        // fulfilled when it goes out of scope. This would prevent the session from receiving
        // responses from the router.
        boost::future<void> f1, f2, f3;

        auto endpoint = boost::asio::ip::tcp::endpoint(boost::asio::ip::address::from_string(argv[1]), atoi(argv[2]));

        auto transport = std::make_shared<autobahn::wamp_tcp_transport>(io, endpoint, debug);

        auto session = std::make_shared<autobahn::wamp_session>(io, debug);

        transport->attach(std::static_pointer_cast<autobahn::wamp_transport_handler>(session));

        f1 = transport->connect().then([&](boost::future<void> connected) {

            connected.get();

            log("transport connected");

            f2 = session->start().then(boost::launch::deferred, [&](boost::future<void> started) {

                started.get();

                log("session started");

                f3 = session->join("realm1").then(boost::launch::deferred, [&](boost::future<uint64_t> joined) {

                    joined.get();

                    log("joined realm");

                    auto f4 = session->provide("com.examples.calculator.add2", &add2).then(
                        boost::launch::deferred,
                        [&](boost::future<autobahn::wamp_registration> registration) {
                        log("registered procedure com.examples.calculator.add2");
                    });

                    auto f5 = session->provide("com.myapp.longop", &longop).then(
                        boost::launch::deferred,
                        [&](boost::future<autobahn::wamp_registration> registration) {
                        log("registered procedure com.myapp.longop");
                    });

                    f5.get();
                    f4.get();
                });
                f3.get();

            });
            f2.get();
        });

        log("starting io service ..");
        io.run();
        log("stopped io service");
    }
    catch (const std::exception& e) {
        log(e.what());
        return -1;
    }

    return 0;
}