mpd/src/output/HttpdOutputPlugin.cxx

556 lines
12 KiB
C++
Raw Normal View History

/*
2013-01-15 18:22:17 +01:00
* 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"
2013-01-15 18:22:17 +01:00
#include "HttpdOutputPlugin.hxx"
#include "HttpdInternal.hxx"
#include "HttpdClient.hxx"
#include "output_api.h"
#include "encoder_plugin.h"
#include "encoder_list.h"
#include "resolver.h"
2013-01-30 09:18:52 +01:00
#include "Page.hxx"
2013-01-30 09:08:50 +01:00
#include "IcyMetaDataServer.hxx"
#include "fd_util.h"
2013-01-30 13:39:12 +01:00
#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 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();
}
static struct audio_output *
httpd_output_init(const struct config_param *param,
GError **error)
{
HttpdOutput *httpd = new HttpdOutput();
if (!ao_base_init(&httpd->base, &httpd_output_plugin, param, error)) {
g_free(httpd);
return NULL;
}
/* read configuration */
httpd->name =
config_get_block_string(param, "name", "Set name in config");
httpd->genre =
config_get_block_string(param, "genre", "Set genre in config");
httpd->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, httpd_output_quark(), 0,
"No such encoder: %s", encoder_name);
ao_base_finish(&httpd->base);
g_free(httpd);
return NULL;
}
httpd->clients_max = config_get_block_unsigned(param,"max_clients", 0);
/* set up bind_to_address */
httpd->server_socket = new ServerSocket(*main_loop,
httpd_listen_in_event, httpd);
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
? httpd->server_socket->AddHost(bind_to_address, port, error)
: httpd->server_socket->AddPort(port, error);
if (!success) {
ao_base_finish(&httpd->base);
g_free(httpd);
return NULL;
}
/* initialize metadata */
httpd->metadata = NULL;
httpd->unflushed_input = 0;
/* initialize encoder */
httpd->encoder = encoder_init(encoder_plugin, param, error);
if (httpd->encoder == NULL) {
ao_base_finish(&httpd->base);
g_free(httpd);
return NULL;
}
/* determine content type */
httpd->content_type = encoder_get_mime_type(httpd->encoder);
if (httpd->content_type == NULL) {
httpd->content_type = "application/octet-stream";
}
return &httpd->base;
}
static void
httpd_output_finish(struct audio_output *ao)
{
HttpdOutput *httpd = (HttpdOutput *)ao;
if (httpd->metadata)
2013-01-30 09:18:52 +01:00
httpd->metadata->Unref();
encoder_finish(httpd->encoder);
delete httpd->server_socket;
ao_base_finish(&httpd->base);
2013-01-15 18:22:17 +01:00
delete httpd;
}
/**
2013-01-15 18:22:17 +01:00
* 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 */
2013-01-15 18:22:17 +01:00
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));
}
}
2013-01-30 09:18:52 +01:00
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;
2013-01-30 09:18:52 +01:00
return Page::Copy(buffer, size);
}
static bool
httpd_output_enable(struct audio_output *ao, GError **error_r)
{
HttpdOutput *httpd = (HttpdOutput *)ao;
return httpd->Bind(error_r);
}
static void
httpd_output_disable(struct audio_output *ao)
{
HttpdOutput *httpd = (HttpdOutput *)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 = (HttpdOutput *)ao;
2013-01-15 18:22:17 +01:00
assert(httpd->clients.empty());
2013-01-15 18:22:17 +01:00
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)
2013-01-30 09:18:52 +01:00
header->Unref();
encoder_close(encoder);
}
static void
httpd_output_close(struct audio_output *ao)
{
HttpdOutput *httpd = (HttpdOutput *)ao;
2013-01-15 18:22:17 +01:00
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);;
2013-01-15 18:22:17 +01:00
prev = i, i = std::next(prev)) {
assert(i != clients.end());
if (&*i == &client) {
clients.erase_after(prev);
clients_cnt--;
2013-01-15 18:22:17 +01:00
break;
}
}
}
void
HttpdOutput::SendHeader(HttpdClient &client) const
{
if (header != NULL)
client.PushPage(header);
}
2010-11-05 09:42:14 +01:00
static unsigned
httpd_output_delay(struct audio_output *ao)
2010-11-05 09:42:14 +01:00
{
HttpdOutput *httpd = (HttpdOutput *)ao;
2010-11-05 09:42:14 +01:00
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;
}
2010-11-05 09:42:14 +01:00
return httpd->timer->started
? timer_delay(httpd->timer)
: 0;
}
void
2013-01-30 09:18:52 +01:00
HttpdOutput::BroadcastPage(Page *page)
{
assert(page != NULL);
const ScopeLock protect(mutex);
for (auto &client : clients)
2013-01-15 18:22:17 +01:00
client.PushPage(page);
}
void
HttpdOutput::BroadcastFromEncoder()
{
mutex.lock();
for (auto &client : clients) {
2013-01-15 18:22:17 +01:00
if (client.GetQueueSize() > 256 * 1024) {
g_debug("client is too slow, flushing its queue");
client.CancelQueue();
}
}
mutex.unlock();
2013-01-30 09:18:52 +01:00
Page *page;
while ((page = ReadPage()) != nullptr) {
BroadcastPage(page);
2013-01-30 09:18:52 +01:00
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 = (HttpdOutput *)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 = (HttpdOutput *)ao;
if (httpd->LockHasClients()) {
2013-01-15 18:22:17 +01:00
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 */
2013-01-30 09:18:52 +01:00
Page *page = ReadPage();
if (page != NULL) {
if (header != NULL)
2013-01-30 09:18:52 +01:00
header->Unref();
header = page;
BroadcastPage(page);
}
} else {
/* use Icy-Metadata */
if (metadata != NULL)
2013-01-30 09:18:52 +01:00
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 = (HttpdOutput *)ao;
httpd->SendTag(tag);
}
static void
httpd_output_cancel(struct audio_output *ao)
{
HttpdOutput *httpd = (HttpdOutput *)ao;
2013-01-15 18:22:17 +01:00
const ScopeLock protect(httpd->mutex);
for (auto &client : httpd->clients)
client.CancelQueue();
}
const struct audio_output_plugin httpd_output_plugin = {
2013-01-15 18:22:17 +01:00
"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,
};