playerradiolive.cc

   1 /*
   2     Copyright 2008-2020 Chris Tallon
   3
   4     This file is part of VOMP.
   5
   6     VOMP is free software; you can redistribute it and/or modify
   7     it under the terms of the GNU General Public License as published by
   8     the Free Software Foundation; either version 2 of the License, or
   9     (at your option) any later version.
  10
  11     VOMP is distributed in the hope that it will be useful,
  12     but WITHOUT ANY WARRANTY; without even the implied warranty of
  13     MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  14     GNU General Public License for more details.
  15
  16     You should have received a copy of the GNU General Public License
  17     along with VOMP.  If not, see <https://www.gnu.org/licenses/>.
  18 */
  19
  20 #include <stdlib.h>
  21 #ifndef WIN32
  22 #include <sys/time.h>
  23 #endif
  24 #include <time.h>
  25
  26 #include "defines.h"
  27 #include "log.h"
  28 #include "audio.h"
  29 #include "demuxerts.h"
  30 #include "vdr.h"
  31 #include "messagequeue.h"
  32 #include "input.h"
  33 #include "message.h"
  34 #include "channel.h"
  35 #include "video.h"
  36
  37 #include "playerradiolive.h"
  38
  39 // ----------------------------------- Called from outside, one offs or info funcs
  40
  41 PlayerRadioLive::PlayerRadioLive(MessageQueue* tmessageQueue, void* tmessageReceiver, ChannelList* tchanList)
  42 : messageQueue(tmessageQueue), messageReceiver(tmessageReceiver), afeed(this), chanList(tchanList)
  43 {
  44   audio = Audio::getInstance();
  45   logger = Log::getInstance();
  46   vdr = VDR::getInstance();
  47
  48   Video::getInstance()->turnVideoOff();
  49 }
  50
  51 PlayerRadioLive::~PlayerRadioLive()
  52 {
  53   if (initted) shutdown();
  54 }
  55
  56 int PlayerRadioLive::init()
  57 {
  58   if (initted) return 0;
  59
  60   demuxer = new DemuxerTS();
  61   if (!demuxer) return 0;
  62
  63   if (!demuxer->init(this, audio, NULL, NULL, 0, 200000, 0))
  64   {
  65     logger->log("PlayerRadioLive", Log::ERR, "Demuxer failed to init");
  66     shutdown();
  67     return 0;
  68   }
  69
  70   audio->stop();
  71
  72   initted = true;
  73   return 1;
  74 }
  75
  76 int PlayerRadioLive::shutdown()
  77 {
  78   if (!initted) return 0;
  79   if (state != S_STOP) // FIXME check when this is called and how. This is not thread-sync bullet proof as-is.
  80   {
  81     logger->log("PlayerRadioLive", Log::DEBUG, "state is not stop, calling");
  82     stop();
  83   }
  84   initted = false;
  85   delete demuxer;
  86   return 1;
  87 }
  88
  89 bool* PlayerRadioLive::getDemuxerMpegAudioChannels()
  90 {
  91   return demuxer->getmpAudioChannels();
  92 }
  93
  94 bool* PlayerRadioLive::getDemuxerAc3AudioChannels()
  95 {
  96   return demuxer->getac3AudioChannels();
  97 }
  98
  99 int PlayerRadioLive::getCurrentAudioChannel()
 100 {
 101   return demuxer->getAID();
 102 }
 103
 104 int* PlayerRadioLive::getTeletxtSubtitlePages()
 105 {
 106   return NULL;
 107 }
 108
 109 int PlayerRadioLive::getCurrentSubtitleChannel()
 110 {
 111   return demuxer->getSubID();
 112 }
 113
 114 void PlayerRadioLive::setAudioChannel(int newChannel, int type, int streamtype)
 115 {
 116   demuxer->setAID(newChannel, type, streamtype, true);
 117 }
 118
 119 void PlayerRadioLive::setSubtitleChannel(int newChannel)
 120 {
 121   demuxer->setSubID(newChannel);
 122 }
 123
 124 // ----------------------------------- Externally called events
 125
 126 void PlayerRadioLive::go(ULONG index)
 127 {
 128   playerThreadMutex.lock();
 129
 130   struct PLInstruction i;
 131   i.instruction = I_SETCHANNEL;
 132   i.channelIndex = index;
 133   instructions.push(i);
 134
 135   playerThread = std::thread([this]
 136   {
 137     playerThreadMutex.lock();
 138     playerThreadMutex.unlock();
 139     threadMethod();
 140   });
 141   playerThreadMutex.unlock();
 142 }
 143
 144 void PlayerRadioLive::setChannel(ULONG index)
 145 {
 146   logger->log("PlayerRadioLive", Log::DEBUG, "setChannel");
 147   struct PLInstruction i;
 148   i.instruction = I_SETCHANNEL;
 149   i.channelIndex = index;
 150   instructions.push(i);
 151   logger->log("PlayerRadioLive", Log::DEBUG, "posted setChannel instruction, now %i in queue", instructions.size());
 152   playerThreadCond.notify_one();
 153 }
 154
 155 void PlayerRadioLive::stop()
 156 {
 157   logger->log("PlayerRadioLive", Log::DEBUG, "stop");
 158   playerThreadMutex.lock();
 159   struct PLInstruction i;
 160   i.instruction = I_STOP;
 161   instructions.push(i);
 162   playerThreadCond.notify_one();
 163   playerThreadMutex.unlock();
 164   playerThread.join();
 165   logger->log("PlayerRadioLive", Log::DEBUG, "stop successful");
 166 }
 167
 168 // ----------------------------------- Callback
 169
 170 void PlayerRadioLive::call(void* /*caller*/)
 171 {
 172 }
 173
 174 // -----------------------------------
 175
 176 void PlayerRadioLive::streamReceive(ULONG flag, void* data, ULONG len)
 177 {
 178   // Flag:
 179   // 0 = normal stream packet
 180   // 1 = stream end
 181   // 2 = connection lost
 182
 183   if (flag == 1)
 184   {
 185     if (data) abort();
 186
 187     Message* m = new Message();
 188     m->from = this;
 189     m->to = messageReceiver;
 190     m->message = Message::PLAYER_EVENT;
 191     m->parameter = PlayerRadioLive::STREAM_END;
 192     messageQueue->postMessage(m);
 193   }
 194
 195   if (streamChunks.size() < 11)
 196   {
 197     StreamChunk s;
 198     s.data = data;
 199     s.len = len;
 200     streamChunks.push(s);
 201     playerThreadCond.notify_one();
 202   }
 203   else
 204   {
 205     // Too many chunks in streamChunks, drop this chunk
 206     free(data);
 207     logger->log("PlayerRadioLive", Log::WARN, "Dropped chunk");
 208   }
 209 }
 210
 211 void PlayerRadioLive::clearStreamChunks()
 212 {
 213   while(streamChunks.size())
 214   {
 215     logger->log("PlayerRadioLive", Log::DEBUG, "Dropping chunk from old stream");
 216     struct StreamChunk s = streamChunks.front();
 217     streamChunks.pop();
 218     free(s.data);
 219   }
 220 }
 221
 222 void PlayerRadioLive::chunkToDemuxer()
 223 {
 224   StreamChunk s = streamChunks.front();
 225   streamChunks.pop();
 226   //logger->log("PlayerRadioLive", Log::DEBUG, "About to call demuxer with %p %lu", s.data, s.len);
 227   /*int a =*/ demuxer->put(static_cast<UCHAR*>(s.data), s.len);
 228   //logger->log("PlayerRadioLive", Log::DEBUG, "put %i to demuxer", a);
 229   free(s.data);
 230 }
 231
 232 void PlayerRadioLive::switchState(UCHAR newState)
 233 {
 234   logger->log("PlayerRadioLive", Log::DEBUG, "Switch from state %u to state %u", state, newState);
 235
 236   switch(state)
 237   {
 238     case S_STOP:   // FROM S_STOP
 239     {
 240       switch(newState)
 241       {
 242         case S_PREBUFFERING:
 243         {
 244           audio->stop();
 245           audio->unPause();
 246           audio->reset();
 247           audio->setStreamType(Audio::MPEG2_PES);
 248           audio->systemMuteOff();
 249           audio->doMuting();
 250           audio->play();
 251           audio->pause();
 252           demuxer->reset();
 253           afeed.start();
 254
 255           state = newState;
 256           preBufferCount = 0;
 257           return;
 258         }
 259         default:
 260         {
 261           logger->log("PlayerRadioLive", Log::EMERG, "Thread called state %u to state %u which is not supported", state, newState);
 262           abort();
 263           break;
 264         }
 265       }
 266     }
 267
 268     case S_PREBUFFERING:    // FROM S_PREBUFFERING
 269     {
 270       switch(newState)
 271       {
 272         case S_PLAY:
 273         {
 274           audio->unPause();
 275           state = newState;
 276           return;
 277         }
 278         case S_STOP:
 279         {
 280           vdr->stopStreaming();
 281           clearStreamChunks();
 282           afeed.stop();
 283           audio->stop();
 284           audio->reset();
 285           state = newState;
 286           return;
 287         }
 288         case S_PREBUFFERING:
 289         {
 290           vdr->stopStreaming();
 291           clearStreamChunks();
 292           afeed.stop();
 293           audio->stop();
 294           audio->reset();
 295           audio->play();
 296           audio->pause();
 297           demuxer->reset();
 298           afeed.start();
 299
 300           state = newState;
 301           preBufferCount = 0;
 302           return;
 303         }
 304         default:
 305         {
 306           logger->log("PlayerRadioLive", Log::EMERG, "Thread called state %u to state %u which is not supported", state, newState);
 307           abort();
 308           break;
 309         }
 310       }
 311     }
 312
 313     case S_PLAY:     // FROM S_PLAY
 314     {
 315       switch(newState)
 316       {
 317         case S_STOP:
 318         {
 319           vdr->stopStreaming();
 320           clearStreamChunks();
 321           afeed.stop();
 322           audio->stop();
 323           audio->reset();
 324           state = newState;
 325           return;
 326         }
 327         case S_PREBUFFERING: // IS THIS HOW IT WORKS?
 328         {
 329           vdr->stopStreaming();
 330           clearStreamChunks();
 331           afeed.stop();
 332           audio->stop();
 333           audio->reset();
 334           audio->play();
 335           audio->pause();
 336           demuxer->reset();
 337           afeed.start();
 338
 339           state = newState;
 340           preBufferCount = 0;
 341           return;
 342         }
 343         default:
 344         {
 345           logger->log("PlayerRadioLive", Log::EMERG, "Thread called state %u to state %u which is not supported", state, newState);
 346           abort();
 347           break;
 348         }
 349       }
 350     }
 351   }
 352 }
 353
 354 bool PlayerRadioLive::checkError()
 355 {
 356   if (!vdr->isConnected())
 357   {
 358     switchState(S_STOP);
 359
 360     Message* m = new Message();
 361     m->from = this;
 362     m->to = messageReceiver;
 363     m->message = Message::PLAYER_EVENT;
 364     m->parameter = PlayerRadioLive::CONNECTION_LOST;
 365     messageQueue->postMessage(m);
 366
 367     return true;
 368   }
 369   return false;
 370 }
 371
 372 void PlayerRadioLive::optimizeInstructionQueue()
 373 {
 374   // Walk the list
 375
 376   // Currently there are only 2 instruction types, so this is a bit overkill...
 377
 378   struct PLInstruction i;
 379   while(instructions.size() > 1)
 380   {
 381     i = instructions.front();
 382     if (i.instruction == I_SETCHANNEL)
 383     {
 384       instructions.pop();  // if this is the first of more than 1 command, currently it cannot possibly be relevant
 385     }
 386     else if (i.instruction == I_STOP)
 387     {
 388       return; // return here and ensure the next instruction will be stop
 389     }
 390   }
 391 }
 392
 393 void PlayerRadioLive::threadMethod()
 394 {
 395   std::unique_lock<std::mutex> ul(playerThreadMutex, std::defer_lock);
 396
 397   while(1)
 398   {
 399     while(!instructions.empty())
 400     {
 401       if (instructions.size() > 1)
 402       {
 403         logger->log("PlayerRadioLive", Log::DEBUG, "Should optimise");
 404         optimizeInstructionQueue();
 405       }
 406
 407       struct PLInstruction i = instructions.front();
 408       instructions.pop();
 409
 410       if (i.instruction == I_SETCHANNEL)
 411       {
 412         logger->log("PlayerRadioLive", Log::DEBUG, "start new stream");
 413
 414         switchState(S_PREBUFFERING);
 415
 416         if (!checkError())
 417         {
 418           Channel* chan = (*chanList)[i.channelIndex];
 419           chan->loadPids();
 420
 421           bool found=false;
 422
 423           if (chan->numAPids > 0)
 424           {
 425             ULONG j = 0;
 426             while (j < chan->numAPids && !found)
 427             {
 428               if (Audio::getInstance()->streamTypeSupported(chan->apids[j].type))
 429               {
 430                 demuxer->setAID(chan->apids[j].pid, 0, chan->apids[j].type, true);
 431                 audio->setStreamType(Audio::MPEG2_PES);
 432                 logger->log("PlayerRadioLive", Log::DEBUG, "Demuxer pids: %u %u %u", chan->vpid, chan->apids[j].pid, chan->apids[j].type);
 433                 found = true;
 434               }
 435               j++;
 436             }
 437           }
 438
 439           if (!found)
 440           {
 441             if (chan->numDPids > 0 && audio->maysupportAc3())
 442             {
 443               ULONG j = 0;
 444               while (j < chan->numDPids && !found)
 445               {
 446                 if (Audio::getInstance()->streamTypeSupported(chan->dpids[j].type))
 447                 {
 448                   demuxer->setAID(chan->dpids[j].pid, 1, chan->dpids[j].type, true);
 449                   audio->setStreamType(Audio::MPEG2_PES);
 450                   logger->log("PlayerRadioLive", Log::DEBUG, "Demuxer pids: %u %u (ac3) %u", chan->vpid, chan->dpids[j].pid, chan->dpids[j].type);
 451                   found=true;
 452                 }
 453                 j++;
 454               }
 455             }
 456             else
 457             {
 458               logger->log("PlayerRadioLive", Log::WARN, "Demuxer no pids!");
 459             }
 460           }
 461
 462           int streamSuccess = vdr->streamChannel(chan->number, this);
 463           if (!checkError() && !streamSuccess)
 464           {
 465             Message* m = new Message();
 466             m->from = this;
 467             m->to = messageReceiver;
 468             m->message = Message::PLAYER_EVENT;
 469             m->parameter = PlayerRadioLive::STREAM_END;
 470             messageQueue->postMessage(m);
 471           }
 472         }
 473       }
 474       else if (i.instruction == I_STOP)
 475       {
 476         logger->log("PlayerRadioLive", Log::DEBUG, "Stopping");
 477         switchState(S_STOP);
 478         checkError();
 479         return;
 480       }
 481     }
 482
 483     while(streamChunks.size())
 484     {
 485       chunkToDemuxer();
 486
 487       if (state == S_PREBUFFERING)
 488       {
 489         ++preBufferCount;
 490         ULONG percentDone = (preBufferCount * 100) / preBufferAmount;
 491
 492         logger->log("PlayerRadioLive", Log::DEBUG, "Prebuffering %lu%%", percentDone);
 493
 494         Message* m = new Message();
 495         m->from = this;
 496         m->to = messageReceiver;
 497         m->message = Message::PLAYER_EVENT;
 498         m->parameter = PlayerRadioLive::PREBUFFERING;
 499         m->tag = percentDone;
 500         messageQueue->postMessage(m);
 501
 502         if (preBufferCount == preBufferAmount)
 503         {
 504           switchState(S_PLAY);
 505           checkError();
 506         }
 507       }
 508     }
 509
 510     ul.lock();
 511     if (!instructions.empty()) { ul.unlock(); continue; }
 512     playerThreadCond.wait(ul);
 513     ul.unlock();
 514   }
 515
 516   logger->log("PlayerRadioLive", Log::DEBUG, "End of thread");
 517 }