u/joerg-krause / sc2mpd - Code / Diff of /src/sc2mpd.cpp

Diff of /src/sc2mpd.cpp [ae9125] .. [b4a189]

Switch to unified view


...
#include <sys/stat.h>
#include <fcntl.h>

using namespace std;

WorkQueue<AudioMessage*> audioqueue("audioqueue", 4);

#ifdef _WIN32

#pragma warning(disable:4355) // use of 'this' in ctor lists safe in this case

...
using namespace OpenHome::Av;


class OhmReceiverDriver : public IOhmReceiverDriver, public IOhmMsgProcessor {
public:
    OhmReceiverDriver(AudioEater* eater, int port);

private:
    // IOhmReceiverDriver
    virtual void Add(OhmMsg& aMsg);
    virtual void Timestamp(OhmMsg& aMsg);
...
            iReset = true;
        }

        void process(OhmMsgAudio& aMsg);
    };
    Observer m_obs;
    AudioEater *m_eater;
};

OhmReceiverDriver::OhmReceiverDriver(AudioEater *eater, int port)
    : m_eater(eater)
{
    AudioEater::Context *ctxt = new AudioEater::Context(&audioqueue, port);
    audioqueue.start(1, m_eater->worker, ctxt);
}

void OhmReceiverDriver::Add(OhmMsg& aMsg)
{
    aMsg.Process(*this);
...
    LOGDEB("=== STARTED ====\n");
}

void OhmReceiverDriver::Connected()
{
    m_obs.reset();
    LOGDEB("=== CONNECTED ====\n");
}

void OhmReceiverDriver::Playing()
{
...
void OhmReceiverDriver::Stopped()
{
    LOGDEB("=== STOPPED ====\n");
}

// Debug and stats only, not needed for main function
void OhmReceiverDriver::Observer::process(OhmMsgAudio& aMsg)
{
    if (++iCount == 400 || aMsg.Halt()) {
        static unsigned long long last_timestamp;
        unsigned long long timestamp = aMsg.MediaTimestamp();
...
    }
}

void OhmReceiverDriver::Process(OhmMsgAudio& aMsg)
{

    if (aMsg.Audio().Bytes() == 0) {
        LOGDEB("OhmReceiverDriver::Process: empty message\n");
        return;
    }

    m_obs.process(aMsg);

    unsigned int bytes = aMsg.Audio().Bytes();
    char *buf = (char *)malloc(bytes);
    if (buf == 0) {
        LOGERR("OhmReceiverDriver::Process: can't allocate " << 
               bytes << " bytes\n");
        return;
    }

    // Songcast data is always msb-first.  Convert to desired order:
    // depends on what downstream wants, and just as well we do it
    // here because we copy the buf anyway.
    bool needswap = false;
    switch (m_eater->input_border) {
    case AudioEater::BO_MSB: break;
    case AudioEater::BO_LSB: needswap = true; break;
    case AudioEater::BO_HOST:
#ifdef WORDS_BIGENDIAN
        needswap = false;
#else
        needswap = true;
#endif
    }

    if (needswap) {
        unsigned char *ocp = (unsigned char *)buf;
        const unsigned char *icp = 
            (const unsigned char *)aMsg.Audio().Ptr();
        const unsigned char *icp0 = icp;
        if (aMsg.BitDepth() == 16) {
            swab(aMsg.Audio().Ptr(), buf, bytes);
        } else if (aMsg.BitDepth() == 24) {
            while (icp - icp0 <= int(bytes) - 3) {
                *ocp++ = icp[2];
                *ocp++ = icp[1];
                *ocp++ = *icp;
                icp += 3;
            }
        } else if (aMsg.BitDepth() == 32) {
            // Never seen this but whatever...



            while (icp - icp0 <= int(bytes) - 4) {
                *ocp++ = icp[3];
                *ocp++ = icp[2];
                *ocp++ = icp[1];
                *ocp++ = *icp;
                icp += 4;
            }
        }
    } else {
        memcpy(buf, aMsg.Audio().Ptr(), bytes);
    }

    AudioMessage *ap = new 
        AudioMessage(aMsg.BitDepth(), aMsg.Channels(), aMsg.Samples(),
                     aMsg.SampleRate(), buf);

    // There is nothing special we can do if put fails: no way to
    // return status. Should we just exit ?
    if (!audioqueue.put(ap, false)) {
    }
}

void OhmReceiverDriver::Process(OhmMsgTrack& aMsg)
{
...

    OptionString optionConfig("-c", "--config", Brn("/etc/upmpdcli.conf"), 
                              "[config] upmpdcli configuration file path");
    parser.AddOption(&optionConfig);

    OptionBool optionDevice("-d", "--direct-alsa", 
                            "[stream] Use alsa directly instead of producing "
                            "http stream");
    parser.AddOption(&optionDevice);

    if (!parser.Parse(aArgc, aArgv)) {
        return (1);
    }

    InitialisationParams* initParams = InitialisationParams::Create();
...

    LOGINF("scmpdcli: using subnet " << (subnet & 0xff) << "." << 
           ((subnet >> 8) & 0xff) << "." << ((subnet >> 16) & 0xff) << "." <<
           ((subnet >> 24) & 0xff) << endl);

    OhmReceiverDriver* driver = new OhmReceiverDriver(
        optionDevice.Value() ? &alsaAudioEater : &httpAudioEater, port);

    OhmReceiver* receiver = new OhmReceiver(lib->Env(), adapter, ttl, *driver);

    CpStack* cpStack = lib->StartCp(subnet);
    cpStack = cpStack; // avoid unused variable warning

	a/src/sc2mpd.cpp		b/src/sc2mpd.cpp
	...		...
49	#include <sys/stat.h>	49	#include <sys/stat.h>
50	#include <fcntl.h>	50	#include <fcntl.h>
51		51
52	using namespace std;	52	using namespace std;
53		53
54	WorkQueue<AudioMessage*> audioqueue("audioqueue", 2);	54	WorkQueue<AudioMessage*> audioqueue("audioqueue", 4);
55		55
56	#ifdef _WIN32	56	#ifdef _WIN32
57		57
58	#pragma warning(disable:4355) // use of 'this' in ctor lists safe in this case	58	#pragma warning(disable:4355) // use of 'this' in ctor lists safe in this case
59		59
	...		...
93	using namespace OpenHome::Av;	93	using namespace OpenHome::Av;
94		94
95		95
96	class OhmReceiverDriver : public IOhmReceiverDriver, public IOhmMsgProcessor {	96	class OhmReceiverDriver : public IOhmReceiverDriver, public IOhmMsgProcessor {
97	public:	97	public:
98	OhmReceiverDriver(int port);	98	OhmReceiverDriver(AudioEater* eater, int port);
99		99
100	private:	100	private:
101	// IOhmReceiverDriver	101	// IOhmReceiverDriver
102	virtual void Add(OhmMsg& aMsg);	102	virtual void Add(OhmMsg& aMsg);
103	virtual void Timestamp(OhmMsg& aMsg);	103	virtual void Timestamp(OhmMsg& aMsg);
	...		...
135	iReset = true;	135	iReset = true;
136	}	136	}
137		137
138	void process(OhmMsgAudio& aMsg);	138	void process(OhmMsgAudio& aMsg);
139	};	139	};
140	Observer obs;	140	Observer m_obs;
		141	AudioEater *m_eater;
141	};	142	};
142		143
143	OhmReceiverDriver::OhmReceiverDriver(int port)	144	OhmReceiverDriver::OhmReceiverDriver(AudioEater *eater, int port)
		145	: m_eater(eater)
144	{	146	{
145	AudioEaterContext *ctxt = new AudioEaterContext(port);	147	AudioEater::Context *ctxt = new AudioEater::Context(&audioqueue, port);
146	audioqueue.start(1, &audioEater, ctxt);	148	audioqueue.start(1, m_eater->worker, ctxt);
147	}	149	}
148		150
149	void OhmReceiverDriver::Add(OhmMsg& aMsg)	151	void OhmReceiverDriver::Add(OhmMsg& aMsg)
150	{	152	{
151	aMsg.Process(*this);	153	aMsg.Process(*this);
	...		...
161	LOGDEB("=== STARTED ====\n");	163	LOGDEB("=== STARTED ====\n");
162	}	164	}
163		165
164	void OhmReceiverDriver::Connected()	166	void OhmReceiverDriver::Connected()
165	{	167	{
166	obs.reset();	168	m_obs.reset();
167	LOGDEB("=== CONNECTED ====\n");	169	LOGDEB("=== CONNECTED ====\n");
168	}	170	}
169		171
170	void OhmReceiverDriver::Playing()	172	void OhmReceiverDriver::Playing()
171	{	173	{
	...		...
182	void OhmReceiverDriver::Stopped()	184	void OhmReceiverDriver::Stopped()
183	{	185	{
184	LOGDEB("=== STOPPED ====\n");	186	LOGDEB("=== STOPPED ====\n");
185	}	187	}
186		188
		189	// Debug and stats only, not needed for main function
187	void OhmReceiverDriver::Observer::process(OhmMsgAudio& aMsg)	190	void OhmReceiverDriver::Observer::process(OhmMsgAudio& aMsg)
188	{	191	{
189	if (++iCount == 400 \|\| aMsg.Halt()) {	192	if (++iCount == 400 \|\| aMsg.Halt()) {
190	static unsigned long long last_timestamp;	193	static unsigned long long last_timestamp;
191	unsigned long long timestamp = aMsg.MediaTimestamp();	194	unsigned long long timestamp = aMsg.MediaTimestamp();
	...		...
252	}	255	}
253	}	256	}
254		257
255	void OhmReceiverDriver::Process(OhmMsgAudio& aMsg)	258	void OhmReceiverDriver::Process(OhmMsgAudio& aMsg)
256	{	259	{
257
258	if (aMsg.Audio().Bytes() == 0) {	260	if (aMsg.Audio().Bytes() == 0) {
259	LOGDEB("OhmReceiverDriver::Process: empty message\n");	261	LOGDEB("OhmReceiverDriver::Process: empty message\n");
260	return;	262	return;
261	}	263	}
		264
		265	m_obs.process(aMsg);
262		266
263	unsigned int bytes = aMsg.Audio().Bytes();	267	unsigned int bytes = aMsg.Audio().Bytes();
264	char buf = (char )malloc(bytes);	268	char buf = (char )malloc(bytes);
265	if (buf == 0) {	269	if (buf == 0) {
266	LOGERR("OhmReceiverDriver::Process: can't allocate " <<	270	LOGERR("OhmReceiverDriver::Process: can't allocate " <<
267	bytes << " bytes\n");	271	bytes << " bytes\n");
268	return;	272	return;
269	}	273	}
270		274
		275	// Songcast data is always msb-first. Convert to desired order:
		276	// depends on what downstream wants, and just as well we do it
		277	// here because we copy the buf anyway.
		278	bool needswap = false;
		279	switch (m_eater->input_border) {
		280	case AudioEater::BO_MSB: break;
		281	case AudioEater::BO_LSB: needswap = true; break;
		282	case AudioEater::BO_HOST:
271	#ifdef WORDS_BIGENDIAN	283	#ifdef WORDS_BIGENDIAN
272	memcpy(buf, aMsg.Audio().Ptr(), bytes);	284	needswap = false;
273	#else	285	#else
274	if (aMsg.BitDepth() == 16) {	286	needswap = true;
275	swab(aMsg.Audio().Ptr(), buf, bytes);	287	#endif
276	} else if (aMsg.BitDepth() == 24) {	288	}
		289
		290	if (needswap) {
277	unsigned char ocp = (unsigned char )buf;	291	unsigned char ocp = (unsigned char )buf;
		292	const unsigned char *icp =
278	const unsigned char icp = (const unsigned char )aMsg.Audio().Ptr();	293	(const unsigned char *)aMsg.Audio().Ptr();
279	const unsigned char *icp0 = icp;	294	const unsigned char *icp0 = icp;
		295	if (aMsg.BitDepth() == 16) {
		296	swab(aMsg.Audio().Ptr(), buf, bytes);
		297	} else if (aMsg.BitDepth() == 24) {
280	while (icp - icp0 <= int(bytes) - 3) {	298	while (icp - icp0 <= int(bytes) - 3) {
281	*ocp++ = icp[2];	299	*ocp++ = icp[2];
282	*ocp++ = icp[1];	300	*ocp++ = icp[1];
283	ocp++ = icp;	301	ocp++ = icp;
284	icp += 3;	302	icp += 3;
285	}	303	}
286	} else if (aMsg.BitDepth() == 32) {	304	} else if (aMsg.BitDepth() == 32) {
287	// Never seen this but whatever...	305	// Never seen this but whatever...
288	unsigned char ocp = (unsigned char )buf;
289	const unsigned char icp = (const unsigned char )aMsg.Audio().Ptr();
290	const unsigned char *icp0 = icp;
291	while (icp - icp0 <= int(bytes) - 4) {	306	while (icp - icp0 <= int(bytes) - 4) {
292	*ocp++ = icp[3];	307	*ocp++ = icp[3];
293	*ocp++ = icp[2];	308	*ocp++ = icp[2];
294	*ocp++ = icp[1];	309	*ocp++ = icp[1];
295	ocp++ = icp;	310	ocp++ = icp;
296	icp += 4;	311	icp += 4;
297	}	312	}
298	}	313	}
299	#endif	314	} else {
		315	memcpy(buf, aMsg.Audio().Ptr(), bytes);
		316	}
300		317
301	AudioMessage *ap = new	318	AudioMessage *ap = new
302	AudioMessage(aMsg.BitDepth(), aMsg.Channels(), aMsg.Samples(),	319	AudioMessage(aMsg.BitDepth(), aMsg.Channels(), aMsg.Samples(),
303	aMsg.SampleRate(), buf);	320	aMsg.SampleRate(), buf);
304		321
305	// There is nothing special we can do if put fails: no way to	322	// There is nothing special we can do if put fails: no way to
306	// return status. Should we just exit ?	323	// return status. Should we just exit ?
307	if (!audioqueue.put(ap, true)) {	324	if (!audioqueue.put(ap, false)) {
308	}	325	}
309	}	326	}
310		327
311	void OhmReceiverDriver::Process(OhmMsgTrack& aMsg)	328	void OhmReceiverDriver::Process(OhmMsgTrack& aMsg)
312	{	329	{
	...		...
349		366
350	OptionString optionConfig("-c", "--config", Brn("/etc/upmpdcli.conf"),	367	OptionString optionConfig("-c", "--config", Brn("/etc/upmpdcli.conf"),
351	"[config] upmpdcli configuration file path");	368	"[config] upmpdcli configuration file path");
352	parser.AddOption(&optionConfig);	369	parser.AddOption(&optionConfig);
353		370
		371	OptionBool optionDevice("-d", "--direct-alsa",
		372	"[stream] Use alsa directly instead of producing "
		373	"http stream");
		374	parser.AddOption(&optionDevice);
		375
354	if (!parser.Parse(aArgc, aArgv)) {	376	if (!parser.Parse(aArgc, aArgv)) {
355	return (1);	377	return (1);
356	}	378	}
357		379
358	InitialisationParams* initParams = InitialisationParams::Create();	380	InitialisationParams* initParams = InitialisationParams::Create();
	...		...
398		420
399	LOGINF("scmpdcli: using subnet " << (subnet & 0xff) << "." <<	421	LOGINF("scmpdcli: using subnet " << (subnet & 0xff) << "." <<
400	((subnet >> 8) & 0xff) << "." << ((subnet >> 16) & 0xff) << "." <<	422	((subnet >> 8) & 0xff) << "." << ((subnet >> 16) & 0xff) << "." <<
401	((subnet >> 24) & 0xff) << endl);	423	((subnet >> 24) & 0xff) << endl);
402		424
403	OhmReceiverDriver* driver = new OhmReceiverDriver(port);	425	OhmReceiverDriver* driver = new OhmReceiverDriver(
		426	optionDevice.Value() ? &alsaAudioEater : &httpAudioEater, port);
404		427
405	OhmReceiver* receiver = new OhmReceiver(lib->Env(), adapter, ttl, *driver);	428	OhmReceiver* receiver = new OhmReceiver(lib->Env(), adapter, ttl, *driver);
406		429
407	CpStack* cpStack = lib->StartCp(subnet);	430	CpStack* cpStack = lib->StartCp(subnet);
408	cpStack = cpStack; // avoid unused variable warning	431	cpStack = cpStack; // avoid unused variable warning