1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
|
#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;
}
|
