source: source/ariba/utility/transport/rfcomm/rfcomm.cpp@ 5423

Last change on this file since 5423 was 5423, checked in by mies, 15 years ago
File size: 9.5 KB
Line 
1#include "rfcomm.hpp"
2
3#include "../asio/asio_io_service.h"
4#include "../asio/rfcomm.hpp"
5#include "../asio/bluetooth_endpoint.hpp"
6
7#include <boost/asio.hpp>
8#include <boost/thread.hpp>
9#include <boost/array.hpp>
10#include <memory.h>
11#include <deque>
12#include <cerrno>
13
14namespace ariba {
15namespace transport {
16
17use_logging_cpp(rfcomm)
18
19using namespace boost::asio;
20using namespace detail;
21using namespace std;
22
23using boost::system::error_code;
24
25class link_data {
26public:
27 uint8_t size_[4];
28 size_t size;
29 uint8_t* buffer;
30};
31
32class link_info {
33public:
34 link_info(io_service& io ) :
35 up(false), local(), remote(), socket(io), connect_retries(0),
36 size(0), buffer(NULL), sending(false) {
37 }
38
39 // state
40 bool up;
41 rfcomm_endpoint local, remote;
42 bluetooth::rfcomm::socket socket;
43 int connect_retries;
44
45 // read buffer
46 uint8_t size_[4];
47 size_t size;
48 uint8_t* buffer;
49
50 // send buffer
51 bool sending;
52 boost::mutex mutex;
53 std::deque<link_data> send_buffer;
54};
55
56void rfcomm::shutdown(link_info* info) {
57 if (info != NULL && info->up) {
58 info->up = false;
59 info->socket.shutdown( bluetooth::rfcomm::socket::shutdown_both );
60 }
61}
62
63
64inline bluetooth::rfcomm::endpoint convert( const rfcomm_endpoint& endpoint ) {
65 return bluetooth::rfcomm::endpoint(
66 endpoint.mac().bluetooth(), endpoint.channel().value()
67 );
68}
69
70inline rfcomm_endpoint convert( const bluetooth::rfcomm::endpoint& endpoint ) {
71 mac_address mac;
72 mac.bluetooth( endpoint.address() );
73 rfcomm_channel_address channel;
74 channel.value( endpoint.channel() );
75 return rfcomm_endpoint( mac, channel );
76}
77
78
79rfcomm::rfcomm(uint16_t channel) :
80 channel(channel), io(asio_io_service::alloc()) {
81 accept_retries = 0;
82}
83
84rfcomm::~rfcomm() {
85 asio_io_service::free();
86}
87
88void rfcomm::start() {
89
90 // start io service
91 asio_io_service::start();
92
93 // create acceptor
94 logging_info( "Binding to channel " << channel );
95 acceptor = new bluetooth::rfcomm::acceptor(io,
96 bluetooth::rfcomm::endpoint(bluetooth::rfcomm::get(), channel )
97 );
98
99 send_data = new link_data();
100
101 // start accepting
102 start_accept();
103}
104
105void rfcomm::stop() {
106 logging_info( "Stopping asio rfcomm" );
107}
108
109void rfcomm::send(const address_v* remote, const uint8_t* data, size_t size) {
110
111 // get end-point
112 rfcomm_endpoint endpoint = *remote;
113 endpoint = convert(convert(endpoint));
114
115 // try to find established connector
116 logging_debug("Trying to find a already existing link.");
117 link_info* info = NULL;
118 for (size_t i = 0; i < links.size(); i++)
119 if (links[i]->remote.mac() == endpoint.mac()) {
120 logging_debug("Using already established link");
121 info = links[i];
122 break;
123 }
124
125 // not found, or not up? ->try to (re-)connect
126 if (info==NULL || !info->up || info->socket.is_open() ) {
127 logging_debug( "Connecting to " << endpoint.to_string() );
128 if (info != NULL && (!info->up || !info->socket.is_open())) {
129 logging_debug("Old link is down. Trying to re-establish link.");
130 } else {
131 info = new link_info(io);
132 }
133 info->connect_retries = 0;
134 info->remote = endpoint;
135 info->socket.async_connect( convert(endpoint), boost::bind(
136 &rfcomm::handle_connect, this,
137 boost::asio::placeholders::error, info
138 ));
139 asio_io_service::start();
140 }
141
142 // copy message
143 link_data ldata;
144 ldata.size = size;
145 ldata.size_[0] = (size >> 24) & 0xFF;
146 ldata.size_[1] = (size >> 16) & 0xFF;
147 ldata.size_[2] = (size >> 8) & 0xFF;
148 ldata.size_[3] = (size >> 0) & 0xFF;
149 ldata.buffer = new uint8_t[size];
150 memcpy(ldata.buffer, data, size);
151
152 // enqueue message
153 info->mutex.lock();
154 info->send_buffer.push_back(ldata);
155 info->mutex.unlock();
156
157 // start writing
158 start_write(info);
159}
160
161void rfcomm::send(const endpoint_set& endpoints, const uint8_t* data, size_t size) {
162 // send a message to each combination of mac-address and channel
163 BOOST_FOREACH( const mac_address mac, endpoints.bluetooth ) {
164 BOOST_FOREACH( const rfcomm_channel_address channel, endpoints.rfcomm ) {
165 rfcomm_endpoint endpoint(mac, channel);
166 address_vf vf = endpoint;
167 send(vf,data,size);
168 }
169 }
170}
171
172void rfcomm::terminate(const address_v* local, const address_v* remote) {
173 // get end-point
174 rfcomm_endpoint endpoint = *remote;
175
176 for (size_t i = 0; i < links.size(); i++)
177 if (links[i]->remote.mac() == endpoint.mac()) {
178 // close socket
179 shutdown(links[i]);
180 break;
181 }
182}
183
184void rfcomm::register_listener(transport_listener* listener) {
185 this->listener = listener;
186}
187
188void rfcomm::start_accept() {
189
190 logging_info( "Waiting for connections ..." );
191
192 // start accepting a connection
193 link_info* info = new link_info(io);
194 acceptor->async_accept(info->socket, boost::bind(
195 // bind parameters
196 &rfcomm::handle_accept, this,
197
198 // handler parameters
199 boost::asio::placeholders::error, info
200 ));
201 asio_io_service::start();
202}
203
204void rfcomm::handle_accept(const error_code& error, link_info* info) {
205 if (error) {
206 logging_error( "Error waiting for new connections. Error code: "<< error.message()
207 << ", trying to recover (attempt " << accept_retries << ")");
208
209 // restart accepting
210 if (accept_retries<3) {
211 accept_retries++;
212 start_accept();
213 } else
214 delete info;
215
216 return;
217 }
218
219 links_mutex.lock();
220
221 // convert endpoints
222 info->up = true;
223 info->local = convert( info->socket.local_endpoint() );
224 info->remote = convert( info->socket.remote_endpoint() );
225
226 logging_debug("Accepted incoming connection from "
227 << info->remote.to_string() );
228
229 // add to list
230 links.push_back(info);
231 links_mutex.unlock();
232
233 // start i/o
234 start_read(info);
235 start_write(info);
236
237 // restart accept
238 start_accept();
239}
240
241void rfcomm::handle_connect( const error_code& error, link_info* info ) {
242 if (error) {
243 logging_error( "Can not connect. Error code: "
244 << error.message() << " Retrying ... "
245 "(attempt " << info->connect_retries << ")" );
246
247 // do we retry this connection? yes->
248 if (info->connect_retries<3) {
249 // increase counter
250 info->connect_retries++;
251
252 // retry connection attempt
253 info->socket.async_connect( convert(info->remote), boost::bind(
254 &rfcomm::handle_connect, this,
255 boost::asio::placeholders::error, info
256 ));
257
258 } else { // no-> delete link and stop
259 return;
260 }
261 }
262
263 // convert endpoints
264 info->up = true;
265 info->local = convert( info->socket.local_endpoint() );
266 info->remote = convert( info->socket.remote_endpoint() );
267
268 logging_debug( "Connected to " << info->remote.to_string() );
269
270 // add to list
271 links_mutex.lock();
272 links.push_back(info);
273 links_mutex.unlock();
274
275 // start i/o
276 start_read(info);
277 start_write(info);
278}
279
280void rfcomm::start_read(link_info* info) {
281 // start read
282 boost::asio::async_read(info->socket,
283
284 // read size of packet
285 boost::asio::buffer(info->size_, 4),
286
287 // bind handler
288 boost::bind(
289
290 // bind parameters
291 &rfcomm::handle_read_header, this,
292
293 // handler parameters
294 placeholders::error, placeholders::bytes_transferred, info
295 )
296 );
297}
298
299void rfcomm::handle_read_header(const error_code& error, size_t bytes,
300 link_info* info) {
301
302 // handle error
303 if (error) {
304 logging_error("Failed to receive message payload. Error code: "
305 << error.message() );
306 shutdown(info);
307 return;
308 }
309
310 // ignore errors and wait for all data to be received
311 if (bytes != 4) return;
312
313 // get size
314 info->size = (info->size_[0]<<24) + (info->size_[1] << 16) +
315 (info->size_[2]<< 8) + (info->size_[3] << 0);
316
317 logging_debug( "Message header received -> receive message of size " << info->size );
318
319 // allocate buffer
320 info->buffer = new uint8_t[info->size];
321
322 // start read
323 boost::asio::async_read(info->socket,
324 // read size of packet
325 boost::asio::buffer(info->buffer, info->size),
326 // bind handler
327 boost::bind(
328 // bind parameters
329 &rfcomm::handle_read_data, this,
330 // handler parameters
331 placeholders::error, placeholders::bytes_transferred, info
332 )
333 );
334}
335
336void rfcomm::handle_read_data(const error_code& error, size_t bytes,
337 link_info* info) {
338
339 // check error
340 if (error) {
341 logging_error("Failed to receive message payload. Error code: "
342 << error.message() );
343 shutdown(info);
344 return;
345 }
346
347 // wait for all data to be received
348 if (bytes != info->size)
349 return;
350
351 logging_debug( "Received message of size " << info->size );
352
353 // deliver data
354 listener->receive_message(this, info->local, info->remote, info->buffer, info->size );
355
356 // free buffers and reset size buffer
357 delete [] info->buffer;
358 for (size_t i=0; i<4; i++) info->size_[i] = 0;
359
360 start_read(info);
361}
362
363void rfcomm::start_write( link_info* info ) {
364 boost::mutex::scoped_lock(info->mutex);
365
366 // do not start writing if sending is in progress
367 if (info->sending || !info->up || info->send_buffer.size()==0) return;
368
369 logging_debug("Sending messages ...");
370
371 // safely remove data from deque
372 *send_data = info->send_buffer.front();
373 info->send_buffer.pop_front();
374
375 boost::array<boost::asio::mutable_buffer, 2> buffer;
376 buffer[0] = boost::asio::buffer(send_data->size_,4);
377 buffer[1] = boost::asio::buffer(send_data->buffer,send_data->size);
378
379 // start writing
380 boost::asio::async_write(info->socket,
381 // read size of packet
382 buffer,
383 // bind handler
384 boost::bind(
385 // bind parameters
386 &rfcomm::handle_write_data, this,
387 // handler parameters
388 placeholders::error, placeholders::bytes_transferred,
389 info, send_data->size, send_data->buffer
390 )
391 );
392}
393
394void rfcomm::handle_write_data(const error_code& error, size_t bytes,
395 link_info* info, size_t size, uint8_t* buffer ) {
396
397 // handle error
398 if (error) {
399 logging_error( "Message sent error. Error code: "<< error.message() );
400 shutdown(info);
401 return;
402 }
403
404 // wait for all data to be sent
405 if (bytes != (size+4) )
406 return;
407
408 logging_debug( "Message sent" );
409
410 // free buffer
411 delete [] buffer;
412
413 // restart-write
414 start_write(info);
415}
416
417}} // namespace ariba::transport
Note: See TracBrowser for help on using the repository browser.