asio UDP sample

#include <cstdlib>
#include <iostream>
#include <boost/bind.hpp>
#include <boost/asio.hpp>
#include <boost/thread.hpp>
#include <boost/date_time.hpp>
#include <boost/foreach.hpp>

using boost::asio::ip::udp;

class server
{
public:
  server(boost::asio::io_service& io_service, short port)
    : io_service_(io_service),
      socket_(io_service, udp::endpoint(udp::v4(), port)),
      messages_sent_(0u),
      should_run_(true)
  {
    boost::thread   t(boost::bind(&server::thread_func, this));
    t_.swap(t);

    socket_.async_receive_from(
        boost::asio::buffer(data_, max_length), sender_endpoint_,
        boost::bind(&server::handle_receive_from, this,
          boost::asio::placeholders::error,
          boost::asio::placeholders::bytes_transferred));
  }

  ~server() {
      should_run_ = false;
      t_.join();
  }

  void handle_receive_from(const boost::system::error_code& error,
      size_t bytes_recvd)
  {
    if (error) {
        std::cout << "receive error!" << std::endl;
        return;
    }

      std::string  message(data_, bytes_recvd);

      // start receiving again
      socket_.async_receive_from(
          boost::asio::buffer(data_, max_length), sender_endpoint_,
          boost::bind(&server::handle_receive_from, this,
            boost::asio::placeholders::error,
            boost::asio::placeholders::bytes_transferred));

      // store the received message in a vector
      boost::lock_guard<boost::recursive_mutex>    lock(mutex_);
      messages_.push_back(message);
      messages_received_++;
  }

  void send(udp::endpoint other_end, const std::string &message) {
      unsigned int sent = socket_.send_to(
                        boost::asio::buffer(message.c_str(), message.size()),
                        other_end);

      if (sent < message.size()) {
          std::cout << "only " << sent << " bytes sent from " << message.size()
                    << std::endl;
      }

      ++messages_sent_;
  }

  void thread_func() {
      while (should_run_) {
          boost::this_thread::yield();
          boost::this_thread::sleep(boost::posix_time::milliseconds(100u));

          // work on the messages received
          boost::lock_guard<boost::recursive_mutex>    lock(mutex_);
          BOOST_FOREACH(const std::string & str, messages_) {
            boost::this_thread::sleep(boost::posix_time::milliseconds(1u));
          }
          messages_.clear();
      }
  }


  std::vector<std::string>  messages_;
  unsigned int messages_sent_;
  unsigned int messages_received_;

private:
  boost::asio::io_service& io_service_;
  udp::socket socket_;
  udp::endpoint sender_endpoint_;
  enum { max_length = 1024 };
  char data_[max_length];
  boost::recursive_mutex  mutex_;
  bool should_run_;
  boost::thread t_;
};

static boost::asio::io_service io_service;
static bool                    should_run;

void service_thread(void) {
    while (should_run) {
        io_service.run();

        io_service.reset();
    }
}


int main(int argc, char* argv[])
{
  try
  {
    server a(io_service, 4567);
    server b(io_service, 4568);

    should_run = true;

    boost::thread   t(service_thread);

    // generate an endpoint for 'a'
    udp::resolver           resolver(io_service);
    udp::resolver::query    query(udp::v4(), "localhost", "4567");
    udp::endpoint           endp_a = *resolver.resolve(query);

    for (unsigned int i = 0; i < 20000; ++i) {
        std::stringstream   sstr;
        for (unsigned int j = 0; j < 1000; ++j) {
            sstr << i;
        }
        b.send(endp_a, sstr.str());

        if (i % 50 == 0) {
            //boost::this_thread::yield();
            //boost::this_thread::sleep(boost::posix_time::milliseconds(1));
        }
    }

    boost::this_thread::sleep(boost::posix_time::milliseconds(1000));

    std::cout << "b sent " << b.messages_sent_ << " messages"
              << std::endl;
    std::cout << "a received " << a.messages_received_ << " messages"
              << std::endl;

    // stop the listening thread
    should_run = false;
    io_service.stop();
    t.join();
  }
  catch (std::exception& e)
  {
    std::cerr << "Exception: " << e.what() << "\n";
  }

  return 0;
}