streaming socket on the stack

server/stream_server.cpp

@@ -764,34 +764,39 @@ session_ptr StreamServer::getStreamSession(const std::string& clientId) const
 
 void StreamServer::startAccept()
 {
+    auto accept_handler = [this](error_code ec, tcp::socket socket) {
+        if (!ec)
+            handleAccept(std::move(socket));
+        else
+            LOG(ERROR) << "Error while accepting socket connection: " << ec.message() << "\n";
+    };
+
     if (acceptor_v4_)
     {
-        socket_ptr socket_v4 = make_shared<tcp::socket>(*io_context_);
+        acceptor_v4_->async_accept(accept_handler);
-        acceptor_v4_->async_accept(*socket_v4, bind(&StreamServer::handleAccept, this, socket_v4));
     }
     if (acceptor_v6_)
     {
-        socket_ptr socket_v6 = make_shared<tcp::socket>(*io_context_);
+        acceptor_v6_->async_accept(accept_handler);
-        acceptor_v6_->async_accept(*socket_v6, bind(&StreamServer::handleAccept, this, socket_v6));
     }
 }
 
 
-void StreamServer::handleAccept(socket_ptr socket)
+void StreamServer::handleAccept(tcp::socket socket)
 {
     try
     {
         struct timeval tv;
         tv.tv_sec = 5;
         tv.tv_usec = 0;
-        setsockopt(socket->native_handle(), SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof(tv));
+        setsockopt(socket.native_handle(), SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof(tv));
-        setsockopt(socket->native_handle(), SOL_SOCKET, SO_SNDTIMEO, &tv, sizeof(tv));
+        setsockopt(socket.native_handle(), SOL_SOCKET, SO_SNDTIMEO, &tv, sizeof(tv));
 
         /// experimental: turn on tcp::no_delay
-        socket->set_option(tcp::no_delay(true));
+        socket.set_option(tcp::no_delay(true));
 
-        SLOG(NOTICE) << "StreamServer::NewConnection: " << socket->remote_endpoint().address().to_string() << endl;
+        SLOG(NOTICE) << "StreamServer::NewConnection: " << socket.remote_endpoint().address().to_string() << endl;
-        shared_ptr<StreamSession> session = make_shared<StreamSession>(this, socket);
+        shared_ptr<StreamSession> session = make_shared<StreamSession>(this, std::move(socket));
 
         session->setBufferMs(settings_.stream.bufferMs);
         session->start();

server/stream_server.hpp

@@ -78,7 +78,7 @@ public:
 
 private:
     void startAccept();
-    void handleAccept(socket_ptr socket);
+    void handleAccept(tcp::socket socket);
     session_ptr getStreamSession(const std::string& mac) const;
     session_ptr getStreamSession(StreamSession* session) const;
     void ProcessRequest(const jsonrpcpp::request_ptr request, jsonrpcpp::entity_ptr& response, jsonrpcpp::notification_ptr& notification) const;

server/stream_session.cpp

@@ -27,10 +27,9 @@ using namespace std;
 
 
 
-StreamSession::StreamSession(MessageReceiver* receiver, std::shared_ptr<tcp::socket> socket)
+StreamSession::StreamSession(MessageReceiver* receiver, tcp::socket&& socket)
-    : active_(false), readerThread_(nullptr), writerThread_(nullptr), messageReceiver_(receiver), pcmStream_(nullptr)
+    : active_(false), readerThread_(nullptr), writerThread_(nullptr), socket_(std::move(socket)), messageReceiver_(receiver), pcmStream_(nullptr)
 {
-    socket_ = socket;
 }
 
 
@@ -76,16 +75,12 @@ void StreamSession::stop()
     try
     {
         boost::system::error_code ec;
-        if (socket_)
+        socket_.shutdown(boost::asio::ip::tcp::socket::shutdown_both, ec);
-        {
+        if (ec)
-            std::lock_guard<std::mutex> socketLock(socketMutex_);
+            LOG(ERROR) << "Error in socket shutdown: " << ec.message() << "\n";
-            socket_->shutdown(boost::asio::ip::tcp::socket::shutdown_both, ec);
+        socket_.close(ec);
-            if (ec)
+        if (ec)
-                LOG(ERROR) << "Error in socket shutdown: " << ec.message() << "\n";
+            LOG(ERROR) << "Error in socket close: " << ec.message() << "\n";
-            socket_->close(ec);
-            if (ec)
-                LOG(ERROR) << "Error in socket close: " << ec.message() << "\n";
-        }
         if (readerThread_ && readerThread_->joinable())
         {
             LOG(DEBUG) << "StreamSession joining readerThread\n";
@@ -104,7 +99,6 @@ void StreamSession::stop()
 
     readerThread_ = nullptr;
     writerThread_ = nullptr;
-    socket_ = nullptr;
     LOG(DEBUG) << "StreamSession stopped\n";
 }
 
@@ -114,7 +108,7 @@ void StreamSession::socketRead(void* _to, size_t _bytes)
     size_t read = 0;
     do
     {
-        read += socket_->read_some(boost::asio::buffer((char*)_to + read, _bytes - read));
+        read += socket_.read_some(boost::asio::buffer((char*)_to + read, _bytes - read));
     } while (active_ && (read < _bytes));
 }
 
@@ -147,22 +141,20 @@ void StreamSession::setBufferMs(size_t bufferMs)
 }
 
 
-bool StreamSession::send(const msg::message_ptr& message) const
+bool StreamSession::send(const msg::message_ptr& message)
 {
     // TODO on exception: set active = false
     //	LOG(INFO) << "send: " << message->type << ", size: " << message->getSize() << ", id: " << message->id << ", refers: " << message->refersTo << "\n";
-    std::lock_guard<std::mutex> socketLock(socketMutex_);
+    std::lock_guard<std::mutex> activeLock(activeMutex_);
-    {
+    if (!active_)
-        std::lock_guard<std::mutex> activeLock(activeMutex_);
+        return false;
-        if (!socket_ || !active_)
+
-            return false;
-    }
     boost::asio::streambuf streambuf;
     std::ostream stream(&streambuf);
     tv t;
     message->sent = t;
     message->serialize(stream);
-    boost::asio::write(*socket_.get(), streambuf);
+    boost::asio::write(socket_, streambuf);
     //	LOG(INFO) << "done: " << message->type << ", size: " << message->size << ", id: " << message->id << ", refers: " << message->refersTo << "\n";
     return true;
 }
@@ -193,10 +185,8 @@ void StreamSession::getNextMessage()
     // baseMessage.refersTo << "\n";
     if (baseMessage.size > buffer.size())
         buffer.resize(baseMessage.size);
-    //	{
+
-    //		std::lock_guard<std::mutex> socketLock(socketMutex_);
     socketRead(&buffer[0], baseMessage.size);
-    //	}
     tv t;
     baseMessage.received = t;
 

server/stream_session.hpp

@@ -57,13 +57,13 @@ class StreamSession
 {
 public:
     /// ctor. Received message from the client are passed to MessageReceiver
-    StreamSession(MessageReceiver* receiver, std::shared_ptr<tcp::socket> socket);
+    StreamSession(MessageReceiver* receiver, tcp::socket&& socket);
     ~StreamSession();
     void start();
     void stop();
 
     /// Sends a message to the client (synchronous)
-    bool send(const msg::message_ptr& message) const;
+    bool send(const msg::message_ptr& message);
 
     /// Sends a message to the client (asynchronous)
     void sendAsync(const msg::message_ptr& message, bool sendNow = false);
@@ -77,7 +77,7 @@ public:
 
     std::string getIP()
     {
-        return socket_->remote_endpoint().address().to_string();
+        return socket_.remote_endpoint().address().to_string();
     }
 
     void setPcmStream(PcmStreamPtr pcmStream);
@@ -95,7 +95,7 @@ protected:
     std::unique_ptr<std::thread> readerThread_;
     std::unique_ptr<std::thread> writerThread_;
     mutable std::mutex socketMutex_;
-    std::shared_ptr<tcp::socket> socket_;
+    tcp::socket socket_;
     MessageReceiver* messageReceiver_;
     Queue<std::shared_ptr<msg::BaseMessage>> messages_;
     size_t bufferMs_;