Add experimental TCP and UDP streams

server/CMakeLists.txt

@@ -12,6 +12,8 @@ set(SERVER_SOURCES
     streamreader/stream_uri.cpp
     streamreader/stream_manager.cpp
     streamreader/pcm_stream.cpp
+    streamreader/tcp_stream.cpp
+    streamreader/udp_stream.cpp
     streamreader/pipe_stream.cpp
     streamreader/file_stream.cpp
     streamreader/airplay_stream.cpp

server/streamreader/pcm_stream.cpp

@@ -170,9 +170,7 @@ json PcmStream::toJson() const
         state = "disabled";
 
     json j = {
-        {"uri", uri_.toJson()},
+        {"uri", uri_.toJson()}, {"id", getId()}, {"status", state},
-        {"id", getId()},
-        {"status", state},
     };
 
     if (meta_)

server/streamreader/pcm_stream.hpp

@@ -97,7 +97,7 @@ protected:
     std::thread thread_;
     std::atomic<bool> active_;
 
-    virtual void worker() = 0;
+    virtual void worker(){};
     virtual bool sleep(int32_t ms);
     void setState(const ReaderState& newState);
 

server/streamreader/stream_manager.cpp

@@ -26,12 +26,15 @@
 #include "librespot_stream.hpp"
 #include "pipe_stream.hpp"
 #include "process_stream.hpp"
+#include "tcp_stream.hpp"
+#include "udp_stream.hpp"
 
 
 using namespace std;
 
 
-StreamManager::StreamManager(PcmListener* pcmListener, boost::asio::io_context& ioc, const std::string& defaultSampleFormat, const std::string& defaultCodec, size_t defaultReadBufferMs)
+StreamManager::StreamManager(PcmListener* pcmListener, boost::asio::io_context& ioc, const std::string& defaultSampleFormat, const std::string& defaultCodec,
+                             size_t defaultReadBufferMs)
     : pcmListener_(pcmListener), sampleFormat_(defaultSampleFormat), codec_(defaultCodec), readBufferMs_(defaultReadBufferMs), ioc_(ioc)
 {
 }
@@ -77,6 +80,14 @@ PcmStreamPtr StreamManager::addStream(const std::string& uri)
     {
         stream = make_shared<AirplayStream>(pcmListener_, ioc_, streamUri);
     }
+    else if (streamUri.scheme == "experimental.tcp")
+    {
+        stream = make_shared<TcpStream>(pcmListener_, ioc_, streamUri);
+    }
+    else if (streamUri.scheme == "experimental.udp")
+    {
+        stream = make_shared<UdpStream>(pcmListener_, ioc_, streamUri);
+    }
     else
     {
         throw SnapException("Unknown stream type: " + streamUri.scheme);

server/streamreader/stream_manager.hpp

@@ -2,17 +2,18 @@
 #define PCM_READER_FACTORY_H
 
 #include "pcm_stream.hpp"
+#include <boost/asio/io_context.hpp>
 #include <memory>
 #include <string>
 #include <vector>
-#include <boost/asio/io_context.hpp>
 
 typedef std::shared_ptr<PcmStream> PcmStreamPtr;
 
 class StreamManager
 {
 public:
-    StreamManager(PcmListener* pcmListener, boost::asio::io_context& ioc, const std::string& defaultSampleFormat, const std::string& defaultCodec, size_t defaultReadBufferMs = 20);
+    StreamManager(PcmListener* pcmListener, boost::asio::io_context& ioc, const std::string& defaultSampleFormat, const std::string& defaultCodec,
+                  size_t defaultReadBufferMs = 20);
 
     PcmStreamPtr addStream(const std::string& uri);
     void removeStream(const std::string& name);

server/streamreader/tcp_stream.cpp

@@ -0,0 +1,140 @@
+/***
+    This file is part of snapcast
+    Copyright (C) 2014-2019  Johannes Pohl
+
+    This program is free software: you can redistribute it and/or modify
+    it under the terms of the GNU General Public License as published by
+    the Free Software Foundation, either version 3 of the License, or
+    (at your option) any later version.
+
+    This program is distributed in the hope that it will be useful,
+    but WITHOUT ANY WARRANTY; without even the implied warranty of
+    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+    GNU General Public License for more details.
+
+    You should have received a copy of the GNU General Public License
+    along with this program.  If not, see <http://www.gnu.org/licenses/>.
+***/
+
+#include <cerrno>
+#include <fcntl.h>
+#include <memory>
+#include <sys/stat.h>
+#include <unistd.h>
+
+#include "common/aixlog.hpp"
+#include "common/snap_exception.hpp"
+#include "common/str_compat.hpp"
+#include "encoder/encoder_factory.hpp"
+#include "tcp_stream.hpp"
+
+
+using namespace std;
+
+
+
+TcpStream::TcpStream(PcmListener* pcmListener, boost::asio::io_context& ioc, const StreamUri& uri) : PcmStream(pcmListener, ioc, uri), timer_(ioc)
+{
+    size_t port = 4953;
+    try
+    {
+        port = cpt::stoi(uri_.getQuery("port", cpt::to_string(port)));
+    }
+    catch (...)
+    {
+    }
+
+    LOG(INFO) << "TcpStream port: " << port << "\n";
+    acceptor_ = make_unique<tcp::acceptor>(ioc_, tcp::endpoint(tcp::v4(), port));
+    chunk_ = make_unique<msg::PcmChunk>(sampleFormat_, pcmReadMs_);
+    LOG(DEBUG) << "Chunk size: " << chunk_->payloadSize << "\n";
+}
+
+
+TcpStream::~TcpStream()
+{
+}
+
+
+void TcpStream::start()
+{
+    encoder_->init(this, sampleFormat_);
+    active_ = true;
+    do_accept();
+}
+
+
+void TcpStream::do_accept()
+{
+    auto self = shared_from_this();
+    acceptor_->async_accept([this, self](boost::system::error_code ec, tcp::socket socket) {
+        if (!ec)
+        {
+            LOG(DEBUG) << "New client connection\n";
+            socket_ = make_unique<tcp::socket>(move(socket));
+            tv_chunk_.tv_sec = 0;
+            first_ = true;
+            do_read();
+        }
+        else
+        {
+            LOG(ERROR) << "Accept failed: " << ec.message() << "\n";
+        }
+    });
+}
+
+
+void TcpStream::do_read()
+{
+    auto self = shared_from_this();
+    chunk_->timestamp.sec = tv_chunk_.tv_sec;
+    chunk_->timestamp.usec = tv_chunk_.tv_usec;
+    boost::asio::async_read(*socket_, boost::asio::buffer(chunk_->payload, chunk_->payloadSize),
+                            [this, self](boost::system::error_code ec, std::size_t length) mutable {
+                                if (ec)
+                                {
+                                    LOG(ERROR) << "Error reading message: " << ec.message() << "\n";
+                                    do_accept();
+                                    return;
+                                }
+                                LOG(DEBUG) << "Read: " << length << " bytes\n";
+                                if (first_)
+                                {
+                                    first_ = false;
+                                    chronos::systemtimeofday(&tv_chunk_);
+                                    chunk_->timestamp.sec = tv_chunk_.tv_sec;
+                                    chunk_->timestamp.usec = tv_chunk_.tv_usec;
+                                    tvEncodedChunk_ = tv_chunk_;
+                                    nextTick_ = chronos::getTickCount();
+                                }
+                                encoder_->encode(chunk_.get());
+                                nextTick_ += pcmReadMs_;
+                                chronos::addUs(tv_chunk_, pcmReadMs_ * 1000);
+                                long currentTick = chronos::getTickCount();
+
+                                if (nextTick_ >= currentTick)
+                                {
+                                    setState(kPlaying);
+                                    timer_.expires_from_now(boost::posix_time::milliseconds(nextTick_ - currentTick));
+                                    timer_.async_wait([self, this](const boost::system::error_code& ec) {
+                                        if (ec)
+                                        {
+                                            LOG(ERROR) << "Error during async wait: " << ec.message() << "\n";
+                                        }
+                                        else
+                                        {
+                                            do_read();
+                                        }
+                                    });
+                                    return;
+                                }
+                                else
+                                {
+                                    chronos::systemtimeofday(&tv_chunk_);
+                                    tvEncodedChunk_ = tv_chunk_;
+                                    pcmListener_->onResync(this, currentTick - nextTick_);
+                                    nextTick_ = currentTick;
+                                    do_read();
+                                }
+                            });
+}

server/streamreader/tcp_stream.hpp

@@ -0,0 +1,56 @@
+/***
+    This file is part of snapcast
+    Copyright (C) 2014-2019  Johannes Pohl
+
+    This program is free software: you can redistribute it and/or modify
+    it under the terms of the GNU General Public License as published by
+    the Free Software Foundation, either version 3 of the License, or
+    (at your option) any later version.
+
+    This program is distributed in the hope that it will be useful,
+    but WITHOUT ANY WARRANTY; without even the implied warranty of
+    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+    GNU General Public License for more details.
+
+    You should have received a copy of the GNU General Public License
+    along with this program.  If not, see <http://www.gnu.org/licenses/>.
+***/
+
+#ifndef TCP_STREAM_H
+#define TCP_STREAM_H
+
+#include "pcm_stream.hpp"
+#include <boost/asio.hpp>
+#include <memory>
+
+using boost::asio::ip::tcp;
+
+
+/// Reads and decodes PCM data from a named pipe
+/**
+ * Reads PCM from a named pipe and passes the data to an encoder.
+ * Implements EncoderListener to get the encoded data.
+ * Data is passed to the PcmListener
+ */
+class TcpStream : public PcmStream, public std::enable_shared_from_this<TcpStream>
+{
+public:
+    /// ctor. Encoded PCM data is passed to the PipeListener
+    TcpStream(PcmListener* pcmListener, boost::asio::io_context& ioc, const StreamUri& uri);
+    ~TcpStream() override;
+
+    void start() override;
+
+protected:
+    void do_accept() override;
+    void do_read() override;
+    std::unique_ptr<tcp::acceptor> acceptor_;
+    std::unique_ptr<tcp::socket> socket_;
+    timeval tv_chunk_;
+    bool first_;
+    long nextTick_;
+    boost::asio::deadline_timer timer_;
+};
+
+
+#endif

server/streamreader/udp_stream.cpp

@@ -0,0 +1,97 @@
+/***
+    This file is part of snapcast
+    Copyright (C) 2014-2019  Johannes Pohl
+
+    This program is free software: you can redistribute it and/or modify
+    it under the terms of the GNU General Public License as published by
+    the Free Software Foundation, either version 3 of the License, or
+    (at your option) any later version.
+
+    This program is distributed in the hope that it will be useful,
+    but WITHOUT ANY WARRANTY; without even the implied warranty of
+    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+    GNU General Public License for more details.
+
+    You should have received a copy of the GNU General Public License
+    along with this program.  If not, see <http://www.gnu.org/licenses/>.
+***/
+
+#include <cerrno>
+#include <fcntl.h>
+#include <memory>
+#include <sys/stat.h>
+#include <unistd.h>
+
+#include "common/aixlog.hpp"
+#include "common/snap_exception.hpp"
+#include "common/str_compat.hpp"
+#include "encoder/encoder_factory.hpp"
+#include "udp_stream.hpp"
+
+
+using namespace std;
+
+
+
+UdpStream::UdpStream(PcmListener* pcmListener, boost::asio::io_context& ioc, const StreamUri& uri) : PcmStream(pcmListener, ioc, uri)
+{
+    size_t port = 5004;
+    try
+    {
+        port = cpt::stoi(uri_.getQuery("port", cpt::to_string(port)));
+    }
+    catch (...)
+    {
+    }
+
+    LOG(INFO) << "UdpStream port: " << port << "\n";
+    socket_ = make_unique<udp::socket>(ioc_, udp::endpoint(udp::v4(), port));
+    chunk_ = make_unique<msg::PcmChunk>(sampleFormat_, pcmReadMs_);
+    LOG(DEBUG) << "Chunk size: " << chunk_->payloadSize << "\n";
+}
+
+
+UdpStream::~UdpStream()
+{
+}
+
+
+void UdpStream::start()
+{
+    encoder_->init(this, sampleFormat_);
+    active_ = true;
+    tv_chunk_.tv_sec = 0;
+    do_read();
+}
+
+
+void UdpStream::do_read()
+{
+    auto self = shared_from_this();
+    if (tv_chunk_.tv_sec != 0)
+    {
+        chunk_->timestamp.sec = tv_chunk_.tv_sec;
+        chunk_->timestamp.usec = tv_chunk_.tv_usec;
+        //        tvEncodedChunk_ = tv_chunk_;
+        chronos::addUs(tv_chunk_, pcmReadMs_ * 1000);
+        //        chronos::systemtimeofday(&tv_chunk_);
+    }
+    socket_->async_receive_from(boost::asio::buffer(chunk_->payload, chunk_->payloadSize), sender_endpoint_,
+                                [this, self](boost::system::error_code ec, std::size_t bytes_recvd) {
+                                    if (ec)
+                                    {
+                                        LOG(ERROR) << "Error reading message: " << ec.message() << "\n";
+                                        return;
+                                    }
+                                    LOG(DEBUG) << "Read: " << bytes_recvd << " bytes\n";
+                                    if (tv_chunk_.tv_sec == 0)
+                                    {
+                                        chronos::systemtimeofday(&tv_chunk_);
+                                        chunk_->timestamp.sec = tv_chunk_.tv_sec;
+                                        chunk_->timestamp.usec = tv_chunk_.tv_usec;
+                                        tvEncodedChunk_ = tv_chunk_;
+                                    }
+                                    // encoder_->encode(chunk_.get());
+                                    do_read();
+                                });
+}

server/streamreader/udp_stream.hpp

@@ -0,0 +1,53 @@
+/***
+    This file is part of snapcast
+    Copyright (C) 2014-2019  Johannes Pohl
+
+    This program is free software: you can redistribute it and/or modify
+    it under the terms of the GNU General Public License as published by
+    the Free Software Foundation, either version 3 of the License, or
+    (at your option) any later version.
+
+    This program is distributed in the hope that it will be useful,
+    but WITHOUT ANY WARRANTY; without even the implied warranty of
+    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+    GNU General Public License for more details.
+
+    You should have received a copy of the GNU General Public License
+    along with this program.  If not, see <http://www.gnu.org/licenses/>.
+***/
+
+#ifndef UDP_STREAM_HPP
+#define UDP_STREAM_HPP
+
+#include "pcm_stream.hpp"
+#include <boost/asio.hpp>
+#include <memory>
+
+using boost::asio::ip::udp;
+
+
+/// Reads and decodes PCM data from a named pipe
+/**
+ * Reads PCM from a named pipe and passes the data to an encoder.
+ * Implements EncoderListener to get the encoded data.
+ * Data is passed to the PcmListener
+ */
+class UdpStream : public PcmStream, public std::enable_shared_from_this<UdpStream>
+{
+public:
+    /// ctor. Encoded PCM data is passed to the PipeListener
+    UdpStream(PcmListener* pcmListener, boost::asio::io_context& ioc, const StreamUri& uri);
+    ~UdpStream() override;
+
+    void start() override;
+
+protected:
+    void do_read();
+    std::unique_ptr<udp::socket> socket_;
+    std::unique_ptr<msg::PcmChunk> chunk_;
+    timeval tv_chunk_;
+    udp::endpoint sender_endpoint_;
+};
+
+
+#endif