/*
* Copyright (C) 2003-2013 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_api.h"
#include "encoder_plugin.h"
#include "encoder_list.h"
#include "resolver.h"
#include "Page.hxx"
#include "IcyMetaDataServer.hxx"
#include "fd_util.h"
#include "event/ServerSocket.hxx"
#include "Main.hxx"
#include <assert.h>
#include <sys/types.h>
#include <unistd.h>
#include <errno.h>
#ifdef HAVE_LIBWRAP
#include <sys/socket.h> /* needed for AF_UNIX */
#include <tcpd.h>
#endif
#undef G_LOG_DOMAIN
#define G_LOG_DOMAIN "httpd_output"
/**
* The quark used for GError.domain.
*/
static inline GQuark
httpd_output_quark(void)
{
return g_quark_from_static_string("httpd_output");
}
static void
httpd_listen_in_event(int fd, const struct sockaddr *address,
size_t address_length, int uid, void *ctx);
inline
HttpdOutput::HttpdOutput(EventLoop &_loop)
:encoder(nullptr), unflushed_input(0),
server_socket(new ServerSocket(_loop, httpd_listen_in_event, this)),
metadata(nullptr)
{
}
HttpdOutput::~HttpdOutput()
{
if (metadata != nullptr)
metadata->Unref();
if (encoder != nullptr)
encoder_finish(encoder);
delete server_socket;
}
inline bool
HttpdOutput::Bind(GError **error_r)
{
open = false;
const ScopeLock protect(mutex);
return server_socket->Open(error_r);
}
inline void
HttpdOutput::Unbind()
{
assert(!open);
const ScopeLock protect(mutex);
server_socket->Close();
}
inline bool
HttpdOutput::Configure(const config_param *param, GError **error_r)
{
/* read configuration */
name = config_get_block_string(param, "name", "Set name in config");
genre = config_get_block_string(param, "genre", "Set genre in config");
website = config_get_block_string(param, "website",
"Set website in config");
guint port = config_get_block_unsigned(param, "port", 8000);
const char *encoder_name =
config_get_block_string(param, "encoder", "vorbis");
const struct encoder_plugin *encoder_plugin =
encoder_plugin_get(encoder_name);
if (encoder_plugin == NULL) {
g_set_error(error_r, httpd_output_quark(), 0,
"No such encoder: %s", encoder_name);
return false;
}
clients_max = config_get_block_unsigned(param,"max_clients", 0);
/* set up bind_to_address */
const char *bind_to_address =
config_get_block_string(param, "bind_to_address", NULL);
bool success = bind_to_address != NULL &&
strcmp(bind_to_address, "any") != 0
? server_socket->AddHost(bind_to_address, port, error_r)
: server_socket->AddPort(port, error_r);
if (!success)
return false;
/* initialize encoder */
encoder = encoder_init(encoder_plugin, param, error_r);
if (encoder == nullptr)
return false;
/* determine content type */
content_type = encoder_get_mime_type(encoder);
if (content_type == nullptr)
content_type = "application/octet-stream";
return true;
}
static struct audio_output *
httpd_output_init(const struct config_param *param,
GError **error_r)
{
HttpdOutput *httpd = new HttpdOutput(*main_loop);
if (!ao_base_init(&httpd->base, &httpd_output_plugin, param,
error_r)) {
delete httpd;
return nullptr;
}
if (!httpd->Configure(param, error_r)) {
ao_base_finish(&httpd->base);
delete httpd;
return nullptr;
}
return &httpd->base;
}
static inline constexpr HttpdOutput *
Cast(audio_output *ao)
{
return (HttpdOutput *)((char *)ao - offsetof(HttpdOutput, base));
}
static void
httpd_output_finish(struct audio_output *ao)
{
HttpdOutput *httpd = Cast(ao);
ao_base_finish(&httpd->base);
delete httpd;
}
/**
* Creates a new #HttpdClient object and adds it into the
* HttpdOutput.clients linked list.
*/
inline void
HttpdOutput::AddClient(int fd)
{
clients.emplace_front(this, fd,
encoder->plugin->tag == NULL);
++clients_cnt;
/* pass metadata to client */
if (metadata != nullptr)
clients.front().PushMetaData(metadata);
}
static void
httpd_listen_in_event(int fd, const struct sockaddr *address,
size_t address_length, G_GNUC_UNUSED int uid, void *ctx)
{
HttpdOutput *httpd = (HttpdOutput *)ctx;
/* the listener socket has become readable - a client has
connected */
#ifdef HAVE_LIBWRAP
if (address->sa_family != AF_UNIX) {
char *hostaddr = sockaddr_to_string(address, address_length, NULL);
const char *progname = g_get_prgname();
struct request_info req;
request_init(&req, RQ_FILE, fd, RQ_DAEMON, progname, 0);
fromhost(&req);
if (!hosts_access(&req)) {
/* tcp wrappers says no */
g_warning("libwrap refused connection (libwrap=%s) from %s",
progname, hostaddr);
g_free(hostaddr);
close_socket(fd);
return;
}
g_free(hostaddr);
}
#else
(void)address;
(void)address_length;
#endif /* HAVE_WRAP */
const ScopeLock protect(httpd->mutex);
if (fd >= 0) {
/* can we allow additional client */
if (httpd->open &&
(httpd->clients_max == 0 ||
httpd->clients_cnt < httpd->clients_max))
httpd->AddClient(fd);
else
close_socket(fd);
} else if (fd < 0 && errno != EINTR) {
g_warning("accept() failed: %s", g_strerror(errno));
}
}
Page *
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 */
encoder_flush(encoder, NULL);
unflushed_input = 0;
}
size_t size = 0;
do {
size_t nbytes = encoder_read(encoder,
buffer + size,
sizeof(buffer) - size);
if (nbytes == 0)
break;
unflushed_input = 0;
size += nbytes;
} while (size < sizeof(buffer));
if (size == 0)
return NULL;
return Page::Copy(buffer, size);
}
static bool
httpd_output_enable(struct audio_output *ao, GError **error_r)
{
HttpdOutput *httpd = Cast(ao);
return httpd->Bind(error_r);
}
static void
httpd_output_disable(struct audio_output *ao)
{
HttpdOutput *httpd = Cast(ao);
httpd->Unbind();
}
inline bool
HttpdOutput::OpenEncoder(struct audio_format *audio_format, GError **error)
{
if (!encoder_open(encoder, audio_format, error))
return false;
/* 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;
return true;
}
inline bool
HttpdOutput::Open(struct audio_format *audio_format, GError **error_r)
{
assert(!open);
assert(clients.empty());
/* open the encoder */
if (!OpenEncoder(audio_format, error_r))
return false;
/* initialize other attributes */
clients_cnt = 0;
timer = timer_new(audio_format);
open = true;
return true;
}
static bool
httpd_output_open(struct audio_output *ao, struct audio_format *audio_format,
GError **error)
{
HttpdOutput *httpd = Cast(ao);
assert(httpd->clients.empty());
const ScopeLock protect(httpd->mutex);
return httpd->Open(audio_format, error);
}
inline void
HttpdOutput::Close()
{
assert(open);
open = false;
timer_free(timer);
clients.clear();
if (header != NULL)
header->Unref();
encoder_close(encoder);
}
static void
httpd_output_close(struct audio_output *ao)
{
HttpdOutput *httpd = Cast(ao);
const ScopeLock protect(httpd->mutex);
httpd->Close();
}
void
HttpdOutput::RemoveClient(HttpdClient &client)
{
assert(clients_cnt > 0);
for (auto prev = clients.before_begin(), i = std::next(prev);;
prev = i, i = std::next(prev)) {
assert(i != clients.end());
if (&*i == &client) {
clients.erase_after(prev);
clients_cnt--;
break;
}
}
}
void
HttpdOutput::SendHeader(HttpdClient &client) const
{
if (header != NULL)
client.PushPage(header);
}
static unsigned
httpd_output_delay(struct audio_output *ao)
{
HttpdOutput *httpd = Cast(ao);
if (!httpd->LockHasClients() && httpd->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(httpd->timer);
/* some arbitrary delay that is long enough to avoid
consuming too much CPU, and short enough to notice
new clients quickly enough */
return 1000;
}
return httpd->timer->started
? timer_delay(httpd->timer)
: 0;
}
void
HttpdOutput::BroadcastPage(Page *page)
{
assert(page != NULL);
const ScopeLock protect(mutex);
for (auto &client : clients)
client.PushPage(page);
}
void
HttpdOutput::BroadcastFromEncoder()
{
mutex.lock();
for (auto &client : clients) {
if (client.GetQueueSize() > 256 * 1024) {
g_debug("client is too slow, flushing its queue");
client.CancelQueue();
}
}
mutex.unlock();
Page *page;
while ((page = ReadPage()) != nullptr) {
BroadcastPage(page);
page->Unref();
}
}
inline bool
HttpdOutput::EncodeAndPlay(const void *chunk, size_t size, GError **error_r)
{
if (!encoder_write(encoder, chunk, size, error_r))
return false;
unflushed_input += size;
BroadcastFromEncoder();
return true;
}
static size_t
httpd_output_play(struct audio_output *ao, const void *chunk, size_t size,
GError **error_r)
{
HttpdOutput *httpd = Cast(ao);
if (httpd->LockHasClients()) {
if (!httpd->EncodeAndPlay(chunk, size, error_r))
return 0;
}
if (!httpd->timer->started)
timer_start(httpd->timer);
timer_add(httpd->timer, size);
return size;
}
static bool
httpd_output_pause(struct audio_output *ao)
{
HttpdOutput *httpd = Cast(ao);
if (httpd->LockHasClients()) {
static const char silence[1020] = { 0 };
return httpd_output_play(ao, silence, sizeof(silence),
NULL) > 0;
} else {
return true;
}
}
inline void
HttpdOutput::SendTag(const struct tag *tag)
{
assert(tag != NULL);
if (encoder->plugin->tag != NULL) {
/* embed encoder tags */
/* flush the current stream, and end it */
encoder_pre_tag(encoder, NULL);
BroadcastFromEncoder();
/* send the tag to the encoder - which starts a new
stream now */
encoder_tag(encoder, tag, NULL);
/* the first page generated by the encoder will now be
used as the new "header" page, which is sent to all
new clients */
Page *page = ReadPage();
if (page != NULL) {
if (header != NULL)
header->Unref();
header = page;
BroadcastPage(page);
}
} else {
/* use Icy-Metadata */
if (metadata != NULL)
metadata->Unref();
static constexpr tag_type types[] = {
TAG_ALBUM, TAG_ARTIST, TAG_TITLE,
TAG_NUM_OF_ITEM_TYPES
};
metadata = icy_server_metadata_page(tag, &types[0]);
if (metadata != NULL) {
const ScopeLock protect(mutex);
for (auto &client : clients)
client.PushMetaData(metadata);
}
}
}
static void
httpd_output_tag(struct audio_output *ao, const struct tag *tag)
{
HttpdOutput *httpd = Cast(ao);
httpd->SendTag(tag);
}
static void
httpd_output_cancel(struct audio_output *ao)
{
HttpdOutput *httpd = Cast(ao);
const ScopeLock protect(httpd->mutex);
for (auto &client : httpd->clients)
client.CancelQueue();
}
const struct audio_output_plugin httpd_output_plugin = {
"httpd",
nullptr,
httpd_output_init,
httpd_output_finish,
httpd_output_enable,
httpd_output_disable,
httpd_output_open,
httpd_output_close,
httpd_output_delay,
httpd_output_tag,
httpd_output_play,
nullptr,
httpd_output_cancel,
httpd_output_pause,
nullptr,
};