Commit 11ef762bb9997032889253ccb1437d8f0029936d - snapcast

+1

-1

CMakeLists.txt less more

0	0	cmake_minimum_required(VERSION 3.2)
1	1
2		project(snapcast LANGUAGES CXX VERSION 0.18.0)
	2	project(snapcast LANGUAGES CXX VERSION 0.18.1)
3	3	set(PROJECT_DESCRIPTION "Multi-room client-server audio player")
4	4	set(PROJECT_URL "https://github.com/badaix/snapcast")
5	5

+11

-1

changelog.md less more

0	0	# Snapcast changelog
1	1
	2	## Version 0.18.1
	3
	4	### Bugfixes
	5
	6	- Fix random server crash or deadlock during stream client disconnect
	7	- Fix random server crash or deadlock during control client disconnect
	8	- Fix airplay stream buffer allocation (PR #536)
	9
	10	_Johannes Pohl <snapcast@badaix.de> Tue, 28 Jan 2020 00:13:37 +0200_
	11
2	12	## Version 0.18.0
3	13
4	14	### Features
5	15
6	16	- Add TCP stream reader
	17	- Configurable number of server worker threads
7	18
8	19	### Bugfixes
9	20

17	28
18	29	- Refactored stream readers
19	30	- Server can run on a single thread
20		- Configurable number of server worker threads
21	31
22	32	_Johannes Pohl <snapcast@badaix.de> Wed, 22 Jan 2020 00:13:37 +0200_
23	33

+1

-1

client/Makefile less more

13	13	# You should have received a copy of the GNU General Public License
14	14	# along with this program. If not, see <http://www.gnu.org/licenses/>.
15	15
16		VERSION = 0.18.0
	16	VERSION = 0.18.1
17	17	BIN = snapclient
18	18
19	19	ifeq ($(TARGET), FREEBSD)

+1

-1

server/Makefile less more

13	13	# You should have received a copy of the GNU General Public License
14	14	# along with this program. If not, see <http://www.gnu.org/licenses/>.
15	15
16		VERSION = 0.18.0
	16	VERSION = 0.18.1
17	17	BIN = snapserver
18	18
19	19	ifeq ($(TARGET), FREEBSD)

+0

-1

server/control_session.hpp less more

25	25	#include <boost/asio.hpp>
26	26	#include <condition_variable>
27	27	#include <memory>
28		#include <mutex>
29	28	#include <set>
30	29	#include <string>
31	30

+11

-13

server/control_session_http.cpp less more

114	114
115	115	void ControlSessionHttp::start()
116	116	{
117		auto self = shared_from_this();
118		http::async_read(socket_, buffer_, req_,
119		boost::asio::bind_executor(strand_, [this, self](boost::system::error_code ec, std::size_t bytes) { on_read(ec, bytes); }));
	117	http::async_read(socket_, buffer_, req_, boost::asio::bind_executor(strand_, [ this, self = shared_from_this() ](
	118	boost::system::error_code ec, std::size_t bytes) { on_read(ec, bytes); }));
120	119	}
121	120
122	121

255	254	// Create a WebSocket session by transferring the socket
256	255	// std::make_shared<websocket_session>(std::move(socket_), state_)->run(std::move(req_));
257	256	ws_ = make_unique<websocket::stream<beast::tcp_stream>>(std::move(socket_));
258		auto self = shared_from_this();
259		ws_->async_accept(req_, [this, self](beast::error_code ec) { on_accept_ws(ec); });
	257	ws_->async_accept(req_, [ this, self = shared_from_this() ](beast::error_code ec) { on_accept_ws(ec); });
260	258	LOG(DEBUG) << "websocket upgrade\n";
261	259	return;
262	260	}

