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 CCECAdapterMessageQueueEntry::CCECAdapterMessageQueueEntry(CCECAdapterMessage *message) :
  43     m_message(message),
  44     m_iPacketsLeft(message->IsTranmission() ? message->Size() / 4 : 1),
  45     m_bSucceeded(false),
  46     m_bWaiting(true) {}
  47
  48 CCECAdapterMessageQueueEntry::~CCECAdapterMessageQueueEntry(void) { }
  49
  50 void CCECAdapterMessageQueueEntry::Broadcast(void)
  51 {
  52   CLockObject lock(m_mutex);
  53   m_condition.Broadcast();
  54 }
  55
  56 bool CCECAdapterMessageQueueEntry::MessageReceived(const CCECAdapterMessage &message)
  57 {
  58   bool bHandled(false);
  59
  60   if (IsResponse(message))
  61   {
  62     switch (message.Message())
  63     {
  64     case MSGCODE_COMMAND_ACCEPTED:
  65       bHandled = MessageReceivedCommandAccepted(message);
  66       break;
  67     case MSGCODE_TRANSMIT_SUCCEEDED:
  68       bHandled = MessageReceivedTransmitSucceeded(message);
  69       break;
  70     default:
  71       bHandled = MessageReceivedResponse(message);
  72       break;
  73     }
  74   }
  75
  76   return bHandled;
  77 }
  78
  79 void CCECAdapterMessageQueueEntry::Signal(void)
  80 {
  81   CLockObject lock(m_mutex);
  82   m_bSucceeded = true;
  83   m_condition.Signal();
  84 }
  85
  86 bool CCECAdapterMessageQueueEntry::Wait(uint32_t iTimeout)
  87 {
  88   bool bReturn(false);
  89   /* wait until we receive a signal when the tranmission succeeded */
  90   {
  91     CLockObject lock(m_mutex);
  92     bReturn = m_bSucceeded ? true : m_condition.Wait(m_mutex, m_bSucceeded, iTimeout);
  93     m_bWaiting = false;
  94   }
  95   return bReturn;
  96 }
  97
  98 bool CCECAdapterMessageQueueEntry::IsWaiting(void)
  99 {
 100   CLockObject lock(m_mutex);
 101   return m_bWaiting;
 102 }
 103
 104 cec_adapter_messagecode CCECAdapterMessageQueueEntry::MessageCode(void)
 105 {
 106   return m_message->Message();
 107 }
 108
 109 bool CCECAdapterMessageQueueEntry::IsResponse(const CCECAdapterMessage &msg)
 110 {
 111   cec_adapter_messagecode msgCode = msg.Message();
 112   return msgCode == MessageCode() ||
 113          msgCode == MSGCODE_TIMEOUT_ERROR ||
 114          msgCode == MSGCODE_COMMAND_ACCEPTED ||
 115          msgCode == MSGCODE_COMMAND_REJECTED ||
 116          (m_message->IsTranmission() && msgCode == MSGCODE_HIGH_ERROR) ||
 117          (m_message->IsTranmission() && msgCode == MSGCODE_LOW_ERROR) ||
 118          (m_message->IsTranmission() && msgCode == MSGCODE_RECEIVE_FAILED) ||
 119          (m_message->IsTranmission() && msgCode == MSGCODE_TRANSMIT_FAILED_LINE) ||
 120          (m_message->IsTranmission() && msgCode == MSGCODE_TRANSMIT_FAILED_ACK) ||
 121          (m_message->IsTranmission() && msgCode == MSGCODE_TRANSMIT_FAILED_TIMEOUT_DATA) ||
 122          (m_message->IsTranmission() && msgCode == MSGCODE_TRANSMIT_FAILED_TIMEOUT_LINE) ||
 123          (m_message->IsTranmission() && msgCode == MSGCODE_TRANSMIT_SUCCEEDED);
 124 }
 125
 126 const char *CCECAdapterMessageQueueEntry::ToString(void) const
 127 {
 128   /* CEC transmissions got the 'set ack polarity' msgcode, which doesn't look nice */
 129   if (m_message->IsTranmission())
 130     return "CEC transmission";
 131   else
 132     return CCECAdapterMessage::ToString(m_message->Message());
 133 }
 134
 135 bool CCECAdapterMessageQueueEntry::MessageReceivedCommandAccepted(const CCECAdapterMessage &message)
 136 {
 137   bool bSendSignal(false);
 138   bool bHandled(false);
 139   {
 140     CLockObject lock(m_mutex);
 141     if (m_iPacketsLeft > 0)
 142     {
 143       /* decrease by 1 */
 144       m_iPacketsLeft--;
 145
 146       /* log this message */
 147       CStdString strLog;
 148       strLog.Format("%s - command accepted", ToString());
 149       if (m_iPacketsLeft > 0)
 150         strLog.AppendFormat(" - waiting for %d more", m_iPacketsLeft);
 151       CLibCEC::AddLog(CEC_LOG_DEBUG, strLog);
 152
 153       /* no more packets left and not a transmission, so we're done */
 154       if (!m_message->IsTranmission() && m_iPacketsLeft == 0)
 155       {
 156         m_message->state = ADAPTER_MESSAGE_STATE_SENT_ACKED;
 157         m_message->response = message.packet;
 158         bSendSignal = true;
 159       }
 160       bHandled = true;
 161     }
 162   }
 163
 164   if (bSendSignal)
 165     Signal();
 166
 167   return bHandled;
 168 }
 169
 170 bool CCECAdapterMessageQueueEntry::MessageReceivedTransmitSucceeded(const CCECAdapterMessage &message)
 171 {
 172   {
 173     CLockObject lock(m_mutex);
 174     if (m_iPacketsLeft == 0)
 175     {
 176       /* transmission succeeded, so we're done */
 177       CLibCEC::AddLog(CEC_LOG_DEBUG, "%s - transmit succeeded", ToString());
 178       m_message->state = ADAPTER_MESSAGE_STATE_SENT_ACKED;
 179       m_message->response = message.packet;
 180     }
 181     else
 182     {
 183       /* error, we expected more acks
 184          since the messages are processed in order, this should not happen, so this is an error situation */
 185       CLibCEC::AddLog(CEC_LOG_WARNING, "%s - received 'transmit succeeded' but not enough 'command accepted' messages (%d left)", ToString(), m_iPacketsLeft);
 186       m_message->state = ADAPTER_MESSAGE_STATE_ERROR;
 187     }
 188   }
 189
 190   Signal();
 191
 192   return true;
 193 }
 194
 195 bool CCECAdapterMessageQueueEntry::MessageReceivedResponse(const CCECAdapterMessage &message)
 196 {
 197   {
 198     CLockObject lock(m_mutex);
 199     CLibCEC::AddLog(CEC_LOG_DEBUG, "%s - received response", ToString());
 200     m_message->response = message.packet;
 201     if (m_message->IsTranmission())
 202       m_message->state = message.Message() == MSGCODE_TRANSMIT_SUCCEEDED ? ADAPTER_MESSAGE_STATE_SENT_ACKED : ADAPTER_MESSAGE_STATE_SENT_NOT_ACKED;
 203     else
 204       m_message->state = ADAPTER_MESSAGE_STATE_SENT_ACKED;
 205   }
 206
 207   Signal();
 208
 209   return true;
 210 }
 211
 212
 213 CCECAdapterMessageQueue::~CCECAdapterMessageQueue(void)
 214 {
 215   Clear();
 216 }
 217
 218 void CCECAdapterMessageQueue::Clear(void)
 219 {
 220   CLockObject lock(m_mutex);
 221   m_messages.clear();
 222 }
 223
 224 void CCECAdapterMessageQueue::MessageReceived(const CCECAdapterMessage &msg)
 225 {
 226   bool bHandled(false);
 227   CLockObject lock(m_mutex);
 228   /* send the received message to each entry in the queue until it is handled */
 229   for (map<uint64_t, CCECAdapterMessageQueueEntry *>::iterator it = m_messages.begin(); !bHandled && it != m_messages.end(); it++)
 230     bHandled = it->second->MessageReceived(msg);
 231
 232   if (!bHandled)
 233   {
 234     /* the message wasn't handled */
 235     bool bIsError(m_com->HandlePoll(msg));
 236     CLibCEC::AddLog(bIsError ? CEC_LOG_WARNING : CEC_LOG_DEBUG, msg.ToString());
 237
 238     /* push this message to the current frame */
 239     if (!bIsError && msg.PushToCecCommand(m_currentCECFrame))
 240     {
 241       /* and push the current frame back over the callback method when a full command was received */
 242       if (m_com->IsInitialised())
 243         m_com->m_callback->OnCommandReceived(m_currentCECFrame);
 244
 245       /* clear the current frame */
 246       m_currentCECFrame.Clear();
 247     }
 248   }
 249 }
 250
 251 void CCECAdapterMessageQueue::AddData(uint8_t *data, size_t iLen)
 252 {
 253   for (size_t iPtr = 0; iPtr < iLen; iPtr++)
 254   {
 255     bool bFullMessage(false);
 256     {
 257       CLockObject lock(m_mutex);
 258       bFullMessage = m_incomingAdapterMessage.PushReceivedByte(data[iPtr]);
 259     }
 260
 261     if (bFullMessage)
 262     {
 263       /* a full message was received */
 264       CCECAdapterMessage newMessage;
 265       newMessage.packet = m_incomingAdapterMessage.packet;
 266       MessageReceived(newMessage);
 267
 268       /* clear the current message */
 269       CLockObject lock(m_mutex);
 270       m_incomingAdapterMessage.Clear();
 271     }
 272   }
 273 }
 274
 275 bool CCECAdapterMessageQueue::Write(CCECAdapterMessage *msg)
 276 {
 277   msg->state = ADAPTER_MESSAGE_STATE_WAITING_TO_BE_SENT;
 278
 279   /* set the correct line timeout */
 280   if (msg->IsTranmission())
 281   {
 282     m_com->SetLineTimeout(msg->lineTimeout);
 283   }
 284
 285   CCECAdapterMessageQueueEntry *entry(NULL);
 286   uint64_t iEntryId(0);
 287   /* add to the wait for ack queue */
 288   if (msg->Message() != MSGCODE_START_BOOTLOADER)
 289   {
 290     CLockObject lock(m_mutex);
 291     entry = new CCECAdapterMessageQueueEntry(msg);
 292     iEntryId = m_iNextMessage++;
 293     m_messages.insert(make_pair(iEntryId, entry));
 294   }
 295
 296   /* TODO write the message async */
 297   if (!m_com->WriteToDevice(msg))
 298   {
 299     /* error! */
 300     Clear();
 301     return false;
 302   }
 303
 304   bool bReturn(true);
 305   if (entry)
 306   {
 307     if (!entry->Wait(msg->transmit_timeout <= 5 ? CEC_DEFAULT_TRANSMIT_WAIT : msg->transmit_timeout))
 308     {
 309       CLibCEC::AddLog(CEC_LOG_DEBUG, "command '%s' was not acked by the controller", CCECAdapterMessage::ToString(msg->Message()));
 310       msg->state = ADAPTER_MESSAGE_STATE_SENT_NOT_ACKED;
 311       bReturn = false;
 312     }
 313
 314     CLockObject lock(m_mutex);
 315     m_messages.erase(iEntryId);
 316   }
 317
 318   return bReturn;
 319 }