/* * Copyright 2003-2017 The Music Player Daemon Project * http://www.musicpd.org * * 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 2 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, write to the Free Software Foundation, Inc., * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. */ #include "config.h" #include "HttpdOutputPlugin.hxx" #include "HttpdInternal.hxx" #include "HttpdClient.hxx" #include "output/OutputAPI.hxx" #include "encoder/EncoderInterface.hxx" #include "encoder/EncoderPlugin.hxx" #include "encoder/EncoderList.hxx" #include "net/SocketAddress.hxx" #include "net/ToString.hxx" #include "Page.hxx" #include "IcyMetaDataServer.hxx" #include "system/fd_util.h" #include "event/Call.hxx" #include "util/RuntimeError.hxx" #include "util/Domain.hxx" #include "util/DeleteDisposer.hxx" #include "Log.hxx" #include #include #include #ifdef HAVE_LIBWRAP #include /* needed for AF_UNIX */ #include #endif const Domain httpd_output_domain("httpd_output"); inline HttpdOutput::HttpdOutput(EventLoop &_loop, const ConfigBlock &block) :ServerSocket(_loop), DeferredMonitor(_loop), base(httpd_output_plugin, block), encoder(nullptr), unflushed_input(0), metadata(nullptr) { /* read configuration */ name = block.GetBlockValue("name", "Set name in config"); genre = block.GetBlockValue("genre", "Set genre in config"); website = block.GetBlockValue("website", "Set website in config"); unsigned port = block.GetBlockValue("port", 8000u); const char *encoder_name = block.GetBlockValue("encoder", "vorbis"); const auto encoder_plugin = encoder_plugin_get(encoder_name); if (encoder_plugin == nullptr) throw FormatRuntimeError("No such encoder: %s", encoder_name); clients_max = block.GetBlockValue("max_clients", 0u); /* set up bind_to_address */ const char *bind_to_address = block.GetBlockValue("bind_to_address"); if (bind_to_address != nullptr && strcmp(bind_to_address, "any") != 0) AddHost(bind_to_address, port); else AddPort(port); /* initialize encoder */ prepared_encoder = encoder_init(*encoder_plugin, block); /* determine content type */ content_type = prepared_encoder->GetMimeType(); if (content_type == nullptr) content_type = "application/octet-stream"; } HttpdOutput::~HttpdOutput() { delete prepared_encoder; } inline void HttpdOutput::Bind() { open = false; BlockingCall(GetEventLoop(), [this](){ ServerSocket::Open(); }); } inline void HttpdOutput::Unbind() { assert(!open); BlockingCall(GetEventLoop(), [this](){ ServerSocket::Close(); }); } HttpdOutput * HttpdOutput::Create(EventLoop &event_loop, const ConfigBlock &block) { return new HttpdOutput(event_loop, block); } /** * Creates a new #HttpdClient object and adds it into the * HttpdOutput.clients linked list. */ inline void HttpdOutput::AddClient(int fd) { auto *client = new HttpdClient(*this, fd, GetEventLoop(), !encoder->ImplementsTag()); clients.push_front(*client); /* pass metadata to client */ if (metadata != nullptr) clients.front().PushMetaData(metadata); } void HttpdOutput::RunDeferred() { /* this method runs in the IOThread; it broadcasts pages from our own queue to all clients */ const std::lock_guard protect(mutex); while (!pages.empty()) { PagePtr page = std::move(pages.front()); pages.pop(); for (auto &client : clients) client.PushPage(page); } /* wake up the client that may be waiting for the queue to be flushed */ cond.broadcast(); } void HttpdOutput::OnAccept(int fd, SocketAddress address, gcc_unused int uid) { /* the listener socket has become readable - a client has connected */ #ifdef HAVE_LIBWRAP if (address.GetFamily() != AF_UNIX) { const auto hostaddr = ToString(address); // TODO: shall we obtain the program name from argv[0]? const char *progname = "mpd"; struct request_info req; request_init(&req, RQ_FILE, fd, RQ_DAEMON, progname, 0); fromhost(&req); if (!hosts_access(&req)) { /* tcp wrappers says no */ FormatWarning(httpd_output_domain, "libwrap refused connection (libwrap=%s) from %s", progname, hostaddr.c_str()); close_socket(fd); return; } } #else (void)address; #endif /* HAVE_WRAP */ const std::lock_guard protect(mutex); if (fd >= 0) { /* can we allow additional client */ if (open && (clients_max == 0 || clients.size() < clients_max)) AddClient(fd); else close_socket(fd); } else if (fd < 0 && errno != EINTR) { LogErrno(httpd_output_domain, "accept() failed"); } } PagePtr HttpdOutput::ReadPage() { if (unflushed_input >= 65536) { /* we have fed a lot of input into the encoder, but it didn't give anything back yet - flush now to avoid buffer underruns */ try { encoder->Flush(); } catch (const std::runtime_error &) { /* ignore */ } unflushed_input = 0; } size_t size = 0; do { size_t nbytes = encoder->Read(buffer + size, sizeof(buffer) - size); if (nbytes == 0) break; unflushed_input = 0; size += nbytes; } while (size < sizeof(buffer)); if (size == 0) return nullptr; return std::make_shared(buffer, size); } inline void HttpdOutput::OpenEncoder(AudioFormat &audio_format) { encoder = prepared_encoder->Open(audio_format); /* we have to remember the encoder header, i.e. the first bytes of encoder output after opening it, because it has to be sent to every new client */ header = ReadPage(); unflushed_input = 0; } inline void HttpdOutput::Open(AudioFormat &audio_format) { assert(!open); assert(clients.empty()); const std::lock_guard protect(mutex); OpenEncoder(audio_format); /* initialize other attributes */ timer = new Timer(audio_format); open = true; } inline void HttpdOutput::Close() { assert(open); delete timer; BlockingCall(GetEventLoop(), [this](){ DeferredMonitor::Cancel(); const std::lock_guard protect(mutex); open = false; clients.clear_and_dispose(DeleteDisposer()); }); header.reset(); delete encoder; } void HttpdOutput::RemoveClient(HttpdClient &client) { assert(!clients.empty()); clients.erase_and_dispose(clients.iterator_to(client), DeleteDisposer()); } void HttpdOutput::SendHeader(HttpdClient &client) const { if (header != nullptr) client.PushPage(header); } inline std::chrono::steady_clock::duration HttpdOutput::Delay() const { if (!LockHasClients() && base.pause) { /* if there's no client and this output is paused, then httpd_output_pause() will not do anything, it will not fill the buffer and it will not update the timer; therefore, we reset the timer here */ timer->Reset(); /* some arbitrary delay that is long enough to avoid consuming too much CPU, and short enough to notice new clients quickly enough */ return std::chrono::seconds(1); } return timer->IsStarted() ? timer->GetDelay() : std::chrono::steady_clock::duration::zero(); } void HttpdOutput::BroadcastPage(PagePtr page) { assert(page != nullptr); { const std::lock_guard lock(mutex); pages.emplace(std::move(page)); } DeferredMonitor::Schedule(); } void HttpdOutput::BroadcastFromEncoder() { /* synchronize with the IOThread */ { const std::lock_guard lock(mutex); while (!pages.empty()) cond.wait(mutex); PagePtr page; while ((page = ReadPage()) != nullptr) pages.push(page); } DeferredMonitor::Schedule(); } inline void HttpdOutput::EncodeAndPlay(const void *chunk, size_t size) { encoder->Write(chunk, size); unflushed_input += size; BroadcastFromEncoder(); } inline size_t HttpdOutput::Play(const void *chunk, size_t size) { if (LockHasClients()) EncodeAndPlay(chunk, size); if (!timer->IsStarted()) timer->Start(); timer->Add(size); return size; } bool HttpdOutput::Pause() { if (LockHasClients()) { static const char silence[1020] = { 0 }; Play(silence, sizeof(silence)); } return true; } inline void HttpdOutput::SendTag(const Tag &tag) { if (encoder->ImplementsTag()) { /* embed encoder tags */ /* flush the current stream, and end it */ try { encoder->PreTag(); } catch (const std::runtime_error &) { /* ignore */ } BroadcastFromEncoder(); /* send the tag to the encoder - which starts a new stream now */ try { encoder->SendTag(tag); } catch (const std::runtime_error &) { /* ignore */ } /* the first page generated by the encoder will now be used as the new "header" page, which is sent to all new clients */ auto page = ReadPage(); if (page != nullptr) { header = std::move(page); BroadcastPage(page); } } else { /* use Icy-Metadata */ static constexpr TagType types[] = { TAG_ALBUM, TAG_ARTIST, TAG_TITLE, TAG_NUM_OF_ITEM_TYPES }; metadata = icy_server_metadata_page(tag, &types[0]); if (metadata != nullptr) { const std::lock_guard protect(mutex); for (auto &client : clients) client.PushMetaData(metadata); } } } inline void HttpdOutput::CancelAllClients() { const std::lock_guard protect(mutex); while (!pages.empty()) { PagePtr page = std::move(pages.front()); pages.pop(); } for (auto &client : clients) client.CancelQueue(); cond.broadcast(); } void HttpdOutput::Cancel() { BlockingCall(GetEventLoop(), [this](){ CancelAllClients(); }); } typedef AudioOutputWrapper Wrapper; const struct AudioOutputPlugin httpd_output_plugin = { "httpd", nullptr, &Wrapper::Init, &Wrapper::Finish, &Wrapper::Enable, &Wrapper::Disable, &Wrapper::Open, &Wrapper::Close, &Wrapper::Delay, &Wrapper::SendTag, &Wrapper::Play, nullptr, &Wrapper::Cancel, &Wrapper::Pause, nullptr, };