source: source/ariba/utility/bootstrap/modules/periodicbroadcast/PeriodicBroadcast.h@ 10706

Last change on this file since 10706 was 10653, checked in by Michael Tänzer, 12 years ago

Merge the ASIO branch back into trunk
File size: 10.7 KB
Line 
1// [License]
2// The Ariba-Underlay Copyright
3//
4// Copyright (c) 2008-2009, Institute of Telematics, UniversitÀt Karlsruhe (TH)
5//
6// Institute of Telematics
7// UniversitÀt Karlsruhe (TH)
8// Zirkel 2, 76128 Karlsruhe
9// Germany
10//
11// Redistribution and use in source and binary forms, with or without
12// modification, are permitted provided that the following conditions are
13// met:
14//
15// 1. Redistributions of source code must retain the above copyright
16// notice, this list of conditions and the following disclaimer.
17// 2. Redistributions in binary form must reproduce the above copyright
18// notice, this list of conditions and the following disclaimer in the
19// documentation and/or other materials provided with the distribution.
20//
21// THIS SOFTWARE IS PROVIDED BY THE INSTITUTE OF TELEMATICS ``AS IS'' AND
22// ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
23// IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
24// PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE ARIBA PROJECT OR
25// CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL,
26// EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
27// PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR
28// PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
29// LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING
30// NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
31// SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
32//
33// The views and conclusions contained in the software and documentation
34// are those of the authors and should not be interpreted as representing
35// official policies, either expressed or implied, of the Institute of
36// Telematics.
37// [License]
38
39#ifndef __PERIODIC_BROADCAST_H
40#define __PERIODIC_BROADCAST_H
41
42#include "ariba/config.h"
43
44#include <map>
45#include <string>
46#include <ctime>
47#include <iostream>
48#include <boost/asio.hpp>
49#include <boost/foreach.hpp>
50#include <boost/thread/mutex.hpp>
51#include <boost/thread/thread.hpp>
52#include "ariba/utility/bootstrap/modules/BootstrapModule.h"
53#include "ariba/utility/logging/Logging.h"
54#include "ariba/utility/system/Timer.h"
55#include "PeriodicBroadcastMessage.h"
56
57//link-local
58#include "ariba/utility/transport/tcpip/tcpip.hpp"
59
60using std::map;
61using std::string;
62using boost::asio::ip::udp;
63
64namespace ariba {
65namespace utility {
66
67class PeriodicBroadcast : public BootstrapModule, public Timer {
68 use_logging_h(PeriodicBroadcast);
69public:
70 PeriodicBroadcast(BootstrapInformationCallback* _callback, string info);
71 virtual ~PeriodicBroadcast();
72
73 virtual void start();
74 virtual void stop();
75
76 virtual string getName();
77 virtual string getInformation();
78 virtual bool isFunctional();
79 virtual void publishService(string name, string info1, string info2, string info3);
80 virtual void revokeService(string name);
81
82protected:
83 virtual void eventFunction();
84
85private:
86 void sendLocalServices();
87 void updateRemoteServices();
88
89 static const long timerinterval; // used to send out updates on our services and check for new services
90 static const long servicetimeout; // timeout after that a service is dead when we did not receive updates
91 static const unsigned int serverport_v4;
92 static const unsigned int serverport_v6;
93
94 class Service {
95 private:
96 string name;
97 string info1;
98 string info2;
99 string info3;
100 time_t lastseen;
101
102 public:
103 Service()
104 : name(""), info1(""), info2(""), info3(""), lastseen(0){
105 }
106
107 Service(const string& _name, const string& _info1,
108 const string& _info2, const string& _info3, const time_t& _lastseen = 0){
109 name.assign (_name);
110 info1.assign(_info1);
111 info2.assign(_info2);
112 info3.assign(_info3);
113 lastseen = _lastseen;
114 }
115
116 Service(const Service& rh){
117 name.assign (rh.name);
118 info1.assign(rh.info1);
119 info2.assign(rh.info2);
120 info3.assign(rh.info3);
121 lastseen = rh.lastseen;
122 }
123
124 string getName() const {
125 return name;
126 }
127
128 string getInfo1() const {
129 return info1;
130 }
131
132 string getInfo2() const {
133 return info2;
134 }
135
136 string getInfo3() const {
137 return info3;
138 }
139
140 time_t getLastseen() const {
141 return lastseen;
142 }
143
144 void setName(string _name){
145 name.assign(_name);
146 }
147
148 void setInfo1(string _info1){
149 info1.assign(_info1);
150 }
151
152 void setInfo2(string _info2){
153 info2.assign(_info2);
154 }
155
156 void setInfo3(string _info3){
157 info3.assign(_info3);
158 }
159
160 void setLastseen(time_t _lastseen){
161 lastseen = _lastseen;
162 }
163
164 Service& operator=(const Service& rh){
165 this->name.assign( rh.getName() );
166 this->info1.assign( rh.getInfo1() );
167 this->info2.assign( rh.getInfo2() );
168 this->info3.assign( rh.getInfo3() );
169 this->lastseen = rh.lastseen;
170 return *this;
171 }
172 };
173
174 typedef map<string,Service> ServiceList;
175
176 ServiceList localServices;
177 boost::mutex localServicesMutex;
178
179 ServiceList remoteServices;
180 boost::mutex remoteServicesMutex;
181
182 ServiceList newRemoteServices;
183 boost::mutex newRemoteServicesMutex;
184
185 boost::asio::io_service io_service;
186 boost::thread* io_service_thread;
187 static void threadFunc(PeriodicBroadcast* obj);
188
189 class udp_server {
190 private:
191 udp::socket socket_v4;
192 udp::socket socket_v6;
193 udp::endpoint remote_endpoint_;
194 boost::array<char, 1500> recv_buffer_4;
195 boost::array<char, 1500> recv_buffer_6;
196 ServiceList* services;
197 boost::mutex* servicesmutex;
198
199 public:
200 udp_server(boost::asio::io_service& io_service, ServiceList* _services, boost::mutex* _servicesmutex)
201 : socket_v4(io_service), socket_v6(io_service),
202 services(_services), servicesmutex(_servicesmutex) {
203
204 if( open4() ) start_receive_4();
205 if( open6() ) start_receive_6();
206 }
207
208 bool open4(){
209 boost::system::error_code err;
210
211 boost::asio::ip::udp::endpoint listen_endpoint_v4(
212 boost::asio::ip::address_v4::any(),
213 PeriodicBroadcast::serverport_v4);
214
215 err = socket_v4.open( listen_endpoint_v4.protocol(), err );
216 if(err){
217 logging_warn("failed opening ipv4 socket");
218 return false;
219 }
220
221 err = socket_v4.set_option( boost::asio::ip::udp::socket::reuse_address(true), err );
222 if(err){
223 logging_warn("failed setting reuse address option on ipv4 socket");
224 return false;
225 }
226
227 err = socket_v4.set_option( boost::asio::socket_base::broadcast(true), err );
228 if(err){
229 logging_warn("failed setting broadcast option on ipv4 socket");
230 return false;
231 }
232
233 err = socket_v4.bind( listen_endpoint_v4, err );
234 if(err){
235 logging_warn("failed binding ipv4 socket");
236 return false;
237 }
238
239 return true;
240 }
241
242 bool open6(){
243 boost::system::error_code err;
244
245 boost::asio::ip::udp::endpoint listen_endpoint_v6(
246 boost::asio::ip::address_v6::any(),
247 PeriodicBroadcast::serverport_v6);
248
249 err = socket_v6.open( listen_endpoint_v6.protocol(), err );
250 if(err){
251 logging_warn("failed opening ipv6 socket");
252 return false;
253 }
254
255 err = socket_v6.set_option( boost::asio::ip::udp::socket::reuse_address(true), err );
256 if(err){
257 logging_warn("failed setting reuse address option on ipv6 socket");
258 return false;
259 }
260
261 err = socket_v6.set_option( boost::asio::socket_base::broadcast(true), err );
262 if(err){
263 logging_warn("failed setting broadcast option on ipv6 socket");
264 return false;
265 }
266
267 err = socket_v6.bind( listen_endpoint_v6, err );
268 if(err){
269 logging_warn("failed binding ipv6 socket");
270 return false;
271 }
272
273 return true;
274 }
275
276 void sendservice(Service service){
277
278 PeriodicBroadcastMessage msg;
279 if(service.getName().empty()) return;
280
281 msg.setName( service.getName() );
282 msg.setInfo1( service.getInfo1() );
283 msg.setInfo2( service.getInfo2() );
284 msg.setInfo3( service.getInfo3() );
285
286 Data data = data_serialize( msg, DEFAULT_V );
287 uint8_t* pnt = data.getBuffer();
288 size_t len = data.getLength() / 8;
289
290 boost::system::error_code err;
291
292 {
293 udp::endpoint endp(udp::v4(), PeriodicBroadcast::serverport_v4);
294 endp.address( boost::asio::ip::address_v4::broadcast() );
295 socket_v4.send_to( boost::asio::buffer(pnt, len), endp, 0, err );
296 if(err) logging_warn("failed sending message through ipv4 socket");
297 }
298 {
299 udp::endpoint endp(udp::v6(), PeriodicBroadcast::serverport_v6);
300 boost::asio::ip::address_v6 all_nodes = boost::asio::ip::address_v6::from_string("ff02::1");
301
302 // include all link-local interfaces
303 vector<uint64_t> scope_ids = ariba::transport::tcpip::get_interface_scope_ids();
304
305 BOOST_FOREACH ( uint64_t id, scope_ids )
306 {
307 all_nodes.scope_id(id);
308 endp.address( all_nodes );
309
310 socket_v6.send_to( boost::asio::buffer(pnt, len), endp, 0, err );
311 if(err) logging_warn("failed sending message through ipv6 socket");
312 }
313 }
314 }
315
316 private:
317 void start_receive_4(){
318 socket_v4.async_receive_from(
319 boost::asio::buffer(recv_buffer_4), remote_endpoint_,
320 boost::bind(&udp_server::handle_receive_4, this,
321 boost::asio::placeholders::error,
322 boost::asio::placeholders::bytes_transferred));
323 }
324
325 void start_receive_6(){
326 socket_v6.async_receive_from(
327 boost::asio::buffer(recv_buffer_6), remote_endpoint_,
328 boost::bind(&udp_server::handle_receive_6, this,
329 boost::asio::placeholders::error,
330 boost::asio::placeholders::bytes_transferred));
331 }
332
333 void handle_receive_4(const boost::system::error_code& error,
334 std::size_t bytes_transferred){
335
336 if (!error || error == boost::asio::error::message_size)
337 handle_info(recv_buffer_4, bytes_transferred);
338 else
339 logging_warn("failed receiving broadcast data: " << error.message());
340
341 start_receive_4();
342 }
343
344 void handle_receive_6(const boost::system::error_code& error,
345 std::size_t bytes_transferred){
346
347 if (!error || error == boost::asio::error::message_size)
348 handle_info(recv_buffer_6, bytes_transferred);
349 else
350 logging_warn("failed receiving broadcast data: " << error.message());
351
352 start_receive_6();
353 }
354
355 void handle_info(boost::array<char, 1500>& buffer, std::size_t length){
356
357 try {
358
359 PeriodicBroadcastMessage msg;
360
361 Data data( (uint8_t*)buffer.data(), length*8 );
362 data_deserialize( msg, data );
363
364 { // insert new found service
365 boost::mutex::scoped_lock lock( *servicesmutex );
366 if(msg.getName().empty()) return;
367
368 ServiceList::iterator it = services->find( msg.getName() );
369 if( it != services->end() ){
370 it->second.setLastseen( time(NULL) );
371 } else {
372 Service s( msg.getName(), msg.getInfo1(), msg.getInfo2(), msg.getInfo3(), time(NULL));
373 services->insert( std::make_pair(msg.getName(), s) );
374 }
375 }
376
377 }catch(...){
378 /* ignore error */
379 }
380 }
381
382 void handle_send(boost::shared_ptr<std::string> /*message*/,
383 const boost::system::error_code& error,
384 std::size_t /*bytes_transferred*/){
385
386 if(error)
387 logging_warn("failed sending out message");
388 }
389 };
390
391 udp_server server;
392};
393
394}} //namespace ariba, utility
395
396#endif // __BLUETOOTH_SDP_H
Note: See TracBrowser for help on using the repository browser.