Commit 81709049 authored by Michael Smith's avatar Michael Smith
Browse files

General code cleanup and reorganisation

svn path=/trunk/ices/; revision=2078
parent 12b976d8
......@@ -83,6 +83,7 @@ typedef struct _config_tag
int shutdown;
char *metadata_filename;
cond_t queue_cond;
cond_t event_pending_cond;
mutex_t refcount_lock;
mutex_t flush_lock;
input_module_t *inmod;
......
......@@ -215,6 +215,7 @@ void input_loop(void)
ices_config->inmod = inmod;
thread_cond_create(&ices_config->queue_cond);
thread_cond_create(&ices_config->event_pending_cond);
thread_mutex_create(&ices_config->refcount_lock);
thread_mutex_create(&ices_config->flush_lock);
......@@ -227,7 +228,7 @@ void input_loop(void)
while(instance)
{
stream_description *arg = malloc(sizeof(stream_description));
stream_description *arg = calloc(1, sizeof(stream_description));
arg->stream = instance;
arg->input = inmod;
if(instance->savefilename != NULL)
......@@ -392,6 +393,7 @@ void input_loop(void)
LOG_DEBUG0("All instances removed, shutting down control thread.");
thread_cond_destroy(&ices_config->queue_cond);
thread_cond_destroy(&ices_config->event_pending_cond);
thread_mutex_destroy(&ices_config->flush_lock);
thread_mutex_destroy(&ices_config->refcount_lock);
......
......@@ -12,13 +12,22 @@
#ifndef __INPUT_H__
#define __INPUT_H__
#include <shout/shout.h>
#include <vorbis/codec.h>
#include "config.h"
#include "inputmodule.h"
#include "stream.h"
#include "reencode.h"
#include "encode.h"
typedef struct {
instance_t *stream;
input_module_t *input;
reencode_state *reenc;
encoder_state *enc;
shout_conn_t conn;
vorbis_comment vc;
} stream_description;
......
......@@ -18,6 +18,7 @@
#include "config.h"
#include "inputmodule.h"
#include "event.h"
#include "thread/thread.h"
#define MODULE "metadata/"
#include "logging.h"
......@@ -74,7 +75,7 @@ void *metadata_thread_signal(void *arg)
FILE *file;
while(metadata_update_signalled == 0)
sleep(1);
thread_cond_wait(&ices_config->event_pending_cond);
metadata_update_signalled = 0;
......
......@@ -30,6 +30,7 @@ void signal_usr1_handler(int signum)
{
LOG_INFO0("Metadata update requested");
metadata_update_signalled = 1;
thread_cond_broadcast(&ices_config->event_pending_cond);
signal(SIGUSR1, signal_usr1_handler);
}
......@@ -42,7 +43,6 @@ void signal_hup_handler(int signum)
/* Now, let's tell it to move to the next track */
ices_config->inmod->handle_event(ices_config->inmod,EVENT_NEXTTRACK,NULL);
/* Do we need to do this? */
signal(SIGHUP, signal_hup_handler);
}
......
......@@ -43,61 +43,55 @@
*/
void *ices_instance_stream(void *arg)
{
shout_conn_t conn;
int ret;
int errors=0;
ref_buffer *buffer;
stream_description *sdsc = arg;
instance_t *stream = sdsc->stream;
input_module_t *inmod = sdsc->input;
queue_item *old;
reencode_state *reenc=NULL;
int reencoding = (inmod->type == ICES_INPUT_VORBIS) && stream->encode;
int encoding = (inmod->type == ICES_INPUT_PCM) && stream->encode;
encoder_state *enc=NULL;
vorbis_comment vc;
vorbis_comment_init(&vc);
vorbis_comment_init(&sdsc->vc);
shout_init_connection(&conn);
shout_init_connection(&sdsc->conn);
signal(SIGPIPE, signal_hup_handler);
conn.ip = malloc(16);
if(!resolver_getip(stream->hostname, conn.ip, 16))
sdsc->conn.ip = malloc(16);
if(!resolver_getip(stream->hostname, sdsc->conn.ip, 16))
{
LOG_ERROR1("Could not resolve hostname \"%s\"", stream->hostname);
free(conn.ip);
free(sdsc->conn.ip);
stream->died = 1;
return NULL;
}
conn.port = stream->port;
conn.password = strdup(stream->password);
conn.mount = strdup(stream->mount);
sdsc->conn.port = stream->port;
sdsc->conn.password = strdup(stream->password);
sdsc->conn.mount = strdup(stream->mount);
/* set the metadata for the stream */
if (ices_config->stream_name)
conn.name = strdup(ices_config->stream_name);
sdsc->conn.name = strdup(ices_config->stream_name);
if (ices_config->stream_genre)
conn.genre = strdup(ices_config->stream_genre);
sdsc->conn.genre = strdup(ices_config->stream_genre);
if (ices_config->stream_description)
conn.description = strdup(ices_config->stream_description);
sdsc->conn.description = strdup(ices_config->stream_description);
if(encoding)
{
if(inmod->metadata_update)
inmod->metadata_update(inmod->internal, &vc);
enc = encode_initialise(stream->channels, stream->samplerate,
stream->bitrate, stream->serial++, &vc);
inmod->metadata_update(inmod->internal, &sdsc->vc);
sdsc->enc = encode_initialise(stream->channels, stream->samplerate,
stream->bitrate, stream->serial++, &sdsc->vc);
}
else if(reencoding)
reenc = reencode_init(stream);
sdsc->reenc = reencode_init(stream);
if(shout_connect(&conn))
if(shout_connect(&sdsc->conn))
{
LOG_INFO3("Connected to server: %s:%d%s",
conn.ip, conn.port, conn.mount);
sdsc->conn.ip, sdsc->conn.port, sdsc->conn.mount);
while(1)
{
......@@ -127,74 +121,20 @@ void *ices_instance_stream(void *arg)
continue;
}
if(encoding)
{
ogg_page og;
int be = (inmod->subtype == INPUT_PCM_BE_16)?1:0;
/* We use critical as a flag to say 'start a new stream' */
if(buffer->critical)
{
encode_finish(enc);
while(encode_flush(enc, &og) != 0)
{
ret = shout_send_data(&conn, og.header, og.header_len);
ret = shout_send_data(&conn, og.body, og.body_len);
}
encode_clear(enc);
if(inmod->metadata_update)
{
vorbis_comment_clear(&vc);
vorbis_comment_init(&vc);
inmod->metadata_update(inmod->internal, &vc);
}
enc = encode_initialise(stream->channels,stream->samplerate,
stream->bitrate, stream->serial++, &vc);
}
encode_data(enc, (signed char *)(buffer->buf), buffer->len, be);
while(encode_dataout(enc, &og) > 0)
{
/* FIXME: This is wrong. Get the return values right. */
ret=shout_send_data(&conn, og.header, og.header_len);
ret=shout_send_data(&conn, og.body, og.body_len);
}
}
else if(reencoding)
{
unsigned char *buf;
int buflen,ret2;
ret2 = reencode_page(reenc, buffer, &buf, &buflen);
if(ret2 > 0)
{
ret = shout_send_data(&conn, buf, buflen);
free(buf);
}
else if(ret2==0)
{
ret = -1; /* This way we don't enter the error handling
code */
}
else
{
LOG_ERROR0("Fatal reencoding error encountered");
errors = MAX_ERRORS+1;
continue;
}
}
else
ret = shout_send_data(&conn, buffer->buf, buffer->len);
ret = process_and_send_buffer(sdsc, buffer);
if(!ret)
if(ret == -1)
continue;
else if(ret == -2)
{
errors = MAX_ERRORS+1;
continue;
}
else if(ret == 0)
{
LOG_ERROR1("Send error: %s", shout_strerror(&conn, conn.error));
if(conn.error == SHOUTERR_SOCKET)
LOG_ERROR1("Send error: %s",
shout_strerror(&sdsc->conn, sdsc->conn.error));
if(sdsc->conn.error == SHOUTERR_SOCKET)
{
int i=0;
......@@ -214,18 +154,19 @@ void *ices_instance_stream(void *arg)
{
i++;
LOG_WARN0("Trying reconnect after server socket error");
shout_disconnect(&conn);
if(shout_connect(&conn))
shout_disconnect(&sdsc->conn);
if(shout_connect(&sdsc->conn))
{
LOG_INFO3("Connected to server: %s:%d%s",
conn.ip, conn.port, conn.mount);
sdsc->conn.ip, sdsc->conn.port,
sdsc->conn.mount);
break;
}
else
{
LOG_ERROR3("Failed to reconnect to %s:%d (%s)",
conn.ip,conn.port,
shout_strerror(&conn,conn.error));
sdsc->conn.ip,sdsc->conn.port,
shout_strerror(&sdsc->conn,sdsc->conn.error));
if(i==stream->reconnect_attempts)
{
LOG_ERROR0("Reconnect failed too many times, "
......@@ -250,20 +191,18 @@ void *ices_instance_stream(void *arg)
else
{
LOG_ERROR3("Failed initial connect to %s:%d (%s)",
conn.ip,conn.port,shout_strerror(&conn,conn.error));
sdsc->conn.ip,sdsc->conn.port,
shout_strerror(&sdsc->conn,sdsc->conn.error));
}
shout_disconnect(&conn);
shout_disconnect(&sdsc->conn);
free(conn.ip);
encode_clear(enc);
reencode_clear(reenc);
vorbis_comment_clear(&vc);
free(sdsc->conn.ip);
encode_clear(sdsc->enc);
reencode_clear(sdsc->reenc);
vorbis_comment_clear(&sdsc->vc);
stream->died = 1;
return NULL;
}
......@@ -19,6 +19,8 @@
#include "inputmodule.h"
#include "stream_shared.h"
#include "stream.h"
#include "reencode.h"
#include "encode.h"
#define MODULE "stream-shared/"
#include "logging.h"
......@@ -77,8 +79,75 @@ ref_buffer *stream_wait_for_data(instance_t *stream)
return buffer;
}
/* Process a buffer (including reencoding or encoding, if desired).
* Returns: >0 - success
* 0 - shout error occurred
* -1 - no data produced
* -2 - fatal error occurred
*/
int process_and_send_buffer(stream_description *sdsc, ref_buffer *buffer)
{
if(sdsc->reenc)
{
unsigned char *buf;
int buflen,ret;
ret = reencode_page(sdsc->reenc, buffer, &buf, &buflen);
if(ret > 0)
{
ret = shout_send_data(&sdsc->conn, buf, buflen);
free(buf);
return ret;
}
else if(ret==0)
return -1;
else
{
LOG_ERROR0("Fatal reencoding error encountered");
return -2;
}
}
else if (sdsc->enc)
{
ogg_page og;
int be = (sdsc->input->subtype == INPUT_PCM_BE_16)?1:0;
int ret=1;
/* We use critical as a flag to say 'start a new stream' */
if(buffer->critical)
{
encode_finish(sdsc->enc);
while(encode_flush(sdsc->enc, &og) != 0)
{
ret = shout_send_data(&sdsc->conn, og.header, og.header_len);
ret = shout_send_data(&sdsc->conn, og.body, og.body_len);
}
encode_clear(sdsc->enc);
if(sdsc->input->metadata_update)
{
vorbis_comment_clear(&sdsc->vc);
vorbis_comment_init(&sdsc->vc);
sdsc->input->metadata_update(sdsc->input->internal, &sdsc->vc);
}
sdsc->enc = encode_initialise(sdsc->stream->channels,
sdsc->stream->samplerate, sdsc->stream->bitrate,
sdsc->stream->serial++, &sdsc->vc);
}
encode_data(sdsc->enc, (signed char *)(buffer->buf), buffer->len, be);
while(encode_dataout(sdsc->enc, &og) > 0)
{
ret = shout_send_data(&sdsc->conn, og.header, og.header_len);
ret = shout_send_data(&sdsc->conn, og.body, og.body_len);
}
return ret;
}
else
return shout_send_data(&sdsc->conn, buffer->buf, buffer->len);
}
......@@ -14,9 +14,11 @@
#include "stream.h"
#include "config.h"
#include "input.h"
ref_buffer *stream_wait_for_data(instance_t *stream);
void stream_release_buffer(ref_buffer *buf);
int process_and_send_buffer(stream_description *sdsc, ref_buffer *buffer);
#endif
......
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment