Commit 39acba48 authored by Jean-Francois Dockes's avatar Jean-Francois Dockes
Browse files

ckpt: notifyEvent mostly working, checks and cleanup needed

parent ae97062d
......@@ -106,6 +106,8 @@ UpMpdAVTransport::UpMpdAVTransport(
#endif
m_autoplay = g_config->getBool("avtautoplay", false);
keepconsume = g_config->getBool("keepconsume", false);
m_dev->getmpdcli()->subscribe(
MPDCli::MpdPlayerEvt, bind(&UpMpdAVTransport::onMpdEvent, this, _1));
}
// AVTransport Errors
......@@ -394,6 +396,16 @@ bool UpMpdAVTransport::getEventData(bool all, std::vector<std::string>& names,
return true;
}
void UpMpdAVTransport::onMpdEvent(const MpdStatus*)
{
LOGDEB0("AVTransport::onMpdEvent()\n");
std::vector<std::string> names, values;
getEventData(false, names, values);
if (!names.empty()) {
m_udev->notifyEvent(this, names, values);
}
}
// http://192.168.4.4:8200/MediaItems/246.mp3
int UpMpdAVTransport::setAVTransportURI(const SoapIncoming& sc,
SoapOutgoing& data, bool setnext)
......@@ -541,7 +553,6 @@ int UpMpdAVTransport::setAVTransportURI(const SoapIncoming& sc,
m_songids.insert(songid);
}
m_udev->loopWakeup();
return UPNP_E_SUCCESS;
}
......@@ -699,7 +710,6 @@ int UpMpdAVTransport::playcontrol(const SoapIncoming& sc, SoapOutgoing& data, in
}
break;
}
m_udev->loopWakeup();
return ok ? UPNP_E_SUCCESS : UPNP_E_INTERNAL_ERROR;
}
......@@ -719,7 +729,6 @@ int UpMpdAVTransport::seqcontrol(const SoapIncoming& sc, SoapOutgoing& data, int
case 0: ok = m_dev->getmpdcli()->next();break;
case 1: ok = m_dev->getmpdcli()->previous();break;
}
m_udev->loopWakeup();
return ok ? UPNP_E_SUCCESS : UPNP_E_INTERNAL_ERROR;
}
......@@ -775,7 +784,6 @@ int UpMpdAVTransport::setPlayMode(const SoapIncoming& sc, SoapOutgoing& data)
} else {
return UPNP_E_INVALID_PARAM;
}
m_udev->loopWakeup();
return ok ? UPNP_E_SUCCESS : UPNP_E_INTERNAL_ERROR;
}
......@@ -826,7 +834,6 @@ int UpMpdAVTransport::seek(const SoapIncoming& sc, SoapOutgoing& data)
LOGDEB("UpMpdAVTransport::seek: seeking to " << abs_seconds <<
" seconds (" << upnpduration(abs_seconds * 1000) << ")" << endl);
m_udev->loopWakeup();
return m_dev->getmpdcli()->seek(abs_seconds) ?
UPNP_E_SUCCESS : UPNP_E_INTERNAL_ERROR;
}
......@@ -28,6 +28,7 @@
class OHPlaylist;
class UpMpdMediaRenderer;
class UpMpd;
class MpdStatus;
using namespace UPnPP;
......@@ -39,6 +40,7 @@ public:
std::vector<std::string>& values);
virtual const std::string serviceErrString(int) const;
void onMpdEvent(const MpdStatus*);
void setOHP(OHPlaylist *ohp) {
m_ohp = ohp;
......
......@@ -157,6 +157,7 @@ ohProductDesc_t ohProductDesc = {
// Static for cleanup in sig handler.
static vector<UpnpDevice *> devs;
static MPDCli *mpdclip{nullptr};
string g_datadir(DATADIR "/");
string g_cachedir("/var/cache/upmpdcli");
......@@ -174,6 +175,9 @@ static void onsig(int)
for (auto& dev : devs) {
dev->shouldExit();
}
if (mpdclip) {
mpdclip->shouldExit();
}
}
static const int catchedSigs[] = {SIGINT, SIGQUIT, SIGTERM};
......@@ -640,7 +644,6 @@ int main(int argc, char *argv[])
// Initialize MPD client object. Retry until it works or power fail.
MPDCli *mpdclip = 0;
if (!msonly) {
int mpdretrysecs = 2;
for (;;) {
......@@ -783,14 +786,17 @@ int main(int argc, char *argv[])
} else if (enableMediaServer) {
startMsOnlyProcess();
}
if (!msonly) {
LOGDEB("Renderer event loop" << endl);
mediarenderer->startloops();
}
if (inprocessms && enableMediaServer) {
mediaserver->startloop();
}
if (!msonly) {
LOGDEB("Renderer event loop" << endl);
mediarenderer->startnoloops();
// mediarenderer->startloops();
mpdclip->startEventLoop();
}
pause();
LOGDEB("Event loop returned" << endl);
return 0;
}
......
......@@ -67,6 +67,7 @@ MPDCli::MPDCli(const string& host, int port, const string& pass)
MPDCli::~MPDCli()
{
std::unique_lock<std::mutex> lock(m_mutex);
stopEventLoop();
closeconn();
regfree(&m_tpuexpr);
}
......@@ -131,6 +132,140 @@ bool MPDCli::openconn()
return true;
}
bool MPDCli::startEventLoop()
{
LOGDEB("MPDCli::startEventLoop\n");
if (nullptr == m_idleconn) {
m_idlethread = std::thread(bind(&MPDCli::eventLoop, this));
} else {
LOGINF("MPDCli::startEventLoop: already started\n");
}
return true;
}
void MPDCli::stopEventLoop()
{
LOGDEB("MPDCli::stopEventLoop\n");
if (m_idleconn) {
mpd_send_noidle(m_idleconn);
m_idlethread.join();
m_idlethread = std::thread{};
}
}
bool MPDCli::takeEvents(MPDCli *from)
{
if (from->m_idleconn) {
from->stopEventLoop();
m_subs = from->m_subs;
return startEventLoop();
} else {
return true;
}
}
void MPDCli::pollerCtl(MpdStatus::State st)
{
if (st == MpdStatus::MPDS_PLAY) {
LOGDEB("MPDCli::pollerCtl: mpd is playing\n");
std::unique_lock<std::mutex> lock(m_pollmutex);
if (!m_dopoll) {
LOGDEB("MPDCli::eventloop: start polling thread\n");
m_dopoll = true;
m_pollerthread = std::thread(bind(&MPDCli::timepoller, this));
}
} else {
LOGDEB("MPDCli::pollerCtl: mpd is not playing\n");
std::unique_lock<std::mutex> lock(m_pollmutex);
m_dopoll = false;
m_pollcv.notify_all();
}
if (!m_dopoll && m_pollerthread.joinable()) {
m_pollerthread.join();
}
}
static int o_idle_mask =
MPD_IDLE_QUEUE | /* the queue has been modified */
MPD_IDLE_PLAYER | /* the player state has changed: play, stop, ... */
MPD_IDLE_MIXER | /* the volume has been modified */
MPD_IDLE_OPTIONS; /* options have changed: crossfade, random, repeat, ... */
bool MPDCli::eventLoop()
{
m_idleconn = mpd_connection_new(m_host.c_str(), m_port, m_timeoutms);
if (nullptr == m_idleconn) {
LOGERR("MPDCli::eventloop: could not open connection\n");
return false;
}
{
std::unique_lock<std::mutex> lock(m_mutex);
updStatus();
}
pollerCtl(m_stat.state);
for (;;) {
enum mpd_idle mask =
mpd_run_idle_mask(m_idleconn, (enum mpd_idle)o_idle_mask);
if (mask == 0) {
LOGERR("MPDCli::eventloop: mpd_run_idle_mask returned 0\n");
break;
}
LOGDEB0("MPDCli::eventloop: mpd_run_idle_mask returned " << std::hex <<
mask << std::dec << "\n");
{
std::unique_lock<std::mutex> lock(m_mutex);
updStatus();
}
pollerCtl(m_stat.state);
for (auto& sub : m_subs) {
if (sub.first & mask) {
sub.second(&m_stat);
}
}
}
pollerCtl(MpdStatus::MPDS_STOP);
if (m_idleconn) {
mpd_connection_free(m_idleconn);
m_idleconn = nullptr;
}
return false;
}
void MPDCli::timepoller()
{
for (;;) {
LOGDEB0("MPDCli::timepoller\n");
{
std::unique_lock<std::mutex> lock(m_mutex);
updStatus();
}
for (auto& sub : m_subs) {
if (sub.first & MpdPlayerEvt) {
sub.second(&m_stat);
}
}
std::unique_lock<std::mutex> lock(m_pollmutex);
m_pollcv.wait_for(lock, std::chrono::seconds(1));
if (!m_dopoll) {
LOGDEB("MPDCli::timepoller: returning\n");
return;
}
}
}
void MPDCli::shouldExit()
{
LOGDEB("MPDCli::shouldExit\n");
stopEventLoop();
}
bool MPDCli::subscribe(int mask, evtFunc func)
{
m_subs.push_back({mask, func});
return true;
}
// Call with lock held
bool MPDCli::showError(const string& who)
{
......@@ -157,7 +292,7 @@ bool MPDCli::showError(const string& who)
return false;
}
#define RETRY_CMD(CMD, ERROR) { \
#define RETRY_CMD_NOLOCK(CMD, ERROR) { \
if (!ok()) { \
return ERROR; \
} \
......@@ -169,42 +304,51 @@ bool MPDCli::showError(const string& who)
} \
}
#define RETRY_CMD_WITH_SLEEP(CMD, ERROR) { \
if (!ok()) { \
return ERROR; \
} \
for (int i = 0; i < 2; i++) { \
if ((CMD)) \
break; \
sleep(1); \
if (i == 1 || !showError(#CMD)) \
return ERROR; \
} \
}
#define RETRY_CMD(CMD, ERROR) do { \
std::unique_lock<std::mutex> lock(m_connmutex); \
RETRY_CMD_NOLOCK(CMD, ERROR); \
} while (false);
#define RETRY_CMD_WITH_SLEEP(CMD, ERROR) do { \
std::unique_lock<std::mutex> lock(m_connmutex); \
if (!ok()) { \
return ERROR; \
} \
for (int i = 0; i < 2; i++) { \
if ((CMD)) \
break; \
sleep(1); \
if (i == 1 || !showError(#CMD)) \
return ERROR; \
} \
} while (false);
// Call with lock held
bool MPDCli::updStatus()
{
if (!ok() && !openconn()) {
LOGERR("MPDCli::updStatus: no connection" << endl);
return false;
}
mpd_status *mpds = 0;
mpds = mpd_run_status(m_conn);
if (mpds == 0) {
if (!openconn()) {
LOGERR("MPDCli::updStatus: connection failed\n");
{
std::unique_lock<std::mutex> lock(m_connmutex);
if (!ok() && !openconn()) {
LOGERR("MPDCli::updStatus: no connection" << endl);
return false;
}
mpds = mpd_run_status(m_conn);
if (mpds == 0) {
LOGERR("MPDCli::updStatus: can't get status" << endl);
showError("MPDCli::updStatus");
if (!openconn()) {
LOGERR("MPDCli::updStatus: connection failed\n");
return false;
}
mpds = mpd_run_status(m_conn);
if (mpds == 0) {
LOGERR("MPDCli::updStatus: can't get status" << endl);
showError("MPDCli::updStatus");
}
return false;
}
return false;
}
if (m_externalvolumecontrol && !m_getexternalvolume.empty()) {
string result;
if (ExecCmd::backtick(m_getexternalvolume, result)) {
......@@ -288,7 +432,7 @@ bool MPDCli::updStatus()
m_stat.songelapsedms = mpd_status_get_elapsed_ms(mpds);
m_stat.songlenms = mpd_status_get_total_time(mpds) * 1000;
m_stat.kbrate = mpd_status_get_kbit_rate(mpds);
const struct mpd_audio_format *maf =
const struct mpd_audio_format *maf =
mpd_status_get_audio_format(mpds);
if (maf) {
m_stat.bitdepth = maf->bits;
......@@ -322,7 +466,8 @@ bool MPDCli::checkForCommand(const string& cmdname)
{
LOGDEB1("MPDCli::checkForCommand: " << cmdname << endl);
bool found = false;
RETRY_CMD(mpd_send_allowed_commands(m_conn), false);
std::unique_lock<std::mutex> lock(m_connmutex);
RETRY_CMD_NOLOCK(mpd_send_allowed_commands(m_conn), false);
struct mpd_pair *rep;
do {
rep = mpd_recv_command_pair(m_conn);
......@@ -732,6 +877,7 @@ bool MPDCli::send_tag(const char *cid, int tag, const string& _data)
return false;
string data;
neutchars(_data, data, "\r\n", ' ');
std::unique_lock<std::mutex> lock(m_connmutex);
if (!mpd_send_command(m_conn, "addtagid", cid,
mpd_tag_name(mpd_tag_type(tag)),
data.c_str(), NULL)) {
......@@ -870,10 +1016,10 @@ bool MPDCli::deletePosRange(unsigned int start, unsigned int end)
bool MPDCli::statId(int id)
{
LOGDEB("MPDCli::statId " << id << endl);
std::unique_lock<std::mutex> lock(m_mutex);
if (!ok())
return false;
std::unique_lock<std::mutex> clock(m_connmutex);
mpd_song *song = mpd_run_get_queue_song_id(m_conn, (unsigned)id);
if (song) {
mpd_song_free(song);
......@@ -888,7 +1034,8 @@ bool MPDCli::getQueueSongs(vector<mpd_song*>& songs)
//LOGDEB1("MPDCli::getQueueSongs" << endl);
songs.clear();
RETRY_CMD(mpd_send_list_queue_meta(m_conn), false);
std::unique_lock<std::mutex> lock(m_connmutex);
RETRY_CMD_NOLOCK(mpd_send_list_queue_meta(m_conn), false);
struct mpd_song *song;
while ((song = mpd_recv_song(m_conn)) != NULL) {
......
......@@ -22,7 +22,11 @@
#include <cstdio>
#include <vector>
#include <memory>
#include <utility>
#include <functional>
#include <thread>
#include <mutex>
#include <condition_variable>
#include "upmpdutils.hxx"
......@@ -110,10 +114,47 @@ public:
// save (sometimes useful if mpd was stopped)
bool saveState(MpdState& st, int seekms = 0);
bool restoreState(const MpdState& st);
// Event loop. Allows running without polling from libupnpp (no
// call to the device eventloop()). We collect events from MPD
// instead and call the event generating chain when appropriate.
bool eventLoop();
bool startEventLoop();
void stopEventLoop();
bool takeEvents(MPDCli *from);
void shouldExit();
// Event selection mask.
enum SubsSelect {
// Lazy us are using the values from mpd/idle.h. Could translate instead.
MpdQueueEvt = 0x4, /* Queue modified */
MpdPlayerEvt = 0x8, /* Play, stop, etc. + play time, locally generated*/
MpdMixerEvt = 0x10, /* Volume */
MpdOptsEvt = 0x40, /* random, repeat, etc. */
};
// Type of subscription callback
typedef std::function<void (const MpdStatus*)> evtFunc;
// Subscribe to event mask. Called by the services during initialization.
bool subscribe(int mask, evtFunc);
private:
std::mutex m_mutex;
// Main connection for sending commands
std::mutex m_connmutex;
struct mpd_connection *m_conn{nullptr};
// Connection for the event loop
struct mpd_connection *m_idleconn{nullptr};
// thread to listen to MPD events.
std::thread m_idlethread;
// MPD does not report idle events for play time change. We poll
// it every second while playing to report the current time
// (needed for OHTime events for example)
std::thread m_pollerthread;
std::mutex m_pollmutex;
std::condition_variable m_pollcv;
bool m_dopoll{false};
// Event subscriptions
std::vector<std::pair<int, evtFunc>> m_subs;
MpdStatus m_stat;
// Saved volume while muted.
int m_premutevolume{0};
......@@ -165,6 +206,9 @@ private:
bool seek_i(int seconds);
bool single_i(bool on);
bool statSong_i(UpSong& usong, int pos = -1, bool isId = false);
// thread routine for polling play time while playing.
void timepoller();
void pollerCtl(MpdStatus::State st);
};
#endif /* _MPDCLI_H_X_INCLUDED_ */
......@@ -524,6 +524,7 @@ int OHCredentials::actSet(const SoapIncoming& sc, SoapOutgoing& data)
}
m->seq++;
m->save();
onEvent(nullptr);
if (m->setEnabled(in_Id, true)) {
return UPNP_E_SUCCESS;
} else {
......@@ -706,5 +707,6 @@ int OHCredentials::actGetSequenceNumber(const SoapIncoming& sc,
{
LOGDEB("OHCredentials::actGetSequenceNumber: " << endl);
data.addarg("SequenceNumber", SoapHelp::i2s(m->seq));
onEvent(nullptr);
return UPNP_E_SUCCESS;
}
......@@ -170,5 +170,6 @@ void OHInfo::setMetatext(const string& metatext)
if (metatext.compare(m_metatext)) {
m_metatext = metatext;
m_metatextcnt++;
onEvent(nullptr);
}
}
......@@ -117,6 +117,8 @@ OHPlaylist::OHPlaylist(UpMpd *dev, UpMpdOpenHome *udev, unsigned int cssleep)
}
}
keepconsume = g_config->getBool("keepconsume", false);
m_dev->getmpdcli()->subscribe(
MPDCli::MpdPlayerEvt, std::bind(&OHService::onEvent, this, _1));
}
static const int tracksmax = 16384;
......@@ -304,8 +306,9 @@ void OHPlaylist::refreshState()
void OHPlaylist::maybeWakeUp(bool ok)
{
if (ok && m_dev)
m_udev->loopWakeup();
if (ok && m_dev) {
onEvent(nullptr);
}
}
void OHPlaylist::setActive(bool onoff)
......
......@@ -250,7 +250,7 @@ int OHProduct::setStandby(const SoapIncoming& sc, SoapOutgoing& data)
LOGDEB("OHProduct: standby is " << m_standby << endl);
}
}
m_udev->loopWakeup();
onEvent(nullptr);
return UPNP_E_SUCCESS;
}
......@@ -359,7 +359,7 @@ int OHProduct::iSetSourceIndex(int sindex)
if (g_state) {
g_state->set(cstr_stsrcnmkey, newnm);
}
m_udev->loopWakeup();
onEvent(nullptr);
return ok ? UPNP_E_SUCCESS : UPNP_E_INTERNAL_ERROR;
}
......
......@@ -177,6 +177,9 @@ OHRadio::OHRadio(UpMpd *dev, UpMpdOpenHome *udev)
udev->addActionMapping(this, "TransportState",
bind(&OHRadio::transportState, this, _1, _2));
keepconsume = g_config->getBool("keepconsume", false);
m_dev->getmpdcli()->subscribe(
MPDCli::MpdPlayerEvt, std::bind(&OHService::onEvent, this, _1));
}
static void getRadiosFromConf(ConfSimple* conf)
......@@ -400,8 +403,8 @@ bool OHRadio::makestate(unordered_map<string, string>& st)
void OHRadio::maybeWakeUp(bool ok)
{
if (ok && m_udev) {
m_udev->loopWakeup();
if (ok) {
onEvent(nullptr);
}
}
......
......@@ -136,8 +136,9 @@ bool OHReceiver::makestate(unordered_map<string, string> &st)
void OHReceiver::maybeWakeUp(bool ok)
{
if (ok && m_udev)
m_udev->loopWakeup();
if (ok) {
onEvent(nullptr);
}
}
void OHReceiver::setActive(bool onoff)
......
......@@ -20,8 +20,10 @@
#include <string>
#include <unordered_map>
#include <vector>
#include <functional>
#include "libupnpp/device/device.hxx"
#include "libupnpp/log.h"
#include "upmpdutils.hxx"
#include "upmpd.hxx"
#include "mpdcli.hxx"
......@@ -38,6 +40,15 @@ public:
}
virtual ~OHService() { }
virtual void onEvent(const MpdStatus*) {
LOGDEB1("OHService::onEvent()\n");
std::vector<std::string> names, values;
getEventData(false, names, values);
if (!names.empty()) {
m_udev->notifyEvent(this, names, values);
}
}
virtual bool getEventData(bool all, std::vector<std::string>& names,
std::vector<std::string>& values) {
//LOGDEB("OHService::getEventData" << std::endl);
......
......@@ -227,6 +227,7 @@ bool SenderReceiver::start(const string& script, int seekms)
// full scale. Else we are compositing the two volumes.
m->origmpd->setVolume(100);
}
m->dev->getmpdcli()->takeEvents(m->origmpd);
} else {
m->origmpd = 0;
}
......@@ -248,6 +249,7 @@ bool SenderReceiver::stop()
copyMpd(m->mpd, m->origmpd, -1);
m->mpd->stop();
m->dev->setmpdcli(m->origmpd);
m->dev->getmpdcli()->takeEvents(m->mpd);
m->origmpd = 0;
}
......
......@@ -25,6 +25,8 @@
#include <unordered_map> // for unordered_map, etc
#include <utility> // for pair
#include <vector> // for vector
#include <thread>
#include <chrono>
#include "libupnpp/log.hxx" // for LOGDEB