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