/*
Copyright 2012, OpenHome. All rights reserved.
Redistribution and use in source and binary forms, with or without
modification, are permitted provided that the following conditions are
met:
1. Redistributions of source code must retain the above copyright
notice, this list of conditions and the following disclaimer.
2. Redistributions in binary form must reproduce the above
copyright notice, this list of conditions and the following
disclaimer in the documentation and/or other materials provided
with the distribution.
THIS SOFTWARE IS PROVIDED BY OPENHOME ''AS IS'' AND ANY EXPRESS OR
IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
DISCLAIMED. IN NO EVENT SHALL <COPYRIGHT HOLDER> OR CONTRIBUTORS BE
LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR
BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY,
WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE
OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN
IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
The views and conclusions contained in the software and documentation
are those of the authors and should not be interpreted as representing
official policies, either expressed or implied, of OpenHome.
*/
#include <OpenHome/OhNetTypes.h>
#include <OpenHome/Net/Core/DvDevice.h>
#include <OpenHome/Net/Core/OhNet.h>
#include <OpenHome/Net/Core/CpDevice.h>
#include <OpenHome/Net/Core/CpDeviceUpnp.h>
#include <OpenHome/Private/Ascii.h>
#include <OpenHome/Private/Thread.h>
#include <OpenHome/Private/OptionParser.h>
#include <OpenHome/Private/Debug.h>
#include <OpenHome/Os.h>
#include <OpenHome/Private/Env.h>
#include <cstring>
#include <vector>
#include <stdio.h>
#include <unistd.h>
#include <signal.h>
#include <sys/socket.h>
#include <sys/un.h>
#ifdef WITH_OHBUILD
#include "OhmSender.h"
#else
#include <OpenHome/OhmSender.h>
#endif
#include "log.h"
#include "icon.h"
#include "audioreader.h"
#include "openaudio.h"
#include "base64.hxx"
using namespace std;
using namespace OpenHome;
using namespace OpenHome::Net;
using namespace OpenHome::TestFramework;
using namespace OpenHome::Av;
int g_sockfd = -1;
// Sig catcher so that we can interrupt the pause() which will be
// waiting for playing to end. Also set a flag for the benefit of busyRdWr()
bool g_quitrequest = false;
void sigcatcher(int)
{
LOGDEB("sigcatcher\n");
g_quitrequest = true;
// As recvfrom() is a blocking operation we need to shut the corresponding
// socket down so that recvfrom() returns with 0.
if (g_sockfd > 0) {
shutdown(g_sockfd, SHUT_RD);
}
}
class PcmSender {
public:
static const TUint kPeriodMs = 10;
static const TUint kSpeedNormal = 100;
static const TUint kSpeedMin = 75;
static const TUint kSpeedMax = 150;
static const TUint kMaxPacketBytes = 4096;
public:
PcmSender(Environment& aEnv, OhmSender* aSender,
OhmSenderDriver* aDriver, const Brx& aUri,
AudioReader* audio, bool paced);
bool Start(TBool aEnabled);
void Play();
void PlayPause();
void Stop();
void Restart();
~PcmSender();
void busyRdWr();
private:
void CalculatePacketBytes();
void TimerExpired();
private:
Environment& iEnv;
OhmSender* iSender;
OhmSenderDriver* iDriver;
Bws<OhmSender::kMaxTrackUriBytes> iUri;
AudioReader *m_audio;
Timer iTimer;
Mutex iMutex;
TBool iPaused;
TUint iSpeed; // percent, 100%=normal
TUint iIndex; // byte offset read position in source data
TUint iPacketBytes; // how many bytes of audio in each packet
TUint iPacketFrames; // how many audio frames in each packet
TUint iPacketTime; // how much audio time in each packet, uS
TUint64 iLastTimeUs; // last time stamp from system
TInt32 iTimeOffsetUs; // running offset in usec from ideal time
// <0 means sender is behind
// >0 means sender is ahead
TBool iVerbose;
TBool iPaced;
};
PcmSender::PcmSender(Environment& aEnv, OhmSender* aSender,
OhmSenderDriver* aDriver,
const Brx& aUri, AudioReader* audio, bool paced)
: iEnv(aEnv)
, iSender(aSender)
, iDriver(aDriver)
, iUri(aUri)
, m_audio(audio)
, iTimer(aEnv, MakeFunctor(*this, &PcmSender::TimerExpired), "PcmSender")
, iMutex("WAVP")
, iPaused(false)
, iSpeed(kSpeedNormal)
, iIndex(0)
, iLastTimeUs(0)
, iTimeOffsetUs(0)
, iVerbose(false)
, iPaced(paced)
{
CalculatePacketBytes();
LOGDEB("bytes per packet: " << iPacketBytes << endl);
LOGDEB("frames per packet: " << iPacketFrames << endl);
LOGDEB("usec per packet: "<< iPacketTime << endl);
}
#define SOCK_PATH "/tmp/mpd2sc.sock"
#define BUF_SIZE 16
void HandleUserCmd(OhmSender* sender, PcmSender* pcmsender)
{
struct sockaddr_un server_addr, client_addr;
socklen_t len;
ssize_t num_bytes;
char buf[BUF_SIZE];
LOGDEB("PcmSender: Running user command handler\n");
g_sockfd = socket(AF_UNIX, SOCK_DGRAM, 0);
if (g_sockfd == -1) {
LOGERR("Error: Cannot create socket: " << strerror(errno) << endl);
goto _leave;
}
if (remove(SOCK_PATH) == -1 && errno != ENOENT) {
LOGERR("Error: Cannot remove socket path: " << strerror(errno) << endl);
goto _close_socket;
}
memset(&server_addr, 0, sizeof(struct sockaddr_un));
server_addr.sun_family = AF_UNIX;
strncpy(server_addr.sun_path, SOCK_PATH, sizeof(server_addr.sun_path) - 1);
if (bind(g_sockfd, (struct sockaddr *) &server_addr, sizeof(struct sockaddr_un)) == -1) {
LOGERR("Error: Cannot bind to socket: " << strerror(errno) << endl);
goto _close_socket;
}
do {
len = sizeof(struct sockaddr_un);
// recvfrom() is a blocking call and will wait for messages to arrive.
// We are shutting down the socket in the signal catcher in which case
// recvfrom() shall return 0. As g_quitrequest is set in the signal
// catcher, the endless loop is quit.
num_bytes = recvfrom(g_sockfd, buf, BUF_SIZE, 0, (struct sockaddr *) &client_addr, &len);
LOGDEB("HandleUserCmd: Received " << num_bytes << " bytes: " << string(buf, num_bytes) << endl);
if (num_bytes == -1) {
LOGERR("Error: Failed to receive from socket: " << strerror(errno) << endl);
continue; // Ignore failed request
}
if (num_bytes > 0) {
string response("OK");
if (strncmp("play", buf, num_bytes) == 0)
pcmsender->Play();
else if (strncmp("stop", buf, num_bytes) == 0)
pcmsender->Stop();
else if (strncmp("playpause", buf, num_bytes) == 0)
pcmsender->PlayPause();
else if (strncmp("restart", buf, num_bytes) == 0)
pcmsender->Restart();
else if (strncmp("enable", buf, num_bytes) == 0)
sender->SetEnabled(true);
else if (strncmp("disable", buf, num_bytes) == 0)
sender->SetEnabled(false);
else if (strncmp("multicast", buf, num_bytes) == 0)
sender->SetMulticast(true);
else if (strncmp("unicast", buf, num_bytes) == 0)
sender->SetMulticast(false);
ssize_t bytes_sent = sendto(g_sockfd, (const void *)response.c_str(),
response.length(), 0, (struct sockaddr *)&client_addr, len);
if (bytes_sent == -1) {
LOGERR("Error: Cannot write to socket: " <<
strerror(errno) << endl);
}
}
} while (!g_quitrequest);
_close_socket:
if (g_sockfd > 0)
close(g_sockfd);
_leave:
LOGDEB("PcmSender: Leaving user command handler\n");
}
// We return true if the main thread should pause.
bool PcmSender::Start(TBool aEnabled)
{
iDriver->SetAudioFormat(m_audio->sampleRate(), m_audio->byteRate() * 8,
m_audio->numChannels(),
m_audio->bitsPerSample(), true, Brn("WAV"));
iSender->SetEnabled(true);
iSender->SetTrack(iUri, Brx::Empty(), m_audio->sampleCount(), 0);
iSender->SetMetatext(Brn("PcmSender repeated play"));
// It seems that both hijacking the main thread and using the
// timer with a short timeout (see TimerExpired()) work.
// Don't know what's best. The timer approach is closer to the original
// code and leaves the main thread free for control ops if needed.
// Otoh, if no data appears on the fifo, the timer thread is stuck in read
// and the upnp side stops working (no sender advertised).
// Maybe the best approach would be to start a separate thread and
// use busyreading. Using the main thread for now.
static const bool optionbusy = true;
if (iPaced || !optionbusy) {
LOGDEB("PcmSender::Start: using timers. Enabled? " << aEnabled << endl);
if (aEnabled)
iTimer.FireIn(kPeriodMs);
return true;
} else {
LOGDEB("PcmSender::Start: block on reading only\n");
busyRdWr();
return false;
}
}
void PcmSender::Play()
{
iMutex.Wait();
iPaused = false;
iLastTimeUs = 0;
iTimeOffsetUs = 0;
iTimer.FireIn(kPeriodMs);
iMutex.Signal();
}
void PcmSender::Stop()
{
iMutex.Wait();
iPaused = true;
iMutex.Signal();
}
void PcmSender::PlayPause()
{
if (iPaused)
Play();
else
Stop();
}
void PcmSender::Restart()
{
iMutex.Wait();
iIndex = 0;
iMutex.Signal();
}
void PcmSender::CalculatePacketBytes()
{
// in order to let wavsender change the playback rate,
// we keep constant it's idea of how much audio time is in each packet,
// but vary the amount of data that is actually sent
// calculate the amount of time in each packet
TUint norm_bytes = (m_audio->sampleRate() * m_audio->bytesPerFrame() *
kPeriodMs) / 1000;
if (norm_bytes > kMaxPacketBytes) {
norm_bytes = kMaxPacketBytes;
}
TUint norm_packet_samples = norm_bytes / m_audio->bytesPerFrame();
iPacketTime = (norm_packet_samples*1000000/(m_audio->sampleRate()/10) + 5)/10;
// calculate the adjusted speed packet size
TUint bytes = (norm_bytes * iSpeed) / 100;
if (bytes > kMaxPacketBytes) {
bytes = kMaxPacketBytes;
}
iPacketFrames = bytes / m_audio->bytesPerFrame();
iPacketBytes = iPacketFrames * m_audio->bytesPerFrame();
}
void PcmSender::busyRdWr()
{
LOGDEB("PcmSender:busyRdWr: packetbytes " << iPacketBytes << endl);
while (true) {
if (g_quitrequest) {
return;
}
ssize_t nread = 0;
const unsigned char *cp = m_audio->data((size_t)iPacketBytes, nread);
if (cp == 0) {
return;
}
if (g_quitrequest) {
return;
}
#ifdef HAVE_SENDAUDIO_PAUSE_FLAG
iDriver->SendAudio(cp, iPacketBytes, iPaused);
#else
iDriver->SendAudio(cp, iPacketBytes);
#endif
}
}
void PcmSender::TimerExpired()
{
iMutex.Wait();
ssize_t nread = 0;
const unsigned char *cp = m_audio->data((size_t)iPacketBytes, nread);
// If streaming was stopped still read the pipe. Assume, that receiving
// less or no data from the pipe means no further audio data will follow.
// This way, we can set the halt flag in the audio message to indicate
// that the stream of audio has completed.
bool eos = iPaused && ((nread == -1) || (nread < iPacketBytes));
if ((nread > 0) && ((size_t)nread < iPacketBytes))
LOGDEB("PcmSender::TimerExpired: requested " << iPacketBytes
<< " bytes, read " << nread << " bytes" << endl);
if (eos) {
LOGDEB("PcmSender::TimerExpired: EOS, nread=" << nread << endl);
}
if (cp == 0) {
static bool sigsent = false;
if (!sigsent) {
LOGDEB("PcmSender::TimerExpired: killing myself\n");
kill(getpid(), SIGUSR1);
sigsent = true;
}
} else if (nread > 0) {
#ifdef HAVE_SENDAUDIO_PAUSE_FLAG
iDriver->SendAudio(cp, iPacketBytes, eos || g_quitrequest);
#else
iDriver->SendAudio(cp, iPacketBytes);
#endif
}
if (!eos && !g_quitrequest) {
TUint64 now = OsTimeInUs(iEnv.OsCtx());
if (!iPaced) {
// Means we're doing blocking reads on the source, and
// it's setting the pace. I'd like to actually use 0 here
// (ala qt processEvents()), but this appears to busyloop
// and not let the sender do its thing. Anyway, as long
// as we can read from the fifo in much less than (period-2),
// which should always be true, we should be ok.
// I can see not much difference between doing this or
// hijacking the main thread for busy read/write
iTimer.FireIn(2);
} else {
// skip the first packet, and any time the clock value wraps
if (iLastTimeUs && iLastTimeUs < now) {
// will contain the new time out in ms
TUint new_timer_ms = kPeriodMs;
// the difference in usec from where we should be
TInt32 diff = (TInt32)(now - iLastTimeUs) - iPacketTime;
// increment running offset
iTimeOffsetUs -= diff;
// determine new timer value based upon current offset from ideal
if (iTimeOffsetUs < -1000) {
// we are late
TInt32 time_offset_ms = iTimeOffsetUs/1000;
if (time_offset_ms < 1-(TInt32)kPeriodMs) {
// in case callback is severely late, we can only
// catch up so much
new_timer_ms = 1;
} else {
new_timer_ms = kPeriodMs + time_offset_ms;
}
} else if (iTimeOffsetUs > 1000) {
// we are early
new_timer_ms = kPeriodMs+1;
} else {
// we are about on time
new_timer_ms = kPeriodMs;
}
// set timer
iTimer.FireIn(new_timer_ms);
// logging
if (iVerbose) {
if (iTimeOffsetUs >= 1000)
printf ("tnow:%d tlast:%d actual:%4d diff:%4d offset:%5d timer:%d\n", (TUint)now, (TUint)iLastTimeUs, (TUint)(now-iLastTimeUs), diff, iTimeOffsetUs, new_timer_ms);
else
printf ("tnow:%d tlast:%d actual:%4d diff:%4d offset:%4d timer:%d\n", (TUint)now, (TUint)iLastTimeUs, (TUint)(now-iLastTimeUs), diff, iTimeOffsetUs, new_timer_ms);
}
} else {
iTimer.FireIn(kPeriodMs);
}
iLastTimeUs = now;
}
} else {
LOGDEB("PcmSender::TimerExpired: Stop firing\n");
}
iMutex.Signal();
}
PcmSender::~PcmSender()
{
iTimer.Cancel();
delete (iSender);
delete (iDriver);
}
static char *thisprog;
static char usage [] =
" -h, --help, show this help message and exit.\n"
" -A, --audio, [44100:16:2:0/1] audio params only if they can't be obtained\n"
" from the file. Conflicting values will cause error."
" -a, --adapter, [adapter] index of network adapter to use.\n"
" -c, --channel, [0..65535] sender channel.\n"
" -d, --disabled, [disabled] start up disabled.\n"
" -f, --file, [file] file name to read and send.\n"
" Use xx.wav for an actual wav,\n"
" xx or xx.fifo for a fifo, stdin for stdin.\n"
" -l, --latency, [latency] latency in ms.\n"
" -m, --multicast, [multicast] use multicast instead of unicast.\n"
" -p, --pace, Use internal timer to pace source. Implicit for regular files.\n"
" -n, --name, [name] name of the sender.\n"
" -t, --ttl, [ttl] ttl.\n"
" -u, --udn, [udn] udn for the upnp device.\n"
;
static void
Usage(FILE *fp = stderr)
{
fprintf(fp, "%s: usage:\n%s", thisprog, usage);
exit(1);
}
#include <getopt.h>
static int op_flags;
#define OPT_A 0x1
#define OPT_a 0x2
#define OPT_c 0x4
#define OPT_d 0x8
#define OPT_f 0x10
#define OPT_l 0x20
#define OPT_m 0x40
#define OPT_n 0x80
#define OPT_p 0x100
#define OPT_t 0x200
#define OPT_u 0x400
static struct option long_options[] = {
{"audio", required_argument, 0, 'A'},
{"adapter", required_argument, 0, 'a'},
{"channel", required_argument, 0, 'c'},
{"disabled", 0, 0, 'd'},
{"file", required_argument, 0, 'f'},
{"latency", required_argument, 0, 'l'},
{"multicast", 0, 0, 'm'},
{"name", required_argument, 0, 'n'},
{"pace", 0, 0, 'p'},
{"ttl", required_argument, 0, 't'},
{"udn", required_argument, 0, 'u'},
{0, 0, 0, 0}
};
int main(int argc, char **argv)
{
thisprog = argv[0];
int ret;
(void)op_flags;
string audioparams, sfile, sname("Openhome WavSender"), sudn("12345678");
unsigned int adaptidx(0), channel(0), ttl(1), latency(100);
while ((ret = getopt_long(argc, argv, "A:a:c:df:l:mn:pt:u:",
long_options, NULL)) != -1) {
switch (ret) {
case 'A': audioparams = optarg;op_flags |= OPT_A; break;
case 'a': adaptidx = atoi(optarg);op_flags |= OPT_a; break;
case 'c': channel = atoi(optarg);op_flags |= OPT_c; break;
case 'd': op_flags |= OPT_d; break;
case 'f': sfile = optarg;op_flags |= OPT_f; break;
case 'h': Usage(stdout);break;
case 'l': latency = atoi(optarg); op_flags |= OPT_l; break;
case 'm': op_flags |= OPT_m; break;
case 'n': sname = optarg;op_flags |= OPT_n; break;
case 'p': op_flags |= OPT_p; break;
case 't': ttl = atoi(optarg);op_flags |= OPT_t; break;
case 'u': sudn = optarg;op_flags |= OPT_u; break;
default:
Usage();
}
}
if (optind != argc )
Usage();
InitialisationParams* initParams = InitialisationParams::Create();
Library* lib = new Library(initParams);
std::vector<NetworkAdapter*>* subnetList = lib->CreateSubnetList();
LOGDEB("adapter list:\n");
for (unsigned i=0; i<subnetList->size(); ++i) {
TIpAddress addr = (*subnetList)[i]->Address();
LOGDEB(" " << i << ": " << (addr&0xff) << "." <<
((addr>>8)&0xff) << "." << ((addr>>16)&0xff) << "." <<
((addr>>24)&0xff) << endl);
}
if (subnetList->size() <= adaptidx) {
LOGERR("ERROR: adapter " << adaptidx << "doesn't exist\n");
return (1);
}
TIpAddress subnet = (*subnetList)[adaptidx]->Subnet();
TIpAddress adapter = (*subnetList)[adaptidx]->Address();
Library::DestroySubnetList(subnetList);
lib->SetCurrentSubnet(subnet);
LOGDEB("using subnet " << (subnet&0xff) << "." << ((subnet>>8)&0xff) << "."
<< ((subnet>>16)&0xff) << "." << ((subnet>>24)&0xff) << endl);
if (sfile.empty()) {
LOGERR("No input file specified\n");
return (1);
}
Brhz file(sfile.c_str());
Brhz udn(sudn.c_str());
Brhz name(sname.c_str());
TBool multicast = op_flags & OPT_m;
TBool disabled = op_flags & OPT_d;
TBool needpace = op_flags & OPT_p;
AudioReader *audio = openAudio(sfile, audioparams, !needpace);
if (!audio || !audio->open()) {
cerr << "Audio file open failed" << endl;
return 1;
}
needpace = !audio->isblocking();
LOGDEB("sample rate: " << audio->sampleRate() << endl);
LOGDEB("sample size: " << audio->bytesPerSample() << endl);
LOGDEB("channels: " << audio->numChannels() << endl);
DvStack* dvStack = lib->StartDv();
DvDeviceStandard* device = new DvDeviceStandard(*dvStack, udn);
device->SetAttribute("Upnp.Domain", "av.openhome.org");
device->SetAttribute("Upnp.Type", "Sender");
device->SetAttribute("Upnp.Version", "1");
device->SetAttribute("Upnp.FriendlyName", name.CString());
device->SetAttribute("Upnp.Manufacturer", "Openhome");
device->SetAttribute("Upnp.ManufacturerUrl", "http://www.openhome.org");
device->SetAttribute("Upnp.ModelDescription", "Openhome WavSender");
device->SetAttribute("Upnp.ModelName", "Openhome WavSender");
device->SetAttribute("Upnp.ModelNumber", "1");
device->SetAttribute("Upnp.ModelUrl", "http://www.openhome.org");
device->SetAttribute("Upnp.SerialNumber", "");
device->SetAttribute("Upnp.Upc", "");
OhmSenderDriver* driver = new OhmSenderDriver(lib->Env());
Brn icon(icon_png, icon_png_len);
OhmSender* sender =
new OhmSender(lib->Env(), *device, *driver, name, channel, adapter, ttl,
latency, multicast, !disabled, icon, Brn("image/png"), 0);
PcmSender* pcmsender = new PcmSender(lib->Env(), sender, driver, file,
audio, needpace);
device->SetEnabled();
const Brx& suri(sender->SenderUri());
string uri((const char*)suri.Ptr(), suri.Bytes());
const Brx& smeta(sender->SenderMetadata());
string meta((const char*)smeta.Ptr(), smeta.Bytes());
cout << "URI " << UPnPP::base64_encode(uri) <<
" METADATA " << UPnPP::base64_encode(meta) << endl;
// cout << "URI " << uri << " METADATA " << meta << endl;
cout.flush();
signal(SIGUSR1, sigcatcher);
signal(SIGINT, sigcatcher);
signal(SIGTERM, sigcatcher);
if (pcmsender->Start(!disabled)) {
HandleUserCmd(sender, pcmsender);
// If HandleUserCmd() fails with an error we do not want to exit, but
// wait for termination.
if (!g_quitrequest)
pause();
else
// Allow the receivers to receive the halt flag we set When
// we set g_quitrequest.
sleep(1);
}
LOGDEB("Main: cleaning up\n");
delete (pcmsender);
delete (device);
UpnpLibrary::Close();
return (0);
}