src/lib/adapter/USBCECAdapterMessageQueue.cpp

   1 /*
   2  * This file is part of the libCEC(R) library.
   3  *
   4  * libCEC(R) is Copyright (C) 2011-2012 Pulse-Eight Limited.  All rights reserved.
   5  * libCEC(R) is an original work, containing original code.
   6  *
   7  * libCEC(R) is a trademark of Pulse-Eight Limited.
   8  *
   9  * This program is dual-licensed; you can redistribute it and/or modify
  10  * it under the terms of the GNU General Public License as published by
  11  * the Free Software Foundation; either version 2 of the License, or
  12  * (at your option) any later version.
  13  *
  14  * This program is distributed in the hope that it will be useful,
  15  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  16  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  17  * GNU General Public License for more details.
  18  *
  19  * You should have received a copy of the GNU General Public License
  20  * along with this program; if not, write to the Free Software
  21  * Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
  22  *
  23  *
  24  * Alternatively, you can license this library under a commercial license,
  25  * please contact Pulse-Eight Licensing for more information.
  26  *
  27  * For more information contact:
  28  * Pulse-Eight Licensing       <license@pulse-eight.com>
  29  *     http://www.pulse-eight.com/
  30  *     http://www.pulse-eight.net/
  31  */
  32
  33 #include "USBCECAdapterMessageQueue.h"
  34 #include "USBCECAdapterCommunication.h"
  35 #include "../platform/sockets/socket.h"
  36 #include "../LibCEC.h"
  37
  38 using namespace CEC;
  39 using namespace PLATFORM;
  40 using namespace std;
  41
  42 #define MESSAGE_QUEUE_SIGNAL_WAIT_TIME 1000
  43
  44 CCECAdapterMessageQueueEntry::CCECAdapterMessageQueueEntry(CCECAdapterMessage *message) :
  45     m_message(message),
  46     m_iPacketsLeft(message->IsTranmission() ? message->Size() / 4 : 1),
  47     m_bSucceeded(false),
  48     m_bWaiting(true) {}
  49
  50 CCECAdapterMessageQueueEntry::~CCECAdapterMessageQueueEntry(void) { }
  51
  52 void CCECAdapterMessageQueueEntry::Broadcast(void)
  53 {
  54   CLockObject lock(m_mutex);
  55   m_condition.Broadcast();
  56 }
  57
  58 bool CCECAdapterMessageQueueEntry::MessageReceived(const CCECAdapterMessage &message)
  59 {
  60   bool bHandled(false);
  61
  62   if (IsResponse(message))
  63   {
  64     switch (message.Message())
  65     {
  66     case MSGCODE_COMMAND_ACCEPTED:
  67       bHandled = MessageReceivedCommandAccepted(message);
  68       break;
  69     case MSGCODE_TRANSMIT_SUCCEEDED:
  70       bHandled = MessageReceivedTransmitSucceeded(message);
  71       break;
  72     default:
  73       bHandled = MessageReceivedResponse(message);
  74       break;
  75     }
  76   }
  77
  78   return bHandled;
  79 }
  80
  81 void CCECAdapterMessageQueueEntry::Signal(void)
  82 {
  83   CLockObject lock(m_mutex);
  84   m_bSucceeded = true;
  85   m_condition.Signal();
  86 }
  87
  88 bool CCECAdapterMessageQueueEntry::Wait(uint32_t iTimeout)
  89 {
  90   bool bReturn(false);
  91   /* wait until we receive a signal when the tranmission succeeded */
  92   {
  93     CLockObject lock(m_mutex);
  94     bReturn = m_bSucceeded ? true : m_condition.Wait(m_mutex, m_bSucceeded, iTimeout);
  95     m_bWaiting = false;
  96   }
  97   return bReturn;
  98 }
  99
 100 bool CCECAdapterMessageQueueEntry::IsWaiting(void)
 101 {
 102   CLockObject lock(m_mutex);
 103   return m_bWaiting;
 104 }
 105
 106 cec_adapter_messagecode CCECAdapterMessageQueueEntry::MessageCode(void)
 107 {
 108   return m_message->Message();
 109 }
 110
 111 bool CCECAdapterMessageQueueEntry::IsResponse(const CCECAdapterMessage &msg)
 112 {
 113   cec_adapter_messagecode msgCode = msg.Message();
 114   return msgCode == MessageCode() ||
 115          msgCode == MSGCODE_TIMEOUT_ERROR ||
 116          msgCode == MSGCODE_COMMAND_ACCEPTED ||
 117          msgCode == MSGCODE_COMMAND_REJECTED ||
 118          (m_message->IsTranmission() && msgCode == MSGCODE_HIGH_ERROR) ||
 119          (m_message->IsTranmission() && msgCode == MSGCODE_LOW_ERROR) ||
 120          (m_message->IsTranmission() && msgCode == MSGCODE_RECEIVE_FAILED) ||
 121          (m_message->IsTranmission() && msgCode == MSGCODE_TRANSMIT_FAILED_LINE) ||
 122          (m_message->IsTranmission() && msgCode == MSGCODE_TRANSMIT_FAILED_ACK) ||
 123          (m_message->IsTranmission() && msgCode == MSGCODE_TRANSMIT_FAILED_TIMEOUT_DATA) ||
 124          (m_message->IsTranmission() && msgCode == MSGCODE_TRANSMIT_FAILED_TIMEOUT_LINE) ||
 125          (m_message->IsTranmission() && msgCode == MSGCODE_TRANSMIT_SUCCEEDED);
 126 }
 127
 128 const char *CCECAdapterMessageQueueEntry::ToString(void) const
 129 {
 130   /* CEC transmissions got the 'set ack polarity' msgcode, which doesn't look nice */
 131   if (m_message->IsTranmission())
 132     return "CEC transmission";
 133   else
 134     return CCECAdapterMessage::ToString(m_message->Message());
 135 }
 136
 137 bool CCECAdapterMessageQueueEntry::MessageReceivedCommandAccepted(const CCECAdapterMessage &message)
 138 {
 139   bool bSendSignal(false);
 140   bool bHandled(false);
 141   {
 142     CLockObject lock(m_mutex);
 143     if (m_iPacketsLeft > 0)
 144     {
 145       /* decrease by 1 */
 146       m_iPacketsLeft--;
 147
 148       /* log this message */
 149       CStdString strLog;
 150       strLog.Format("%s - command accepted", ToString());
 151       if (m_iPacketsLeft > 0)
 152         strLog.AppendFormat(" - waiting for %d more", m_iPacketsLeft);
 153       CLibCEC::AddLog(CEC_LOG_DEBUG, strLog);
 154
 155       /* no more packets left and not a transmission, so we're done */
 156       if (!m_message->IsTranmission() && m_iPacketsLeft == 0)
 157       {
 158         m_message->state = ADAPTER_MESSAGE_STATE_SENT_ACKED;
 159         m_message->response = message.packet;
 160         bSendSignal = true;
 161       }
 162       bHandled = true;
 163     }
 164   }
 165
 166   if (bSendSignal)
 167     Signal();
 168
 169   return bHandled;
 170 }
 171
 172 bool CCECAdapterMessageQueueEntry::MessageReceivedTransmitSucceeded(const CCECAdapterMessage &message)
 173 {
 174   {
 175     CLockObject lock(m_mutex);
 176     if (m_iPacketsLeft == 0)
 177     {
 178       /* transmission succeeded, so we're done */
 179       CLibCEC::AddLog(CEC_LOG_DEBUG, "%s - transmit succeeded", ToString());
 180       m_message->state = ADAPTER_MESSAGE_STATE_SENT_ACKED;
 181       m_message->response = message.packet;
 182     }
 183     else
 184     {
 185       /* error, we expected more acks
 186          since the messages are processed in order, this should not happen, so this is an error situation */
 187       CLibCEC::AddLog(CEC_LOG_WARNING, "%s - received 'transmit succeeded' but not enough 'command accepted' messages (%d left)", ToString(), m_iPacketsLeft);
 188       m_message->state = ADAPTER_MESSAGE_STATE_ERROR;
 189     }
 190   }
 191
 192   Signal();
 193
 194   return true;
 195 }
 196
 197 bool CCECAdapterMessageQueueEntry::MessageReceivedResponse(const CCECAdapterMessage &message)
 198 {
 199   {
 200     CLockObject lock(m_mutex);
 201     CLibCEC::AddLog(CEC_LOG_DEBUG, "%s - received response - %s", ToString(), message.ToString().c_str());
 202     m_message->response = message.packet;
 203     if (m_message->IsTranmission())
 204       m_message->state = message.Message() == MSGCODE_TRANSMIT_SUCCEEDED ? ADAPTER_MESSAGE_STATE_SENT_ACKED : ADAPTER_MESSAGE_STATE_SENT_NOT_ACKED;
 205     else
 206       m_message->state = ADAPTER_MESSAGE_STATE_SENT_ACKED;
 207   }
 208
 209   Signal();
 210
 211   return true;
 212 }
 213
 214
 215 CCECAdapterMessageQueue::~CCECAdapterMessageQueue(void)
 216 {
 217   Clear();
 218   StopThread(0);
 219 }
 220
 221 void CCECAdapterMessageQueue::Clear(void)
 222 {
 223   StopThread(5);
 224   CLockObject lock(m_mutex);
 225   m_writeQueue.Clear();
 226   m_messages.clear();
 227 }
 228
 229 void *CCECAdapterMessageQueue::Process(void)
 230 {
 231   CCECAdapterMessageQueueEntry *message(NULL);
 232   while (!IsStopped())
 233   {
 234     /* wait for a new message */
 235     if (m_writeQueue.Pop(message, MESSAGE_QUEUE_SIGNAL_WAIT_TIME))
 236     {
 237       /* write this message */
 238       m_com->WriteToDevice(message->m_message);
 239       if (message->m_message->state == ADAPTER_MESSAGE_STATE_ERROR)
 240       {
 241         message->Signal();
 242         Clear();
 243         break;
 244       }
 245     }
 246   }
 247   return NULL;
 248 }
 249
 250 void CCECAdapterMessageQueue::MessageReceived(const CCECAdapterMessage &msg)
 251 {
 252   bool bHandled(false);
 253   CLockObject lock(m_mutex);
 254   /* send the received message to each entry in the queue until it is handled */
 255   for (map<uint64_t, CCECAdapterMessageQueueEntry *>::iterator it = m_messages.begin(); !bHandled && it != m_messages.end(); it++)
 256     bHandled = it->second->MessageReceived(msg);
 257
 258   if (!bHandled)
 259   {
 260     /* the message wasn't handled */
 261     bool bIsError(m_com->HandlePoll(msg));
 262     CLibCEC::AddLog(bIsError ? CEC_LOG_WARNING : CEC_LOG_DEBUG, msg.ToString());
 263
 264     /* push this message to the current frame */
 265     if (!bIsError && msg.PushToCecCommand(m_currentCECFrame))
 266     {
 267       /* and push the current frame back over the callback method when a full command was received */
 268       if (m_com->IsInitialised())
 269         m_com->m_callback->OnCommandReceived(m_currentCECFrame);
 270
 271       /* clear the current frame */
 272       m_currentCECFrame.Clear();
 273     }
 274   }
 275 }
 276
 277 void CCECAdapterMessageQueue::AddData(uint8_t *data, size_t iLen)
 278 {
 279   for (size_t iPtr = 0; iPtr < iLen; iPtr++)
 280   {
 281     bool bFullMessage(false);
 282     {
 283       CLockObject lock(m_mutex);
 284       bFullMessage = m_incomingAdapterMessage.PushReceivedByte(data[iPtr]);
 285     }
 286
 287     if (bFullMessage)
 288     {
 289       /* a full message was received */
 290       CCECAdapterMessage newMessage;
 291       newMessage.packet = m_incomingAdapterMessage.packet;
 292       MessageReceived(newMessage);
 293
 294       /* clear the current message */
 295       CLockObject lock(m_mutex);
 296       m_incomingAdapterMessage.Clear();
 297     }
 298   }
 299 }
 300
 301 bool CCECAdapterMessageQueue::Write(CCECAdapterMessage *msg)
 302 {
 303   msg->state = ADAPTER_MESSAGE_STATE_WAITING_TO_BE_SENT;
 304
 305   /* set the correct line timeout */
 306   if (msg->IsTranmission())
 307   {
 308     m_com->SetLineTimeout(msg->lineTimeout);
 309   }
 310
 311   CCECAdapterMessageQueueEntry *entry(NULL);
 312   uint64_t iEntryId(0);
 313   /* add to the wait for ack queue */
 314   if (msg->Message() != MSGCODE_START_BOOTLOADER)
 315   {
 316     CLockObject lock(m_mutex);
 317     entry = new CCECAdapterMessageQueueEntry(msg);
 318     iEntryId = m_iNextMessage++;
 319     m_messages.insert(make_pair(iEntryId, entry));
 320   }
 321
 322   /* add the message to the write queue */
 323   m_writeQueue.Push(entry);
 324
 325   bool bReturn(true);
 326   if (entry)
 327   {
 328     if (!entry->Wait(msg->transmit_timeout <= 5 ? CEC_DEFAULT_TRANSMIT_WAIT : msg->transmit_timeout))
 329     {
 330       CLibCEC::AddLog(CEC_LOG_DEBUG, "command '%s' was not acked by the controller", CCECAdapterMessage::ToString(msg->Message()));
 331       msg->state = ADAPTER_MESSAGE_STATE_SENT_NOT_ACKED;
 332       bReturn = false;
 333     }
 334
 335     CLockObject lock(m_mutex);
 336     m_messages.erase(iEntryId);
 337     delete entry;
 338   }
 339
 340   return bReturn;
 341 }