270	268	auto sp = std::make_shared<response_type>(std::forward<decltype(response)>(response));
271	269
272	270	// Write the response
273		auto self = this->shared_from_this();
274		http::async_write(this->socket_, *sp, boost::asio::bind_executor(strand_, [this, self, sp](beast::error_code ec, std::size_t bytes) {
	271	http::async_write(this->socket_, *sp,
	272	boost::asio::bind_executor(strand_, [ this, self = this->shared_from_this(), sp ](beast::error_code ec, std::size_t bytes) {
275	273	this->on_write(ec, bytes, sp->need_eof());
276	274	}));
277	275	});

314	312	if (!ws_)
315	313	return;
316	314
317		strand_.post([this, message]() {
	315	strand_.post([ this, self = shared_from_this(), message ]() {
318	316	messages_.emplace_back(message);
319	317	if (messages_.size() > 1)
320	318	{

330	328	if (!ws_)
331	329	return;
332	330
333		auto self(shared_from_this());
334	331	auto message = messages_.front();
335		ws_->async_write(boost::asio::buffer(message), boost::asio::bind_executor(strand_, [this, self](std::error_code ec, std::size_t length) {
	332	ws_->async_write(boost::asio::buffer(message),
	333	boost::asio::bind_executor(strand_, [ this, self = shared_from_this() ](std::error_code ec, std::size_t length) {
336	334	messages_.pop_front();
337	335	if (ec)
338	336	{

373	371	void ControlSessionHttp::do_read_ws()
374	372	{
375	373	// Read a message into our buffer
376		auto self(shared_from_this());
377		ws_->async_read(
378		buffer_, boost::asio::bind_executor(strand_, [this, self](beast::error_code ec, std::size_t bytes_transferred) { on_read_ws(ec, bytes_transferred); }));
	374	ws_->async_read(buffer_, boost::asio::bind_executor(strand_, [ this, self = shared_from_this() ](beast::error_code ec, std::size_t bytes_transferred) {
	375	on_read_ws(ec, bytes_transferred);
	376	}));
379	377	}
380	378
381	379

+4

-5

server/control_session_tcp.cpp less more

40	40	void ControlSessionTcp::do_read()
41	41	{
42	42	const std::string delimiter = "\n";
43		auto self(shared_from_this());
44	43	boost::asio::async_read_until(
45		socket_, streambuf_, delimiter, boost::asio::bind_executor(strand_, [this, self, delimiter](const std::error_code& ec, std::size_t bytes_transferred) {
	44	socket_, streambuf_, delimiter,
	45	boost::asio::bind_executor(strand_, [ this, self = shared_from_this(), delimiter ](const std::error_code& ec, std::size_t bytes_transferred) {
46	46	if (ec)
47	47	{
48	48	LOG(ERROR) << "Error while reading from control socket: " << ec.message() << "\n";

91	91
92	92	void ControlSessionTcp::sendAsync(const std::string& message)
93	93	{
94		strand_.post([this, message]() {
	94	strand_.post([ this, self = shared_from_this(), message ]() {
95	95	messages_.emplace_back(message);
96	96	if (messages_.size() > 1)
97	97	{

104	104
105	105	void ControlSessionTcp::send_next()
106	106	{
107		auto self(shared_from_this());
108	107	auto message = messages_.front();
109	108	boost::asio::async_write(socket_, boost::asio::buffer(message + "\r\n"),
110		boost::asio::bind_executor(strand_, [this, self](std::error_code ec, std::size_t length) {
	109	boost::asio::bind_executor(strand_, [ this, self = shared_from_this() ](std::error_code ec, std::size_t length) {
111	110	messages_.pop_front();
112	111	if (ec)
113	112	{

+4

-4

server/encoder/opus_encoder.cpp less more

165	165	// and encode the buffer content in the next iteration
166	166	void OpusEncoder::encode(const msg::PcmChunk* chunk)
167	167	{
168		LOG(DEBUG) << "encode " << chunk->duration<std::chrono::milliseconds>().count() << "ms\n";
	168	// LOG(TRACE) << "encode " << chunk->duration<std::chrono::milliseconds>().count() << "ms\n";
169	169	uint32_t offset = 0;
170	170
171	171	// check if there is something left from the last call to encode and fill the remainder buffer to

174	174	{
175	175	offset = std::min(static_cast<uint32_t>(remainder_max_size_ - remainder_->payloadSize), chunk->payloadSize);
176	176	memcpy(remainder_->payload + remainder_->payloadSize, chunk->payload, offset);
177		LOG(DEBUG) << "remainder buffer size: " << remainder_->payloadSize << "/" << remainder_max_size_ << ", appending " << offset << " bytes\n";
	177	// LOG(TRACE) << "remainder buffer size: " << remainder_->payloadSize << "/" << remainder_max_size_ << ", appending " << offset << " bytes\n";
178	178	remainder_->payloadSize += offset;
179	179
180	180	if (remainder_->payloadSize < remainder_max_size_)

195	195	uint32_t bytes = ms2bytes(duration);
196	196	while (chunk->payloadSize - offset >= bytes)
197	197	{
198		LOG(DEBUG) << "encoding " << duration << "ms (" << bytes << "), offset: " << offset << ", chunk size: " << chunk->payloadSize - offset << "\n";
	198	// LOG(TRACE) << "encoding " << duration << "ms (" << bytes << "), offset: " << offset << ", chunk size: " << chunk->payloadSize - offset << "\n";
199	199	encode(chunk->format, chunk->payload + offset, bytes);
200	200	offset += bytes;
201	201	}

221	221	encoded_.resize(size);
222	222
223	223	opus_int32 len = opus_encode(enc_, (opus_int16*)data, samples_per_channel, encoded_.data(), size);
224		LOG(DEBUG) << "Encode " << samples_per_channel << " frames, size " << size << " bytes, encoded: " << len << " bytes" << '\n';
	224	// LOG(TRACE) << "Encode " << samples_per_channel << " frames, size " << size << " bytes, encoded: " << len << " bytes" << '\n';
225	225
226	226	if (len > 0)
227	227	{

+4

-26

server/stream_session.cpp less more

44	44
45	45	void StreamSession::read_next()
46	46	{
47		shared_ptr<StreamSession> self;
48		try
49		{
50		self = shared_from_this();
51		}
52		catch (const std::bad_weak_ptr& e)
53		{
54		LOG(ERROR, LOG_TAG) << "read_next: Error getting shared from this\n";
55		return;
56		}
57
58	47	boost::asio::async_read(socket_, boost::asio::buffer(buffer_, base_msg_size_),
59		boost::asio::bind_executor(strand_, [this, self](boost::system::error_code ec, std::size_t length) mutable {
	48	boost::asio::bind_executor(strand_, [ this, self = shared_from_this() ](boost::system::error_code ec, std::size_t length) mutable {
60	49	if (ec)
61	50	{
62	51	LOG(ERROR, LOG_TAG) << "Error reading message header of length " << length << ": " << ec.message() << "\n";

138	127
139	128	void StreamSession::send_next()
140	129	{
141		shared_ptr<StreamSession> self;
142		try
143		{
144		self = shared_from_this();
145		}
146		catch (const std::bad_weak_ptr& e)
147		{
148		LOG(ERROR, LOG_TAG) << "send_next: Error getting shared from this\n";
149		return;
150		}
151
152	130	auto buffer = messages_.front();
153
154		boost::asio::async_write(socket_, buffer, boost::asio::bind_executor(strand_, [this, self, buffer](boost::system::error_code ec, std::size_t length) {
	131	boost::asio::async_write(socket_, buffer,
	132	boost::asio::bind_executor(strand_, [ this, self = shared_from_this(), buffer ](boost::system::error_code ec, std::size_t length) {
155	133	messages_.pop_front();
156	134	if (ec)
157	135	{

167	145
168	146	void StreamSession::sendAsync(shared_const_buffer const_buf, bool send_now)
169	147	{
170		strand_.post([this, const_buf, send_now]() {
	148	strand_.post([ this, self = shared_from_this(), const_buf, send_now ]() {
171	149	if (send_now)
172	150	messages_.push_front(const_buf);
173	151	else

+0

-1

server/stream_session.hpp less more

26	26	#include <condition_variable>
27	27	#include <deque>
28	28	#include <memory>
29		#include <mutex>
30	29	#include <set>
31	30	#include <sstream>
32	31	#include <string>

+0

-3

server/streamreader/airplay_stream.cpp less more

58	58
59	59	// XXX: Check if pipe exists, delete or throw error
60	60
61		sampleFormat_ = SampleFormat("44100:16:2");
62		uri_.query["sampleformat"] = sampleFormat_.getFormat();
63
64	61	port_ = cpt::stoul(uri_.getQuery("port", "5000"));
65	62
66	63	string devicename = uri_.getQuery("devicename", "Snapcast");

+1

-1

server/streamreader/asio_stream.hpp less more

102	102	{
103	103	uint64_t last_read = bytes_read_;
104	104	wait(state_timer_, std::chrono::milliseconds(500 + chunk_ms_), [this, last_read] {
105		LOG(DEBUG, "AsioStream") << "check state last: " << last_read << ", read: " << bytes_read_ << "\n";
	105	LOG(TRACE, "AsioStream") << "check state last: " << last_read << ", read: " << bytes_read_ << "\n";
106	106	if (bytes_read_ != last_read)
107	107	setState(ReaderState::kPlaying);
108	108	else

+0

-6

server/streamreader/librespot_stream.cpp less more

33	33
34	34	LibrespotStream::LibrespotStream(PcmListener* pcmListener, boost::asio::io_context& ioc, const StreamUri& uri) : ProcessStream(pcmListener, ioc, uri)
35	35	{
36		sampleFormat_ = SampleFormat("44100:16:2");
37		uri_.query["sampleformat"] = sampleFormat_.getFormat();
38		// chunk is created in PcmStream ctor, using the (possibly wrongly) configured sample format
39		// we have to recreate it using spotify's native sample format
40		chunk_ = std::make_unique<msg::PcmChunk>(sampleFormat_, chunk_ms_);
41
42	36	wd_timeout_sec_ = cpt::stoul(uri_.getQuery("wd_timeout", "7800")); ///< 130min
43	37
44	38	string username = uri_.getQuery("username", "");

+0

-1

server/streamreader/pcm_stream.hpp less more

28	28	#include <boost/asio/io_context.hpp>
29	29	#include <condition_variable>
30	30	#include <map>
31		#include <mutex>
32	31	#include <string>
33	32
34	33

+8

-0

server/streamreader/stream_manager.cpp less more

74	74	}
75	75	else if ((streamUri.scheme == "spotify") \|\| (streamUri.scheme == "librespot"))
76	76	{
	77	// Overwrite sample format here instead of inside the constructor, to make sure
	78	// that all constructors of all parent classes also use the overwritten sample
	79	// format.
	80	streamUri.query[kUriSampleFormat] = "44100:16:2";
77	81	stream = make_shared<LibrespotStream>(pcmListener_, ioc_, streamUri);
78	82	}
79	83	else if (streamUri.scheme == "airplay")
80	84	{
	85	// Overwrite sample format here instead of inside the constructor, to make sure
	86	// that all constructors of all parent classes also use the overwritten sample
	87	// format.
	88	streamUri.query[kUriSampleFormat] = "44100:16:2";
81	89	stream = make_shared<AirplayStream>(pcmListener_, ioc_, streamUri);
82	90	}
83	91	else if (streamUri.scheme == "tcp")