Line data    Source code

       1             : /*
       2             :  * Copyright (c) 2016 Juniper Networks, Inc. All rights reserved.
       3             :  */
       4             : 
       5             : /*
       6             :  * The primary method implemented here is Send(), to transmit a
       7             :  * message over the Unix socket. It uses boost::asio::async_write to
       8             :  * send one message at a time over the socket, that is transmitted
       9             :  * asynchronously. The user can repeatedly call Send(). All those
      10             :  * buffers are tail-queued. Upon write_complete callback, the next
      11             :  * message from the front of the queue is sent.
      12             :  */
      13             : #include "io/usock_server.h"
      14             : 
      15             : using boost::asio::buffer_cast;
      16             : using boost::asio::buffer;
      17             : using boost::asio::mutable_buffer;
      18             : 
      19           2 : UnixDomainSocketSession::~UnixDomainSocketSession() {
      20           2 :     if (observer_) {
      21           0 :         observer_(this, CLOSE);
      22             :     }
      23             : 
      24             :     /* Free up any remaining buffers in the queue. */
      25           2 :     for (BufferQueue::iterator iter = buffer_queue_.begin();
      26           2 :          iter != buffer_queue_.end(); ++iter) {
      27           0 :         DeleteBuffer(*iter);
      28             :     }
      29           2 :     buffer_queue_.clear();
      30           2 : }
      31             : 
      32           1 : void UnixDomainSocketSession::Start() {
      33           1 :     if (observer_) {
      34           0 :         observer_(this, READY);
      35             :     }
      36             : 
      37           1 :     socket_.async_read_some(boost::asio::buffer(data_),
      38           2 :                             boost::bind(&UnixDomainSocketSession::
      39           2 :                                         HandleRead, shared_from_this(),
      40             :                                         boost::asio::placeholders::error,
      41             :                                         boost::asio::placeholders::
      42             :                                         bytes_transferred));
      43           1 : }
      44             : 
      45           1 : void UnixDomainSocketSession::Send(const uint8_t * data, int data_len) {
      46           1 :     if (!data || !data_len) {
      47           0 :         return;
      48             :     }
      49           1 :     bool write_now = buffer_queue_.empty();
      50           1 :     AppendBuffer(data, data_len);
      51           1 :     if (write_now) {
      52           1 :         WriteToSocket();
      53             :     }
      54             : }
      55             : 
      56           2 : void UnixDomainSocketSession::WriteToSocket() {
      57           2 :     if (buffer_queue_.empty()) {
      58           1 :         return;
      59             :     }
      60             : 
      61           1 :     boost::asio::mutable_buffer head = buffer_queue_.front();
      62           1 :     boost::asio::async_write(socket_,
      63           1 :                              buffer(buffer_cast <const uint8_t *>(head),
      64             :                                     boost::asio::buffer_size(head)),
      65           2 :                              boost::bind(&UnixDomainSocketSession::
      66           2 :                                          HandleWrite, shared_from_this(),
      67             :                                          boost::asio::placeholders::error));
      68             : }
      69             : 
      70           1 : void UnixDomainSocketSession::AppendBuffer(const uint8_t *src, int bytes) {
      71           1 :     u_int8_t *data = new u_int8_t[bytes];
      72           1 :     memcpy(data, src, bytes);
      73             :     boost::asio::mutable_buffer buffer =
      74           1 :         boost::asio::mutable_buffer(data, bytes);
      75           1 :     buffer_queue_.push_back(buffer);
      76           1 : }
      77             : 
      78           1 : void UnixDomainSocketSession::DeleteBuffer(boost::asio::mutable_buffer buffer) {
      79           1 :     const uint8_t *data = buffer_cast <const uint8_t *>(buffer);
      80           1 :     delete []data;
      81           1 :     return;
      82             : }
      83             : 
      84           1 : void UnixDomainSocketSession::HandleRead(const boost::system::error_code &error,
      85             :                                          size_t bytes_transferred) {
      86           1 :     if (error) {
      87           1 :         return;
      88             :     }
      89           0 :     if (observer_) {
      90           0 :         observer_(this, READY);
      91             :     }
      92             : }
      93             : 
      94           1 : void UnixDomainSocketSession::HandleWrite(
      95             :         const boost::system::error_code &error) {
      96             :     /*
      97             :      * async_write() is atomic in that it returns success once the entire message
      98             :      * is sent. If there is an error, it's okay to return from here so that the
      99             :      * session gets closed.
     100             :      */
     101           1 :     if (error) {
     102           0 :         return;
     103             :     }
     104             : 
     105             :     /*
     106             :      * We are done with the buffer at the head of the queue. Delete it.
     107             :      */
     108           1 :     DeleteBuffer(buffer_queue_.front());
     109           1 :     buffer_queue_.pop_front();
     110             : 
     111             :     /*
     112             :      * Write the next message, if there.
     113             :      */
     114           1 :     WriteToSocket();
     115             : 
     116             :     /*
     117             :      * Engage on the socket to keep it alive.
     118             :      */
     119           1 :     socket_.async_read_some(boost::asio::buffer(data_),
     120           2 :                             boost::bind(&UnixDomainSocketSession::
     121           2 :                                         HandleRead, shared_from_this(),
     122             :                                         boost::asio::placeholders::error,
     123             :                                         boost::asio::placeholders::
     124             :                                         bytes_transferred));
     125             : }
     126             : 
     127           1 : UnixDomainSocketServer::UnixDomainSocketServer(
     128           1 :         boost::asio::io_context *io, const std::string &file)
     129           1 :   : io_service_(io),
     130           1 :     acceptor_(*io, boost::asio::local::stream_protocol::endpoint(file)),
     131           1 :     session_idspace_(0) {
     132           1 :     SessionPtr new_session(new UnixDomainSocketSession(io_service_));
     133           1 :     acceptor_.async_accept(new_session->socket(),
     134           2 :                            boost::bind(&UnixDomainSocketServer::
     135             :                                        HandleAccept, this, new_session,
     136             :                                        boost::asio::placeholders::error));
     137           1 : }
     138             : 
     139             : void
     140           1 : UnixDomainSocketServer::HandleAccept(SessionPtr session,
     141             :                                      const boost::system::error_code &error) {
     142           1 :     UnixDomainSocketSession *socket_session = session.get();
     143             : 
     144           1 :     if (error) {
     145           0 :         if (observer_) {
     146           0 :             observer_(this, socket_session, DELETE_SESSION);
     147             :         }
     148           0 :         return;
     149             :     }
     150             : 
     151           1 :     socket_session->set_session_id(++session_idspace_);
     152           1 :     if (observer_) {
     153           1 :         observer_(this, socket_session, NEW_SESSION);
     154           1 :         session->Start();
     155             :     }
     156             : 
     157           1 :     SessionPtr new_session(new UnixDomainSocketSession(io_service_));
     158           1 :     acceptor_.async_accept(new_session->socket(),
     159           2 :                            boost::bind(&UnixDomainSocketServer::
     160             :                                        HandleAccept, this, new_session,
     161             :                                        boost::asio::placeholders::error));
     162           1 : }