[deb_libcec.git] / src / lib / adapter / Pulse-Eight / USBCECAdapterMessageQueue.cpp

/*
 * This file is part of the libCEC(R) library.
 *
 * libCEC(R) is Copyright (C) 2011-2013 Pulse-Eight Limited.  All rights reserved.
 * libCEC(R) is an original work, containing original code.
 *
 * libCEC(R) is a trademark of Pulse-Eight Limited.
 *
 * This program is dual-licensed; you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation; either version 2 of the License, or
 * (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program; if not, write to the Free Software
 * Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
 *
 *
 * Alternatively, you can license this library under a commercial license,
 * please contact Pulse-Eight Licensing for more information.
 *
 * For more information contact:
 * Pulse-Eight Licensing       <license@pulse-eight.com>
 *     http://www.pulse-eight.com/
 *     http://www.pulse-eight.net/
 */

#include "env.h"
#include "USBCECAdapterMessageQueue.h"

#include "USBCECAdapterCommunication.h"
#include "USBCECAdapterMessage.h"
#include "lib/platform/sockets/socket.h"
#include "lib/LibCEC.h"
#include "lib/platform/util/StdString.h"

using namespace CEC;
using namespace PLATFORM;
using namespace std;

#define MESSAGE_QUEUE_SIGNAL_WAIT_TIME 1000

CCECAdapterMessageQueueEntry::CCECAdapterMessageQueueEntry(CCECAdapterMessageQueue *queue, CCECAdapterMessage *message) :
    m_queue(queue),
    m_message(message),
    m_iPacketsLeft(message->IsTranmission() ? message->Size() / 4 : 1),
    m_bSucceeded(false),
    m_bWaiting(true),
    m_queueTimeout(message->transmit_timeout) {}

CCECAdapterMessageQueueEntry::~CCECAdapterMessageQueueEntry(void) { }

void CCECAdapterMessageQueueEntry::Broadcast(void)
{
  CLockObject lock(m_mutex);
  m_condition.Broadcast();
}

bool CCECAdapterMessageQueueEntry::MessageReceived(const CCECAdapterMessage &message)
{
  bool bHandled(false);

  if (IsResponse(message))
  {
    switch (message.Message())
    {
    case MSGCODE_COMMAND_ACCEPTED:
      bHandled = MessageReceivedCommandAccepted(message);
      break;
    case MSGCODE_TRANSMIT_SUCCEEDED:
      bHandled = MessageReceivedTransmitSucceeded(message);
      break;
    default:
      bHandled = MessageReceivedResponse(message);
      break;
    }
  }

  return bHandled;
}

void CCECAdapterMessageQueueEntry::Signal(void)
{
  CLockObject lock(m_mutex);
  m_bSucceeded = true;
  m_condition.Signal();
}

bool CCECAdapterMessageQueueEntry::Wait(uint32_t iTimeout)
{
  bool bReturn(false);
  /* wait until we receive a signal when the tranmission succeeded */
  {
    CLockObject lock(m_mutex);
    bReturn = m_bSucceeded ? true : m_condition.Wait(m_mutex, m_bSucceeded, iTimeout);
    m_bWaiting = false;
  }
  return bReturn;
}

bool CCECAdapterMessageQueueEntry::IsWaiting(void)
{
  CLockObject lock(m_mutex);
  return m_bWaiting;
}

cec_adapter_messagecode CCECAdapterMessageQueueEntry::MessageCode(void)
{
  return m_message->Message();
}

bool CCECAdapterMessageQueueEntry::IsResponseOld(const CCECAdapterMessage &msg)
{
  cec_adapter_messagecode msgCode = msg.Message();

  return msgCode == MessageCode() ||
         msgCode == MSGCODE_COMMAND_ACCEPTED ||
         msgCode == MSGCODE_COMMAND_REJECTED ||
         (m_message->IsTranmission() && (msgCode == MSGCODE_TIMEOUT_ERROR ||
             msgCode == MSGCODE_HIGH_ERROR ||
             msgCode == MSGCODE_LOW_ERROR ||
             msgCode == MSGCODE_RECEIVE_FAILED ||
             msgCode == MSGCODE_TRANSMIT_FAILED_LINE ||
             msgCode == MSGCODE_TRANSMIT_FAILED_ACK ||
             msgCode == MSGCODE_TRANSMIT_FAILED_TIMEOUT_DATA ||
             msgCode == MSGCODE_TRANSMIT_FAILED_TIMEOUT_LINE ||
             msgCode == MSGCODE_TRANSMIT_SUCCEEDED));
}

bool CCECAdapterMessageQueueEntry::IsResponse(const CCECAdapterMessage &msg)
{
  if (m_message->state == ADAPTER_MESSAGE_STATE_SENT_ACKED)
    return false;

  cec_adapter_messagecode thisMsgCode = m_message->Message();
  cec_adapter_messagecode msgCode = msg.Message();
  cec_adapter_messagecode msgResponse = msg.ResponseTo();

  // msgcode matches, always a response
  if (msgCode == MessageCode())
    return true;

  if (!ProvidesExtendedResponse())
    return IsResponseOld(msg);

  // response without a msgcode
  if (msgResponse == MSGCODE_NOTHING)
    return false;

  // commands that only repond with accepted/rejected
  if (thisMsgCode == MSGCODE_PING ||
      thisMsgCode == MSGCODE_SET_ACK_MASK ||
      thisMsgCode == MSGCODE_SET_CONTROLLED ||
      thisMsgCode == MSGCODE_SET_AUTO_ENABLED ||
      thisMsgCode == MSGCODE_SET_DEFAULT_LOGICAL_ADDRESS ||
      thisMsgCode == MSGCODE_SET_LOGICAL_ADDRESS_MASK ||
      thisMsgCode == MSGCODE_SET_PHYSICAL_ADDRESS ||
      thisMsgCode == MSGCODE_SET_DEVICE_TYPE ||
      thisMsgCode == MSGCODE_SET_HDMI_VERSION ||
      thisMsgCode == MSGCODE_SET_OSD_NAME ||
      thisMsgCode == MSGCODE_WRITE_EEPROM ||
      thisMsgCode == MSGCODE_TRANSMIT_IDLETIME ||
      thisMsgCode == MSGCODE_SET_ACTIVE_SOURCE)
    return thisMsgCode == msgResponse;

  if (!m_message->IsTranmission())
    return false;

  return ((msgCode == MSGCODE_COMMAND_ACCEPTED || msgCode == MSGCODE_COMMAND_REJECTED) &&
      (msgResponse == MSGCODE_TRANSMIT_ACK_POLARITY || msgResponse == MSGCODE_TRANSMIT || msgResponse == MSGCODE_TRANSMIT_EOM)) ||
      msgCode == MSGCODE_TIMEOUT_ERROR ||
      msgCode == MSGCODE_RECEIVE_FAILED ||
      msgCode == MSGCODE_TRANSMIT_FAILED_ACK ||
      msgCode == MSGCODE_TRANSMIT_FAILED_TIMEOUT_DATA ||
      msgCode == MSGCODE_TRANSMIT_FAILED_TIMEOUT_LINE ||
      msgCode == MSGCODE_TRANSMIT_SUCCEEDED;
}

const char *CCECAdapterMessageQueueEntry::ToString(void) const
{
  /* CEC transmissions got the 'set ack polarity' msgcode, which doesn't look nice */
  if (m_message->IsTranmission())
    return "CEC transmission";
  else
    return CCECAdapterMessage::ToString(m_message->Message());
}

bool CCECAdapterMessageQueueEntry::MessageReceivedCommandAccepted(const CCECAdapterMessage &message)
{
  bool bSendSignal(false);
  bool bHandled(false);
  {
    CLockObject lock(m_mutex);
    if (m_iPacketsLeft > 0)
    {
      /* decrease by 1 */
      m_iPacketsLeft--;

#ifdef CEC_DEBUGGING
      /* log this message */
      CStdString strLog;
      strLog.Format("%s - command accepted", ToString());
      if (m_iPacketsLeft > 0)
        strLog.AppendFormat(" - waiting for %d more", m_iPacketsLeft);
      m_queue->m_com->m_callback->GetLib()->AddLog(CEC_LOG_DEBUG, strLog);
#endif

      /* no more packets left and not a transmission, so we're done */
      if (!m_message->IsTranmission() && m_iPacketsLeft == 0)
      {
        m_message->state = ADAPTER_MESSAGE_STATE_SENT_ACKED;
        m_message->response = message.packet;
        bSendSignal = true;
      }
      bHandled = true;
    }
  }

  if (bSendSignal)
    Signal();

  return bHandled;
}

bool CCECAdapterMessageQueueEntry::MessageReceivedTransmitSucceeded(const CCECAdapterMessage &message)
{
  {
    CLockObject lock(m_mutex);
    if (m_iPacketsLeft == 0)
    {
      /* transmission succeeded, so we're done */
#ifdef CEC_DEBUGGING
      m_queue->m_com->m_callback->GetLib()->AddLog(CEC_LOG_DEBUG, "%s - transmit succeeded", m_message->ToString().c_str());
#endif
      m_message->state = ADAPTER_MESSAGE_STATE_SENT_ACKED;
      m_message->response = message.packet;
    }
    else
    {
      /* error, we expected more acks
         since the messages are processed in order, this should not happen, so this is an error situation */
      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);
      m_message->state = ADAPTER_MESSAGE_STATE_ERROR;
    }
  }

  Signal();

  return true;
}

bool CCECAdapterMessageQueueEntry::MessageReceivedResponse(const CCECAdapterMessage &message)
{
  {
    CLockObject lock(m_mutex);
#ifdef CEC_DEBUGGING
    m_queue->m_com->m_callback->GetLib()->AddLog(CEC_LOG_DEBUG, "%s - received response - %s", ToString(), message.ToString().c_str());
#else
    if (message.IsError())
      m_queue->m_com->m_callback->GetLib()->AddLog(CEC_LOG_DEBUG, "%s - received response - %s", ToString(), message.ToString().c_str());
#endif
    m_message->response = message.packet;
    if (m_message->IsTranmission())
      m_message->state = message.Message() == MSGCODE_TRANSMIT_SUCCEEDED ? ADAPTER_MESSAGE_STATE_SENT_ACKED : ADAPTER_MESSAGE_STATE_SENT_NOT_ACKED;
    else
      m_message->state = ADAPTER_MESSAGE_STATE_SENT_ACKED;
  }

  Signal();

  return true;
}

bool CCECAdapterMessageQueueEntry::ProvidesExtendedResponse(void)
{
  return m_queue && m_queue->ProvidesExtendedResponse();
}

bool CCECAdapterMessageQueueEntry::TimedOutOrSucceeded(void) const
{
  return m_message->bFireAndForget && (m_bSucceeded || m_queueTimeout.TimeLeft() == 0);
}

CCECAdapterMessageQueue::CCECAdapterMessageQueue(CUSBCECAdapterCommunication *com) :
  PLATFORM::CThread(),
  m_com(com),
  m_iNextMessage(0)
{
  m_incomingAdapterMessage = new CCECAdapterMessage;
  m_currentCECFrame.Clear();
}

CCECAdapterMessageQueue::~CCECAdapterMessageQueue(void)
{
  StopThread(-1);
  Clear();
  StopThread();
  delete m_incomingAdapterMessage;
}

void CCECAdapterMessageQueue::Clear(void)
{
  StopThread(5);
  CLockObject lock(m_mutex);
  m_writeQueue.Clear();
  m_messages.clear();
}

void *CCECAdapterMessageQueue::Process(void)
{
  CCECAdapterMessageQueueEntry *message(NULL);
  while (!IsStopped())
  {
    /* wait for a new message */
    if (m_writeQueue.Pop(message, MESSAGE_QUEUE_SIGNAL_WAIT_TIME) && message)
    {
      /* write this message */
      {
        CLockObject lock(m_mutex);
        m_com->WriteToDevice(message->m_message);
      }
      if (message->m_message->state == ADAPTER_MESSAGE_STATE_ERROR ||
          message->m_message->Message() == MSGCODE_START_BOOTLOADER)
      {
        message->Signal();
        Clear();
        break;
      }
    }

    CheckTimedOutMessages();
  }
  return NULL;
}

void CCECAdapterMessageQueue::CheckTimedOutMessages(void)
{
  CLockObject lock(m_mutex);
  vector<uint64_t> timedOut;
  for (map<uint64_t, CCECAdapterMessageQueueEntry *>::iterator it = m_messages.begin(); it != m_messages.end(); it++)
  {
    if (it->second->TimedOutOrSucceeded())
    {
      timedOut.push_back(it->first);
      if (!it->second->m_bSucceeded)
        m_com->m_callback->GetLib()->AddLog(CEC_LOG_DEBUG, "command '%s' was not acked by the controller", CCECAdapterMessage::ToString(it->second->m_message->Message()));
      delete it->second->m_message;
      delete it->second;
    }
  }

  for (vector<uint64_t>::iterator it = timedOut.begin(); it != timedOut.end(); it++)
  {
    uint64_t iEntryId = *it;
    m_messages.erase(iEntryId);
  }
}

void CCECAdapterMessageQueue::MessageReceived(const CCECAdapterMessage &msg)
{
  bool bHandled(false);
  CLockObject lock(m_mutex);
  /* send the received message to each entry in the queue until it is handled */
  for (map<uint64_t, CCECAdapterMessageQueueEntry *>::iterator it = m_messages.begin(); !bHandled && it != m_messages.end(); it++)
    bHandled = it->second->MessageReceived(msg);

  if (!bHandled)
  {
    /* the message wasn't handled */
    bool bIsError(m_com->HandlePoll(msg));
#ifdef CEC_DEBUGGING
    m_com->m_callback->GetLib()->AddLog(bIsError ? CEC_LOG_WARNING : CEC_LOG_DEBUG, msg.ToString().c_str());
#else
    if (bIsError)
      m_com->m_callback->GetLib()->AddLog(CEC_LOG_WARNING, msg.ToString().c_str());
#endif

    /* push this message to the current frame */
    if (!bIsError && msg.PushToCecCommand(m_currentCECFrame))
    {
      /* and push the current frame back over the callback method when a full command was received */
      if (m_com->IsInitialised())
        m_com->m_callback->OnCommandReceived(m_currentCECFrame);

      /* clear the current frame */
      m_currentCECFrame.Clear();
    }
  }
}

void CCECAdapterMessageQueue::AddData(uint8_t *data, size_t iLen)
{
  for (size_t iPtr = 0; iPtr < iLen; iPtr++)
  {
    bool bFullMessage(false);
    {
      CLockObject lock(m_mutex);
      bFullMessage = m_incomingAdapterMessage->PushReceivedByte(data[iPtr]);
    }

    if (bFullMessage)
    {
      /* a full message was received */
      CCECAdapterMessage newMessage;
      newMessage.packet = m_incomingAdapterMessage->packet;
      MessageReceived(newMessage);

      /* clear the current message */
      CLockObject lock(m_mutex);
      m_incomingAdapterMessage->Clear();
    }
  }
}

bool CCECAdapterMessageQueue::Write(CCECAdapterMessage *msg)
{
  msg->state = ADAPTER_MESSAGE_STATE_WAITING_TO_BE_SENT;

  /* set the correct line timeout */
  if (msg->IsTranmission())
  {
    m_com->SetLineTimeout(msg->lineTimeout);
  }

  CCECAdapterMessageQueueEntry *entry = new CCECAdapterMessageQueueEntry(this, msg);
  if (!entry)
  {
    m_com->m_callback->GetLib()->AddLog(CEC_LOG_ERROR, "couldn't create queue entry for '%s'", CCECAdapterMessage::ToString(msg->Message()));
    msg->state = ADAPTER_MESSAGE_STATE_ERROR;
    return false;
  }

  uint64_t iEntryId(0);
  /* add to the wait for ack queue */
  if (msg->Message() != MSGCODE_START_BOOTLOADER)
  {
    CLockObject lock(m_mutex);
    iEntryId = m_iNextMessage++;
    m_messages.insert(make_pair(iEntryId, entry));
  }

  /* add the message to the write queue */
  m_writeQueue.Push(entry);

  bool bReturn(true);
  if (!msg->bFireAndForget)
  {
    if (!entry->Wait(msg->transmit_timeout <= 5 ? CEC_DEFAULT_TRANSMIT_WAIT : msg->transmit_timeout))
    {
      m_com->m_callback->GetLib()->AddLog(CEC_LOG_DEBUG, "command '%s' was not acked by the controller", CCECAdapterMessage::ToString(msg->Message()));
      msg->state = ADAPTER_MESSAGE_STATE_SENT_NOT_ACKED;
      bReturn = false;
    }

    if (msg->Message() != MSGCODE_START_BOOTLOADER)
    {
      CLockObject lock(m_mutex);
      m_messages.erase(iEntryId);
    }

    if (msg->ReplyIsError() && msg->state != ADAPTER_MESSAGE_STATE_SENT_NOT_ACKED)
      msg->state = ADAPTER_MESSAGE_STATE_ERROR;

    delete entry;
  }

  return bReturn;
}

bool CCECAdapterMessageQueue::ProvidesExtendedResponse(void)
{
  return m_com && m_com->ProvidesExtendedResponse();
}
Commit	Line	Data
cbbe90dd JB	1	/*
	2	* This file is part of the libCEC(R) library.
	3	*
	4	* libCEC(R) is Copyright (C) 2011-2013 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 "env.h"
	34	#include "USBCECAdapterMessageQueue.h"
	35
	36	#include "USBCECAdapterCommunication.h"
	37	#include "USBCECAdapterMessage.h"
	38	#include "lib/platform/sockets/socket.h"
	39	#include "lib/LibCEC.h"
	40	#include "lib/platform/util/StdString.h"
	41
	42	using namespace CEC;
	43	using namespace PLATFORM;
	44	using namespace std;
	45
	46	#define MESSAGE_QUEUE_SIGNAL_WAIT_TIME 1000
	47
	48	CCECAdapterMessageQueueEntry::CCECAdapterMessageQueueEntry(CCECAdapterMessageQueue queue, CCECAdapterMessage message) :
	49	m_queue(queue),
	50	m_message(message),
	51	m_iPacketsLeft(message->IsTranmission() ? message->Size() / 4 : 1),
	52	m_bSucceeded(false),
	53	m_bWaiting(true),
	54	m_queueTimeout(message->transmit_timeout) {}
	55
	56	CCECAdapterMessageQueueEntry::~CCECAdapterMessageQueueEntry(void) { }
	57
	58	void CCECAdapterMessageQueueEntry::Broadcast(void)
	59	{
	60	CLockObject lock(m_mutex);
	61	m_condition.Broadcast();
	62	}
	63
	64	bool CCECAdapterMessageQueueEntry::MessageReceived(const CCECAdapterMessage &message)
65	{
66	bool bHandled(false);
67
68	if (IsResponse(message))
69	{
70	switch (message.Message())
71	{
72	case MSGCODE_COMMAND_ACCEPTED:
73	bHandled = MessageReceivedCommandAccepted(message);
74	break;
75	case MSGCODE_TRANSMIT_SUCCEEDED:
76	bHandled = MessageReceivedTransmitSucceeded(message);
77	break;
78	default:
79	bHandled = MessageReceivedResponse(message);
80	break;
81	}
82	}
83
84	return bHandled;
85	}
86
87	void CCECAdapterMessageQueueEntry::Signal(void)
88	{
89	CLockObject lock(m_mutex);
90	m_bSucceeded = true;
91	m_condition.Signal();
92	}
93
94	bool CCECAdapterMessageQueueEntry::Wait(uint32_t iTimeout)
95	{
96	bool bReturn(false);
97	/* wait until we receive a signal when the tranmission succeeded */
98	{
99	CLockObject lock(m_mutex);
100	bReturn = m_bSucceeded ? true : m_condition.Wait(m_mutex, m_bSucceeded, iTimeout);
101	m_bWaiting = false;
102	}
103	return bReturn;
104	}
105
106	bool CCECAdapterMessageQueueEntry::IsWaiting(void)
107	{
108	CLockObject lock(m_mutex);
109	return m_bWaiting;
110	}
111
112	cec_adapter_messagecode CCECAdapterMessageQueueEntry::MessageCode(void)
113	{
114	return m_message->Message();
115	}
116
117	bool CCECAdapterMessageQueueEntry::IsResponseOld(const CCECAdapterMessage &msg)
118	{
119	cec_adapter_messagecode msgCode = msg.Message();
120
121	return msgCode == MessageCode() \|\|
122	msgCode == MSGCODE_COMMAND_ACCEPTED \|\|
123	msgCode == MSGCODE_COMMAND_REJECTED \|\|
124	(m_message->IsTranmission() && (msgCode == MSGCODE_TIMEOUT_ERROR \|\|
125	msgCode == MSGCODE_HIGH_ERROR \|\|
126	msgCode == MSGCODE_LOW_ERROR \|\|
127	msgCode == MSGCODE_RECEIVE_FAILED \|\|
128	msgCode == MSGCODE_TRANSMIT_FAILED_LINE \|\|
129	msgCode == MSGCODE_TRANSMIT_FAILED_ACK \|\|
130	msgCode == MSGCODE_TRANSMIT_FAILED_TIMEOUT_DATA \|\|
131	msgCode == MSGCODE_TRANSMIT_FAILED_TIMEOUT_LINE \|\|
132	msgCode == MSGCODE_TRANSMIT_SUCCEEDED));
133	}
134
135	bool CCECAdapterMessageQueueEntry::IsResponse(const CCECAdapterMessage &msg)
136	{
137	if (m_message->state == ADAPTER_MESSAGE_STATE_SENT_ACKED)
138	return false;
139
140	cec_adapter_messagecode thisMsgCode = m_message->Message();
141	cec_adapter_messagecode msgCode = msg.Message();
142	cec_adapter_messagecode msgResponse = msg.ResponseTo();
143
144	// msgcode matches, always a response
145	if (msgCode == MessageCode())
146	return true;
147
148	if (!ProvidesExtendedResponse())
149	return IsResponseOld(msg);
150
151	// response without a msgcode
152	if (msgResponse == MSGCODE_NOTHING)
153	return false;
154
155	// commands that only repond with accepted/rejected
156	if (thisMsgCode == MSGCODE_PING \|\|
157	thisMsgCode == MSGCODE_SET_ACK_MASK \|\|
158	thisMsgCode == MSGCODE_SET_CONTROLLED \|\|
159	thisMsgCode == MSGCODE_SET_AUTO_ENABLED \|\|
160	thisMsgCode == MSGCODE_SET_DEFAULT_LOGICAL_ADDRESS \|\|
161	thisMsgCode == MSGCODE_SET_LOGICAL_ADDRESS_MASK \|\|
162	thisMsgCode == MSGCODE_SET_PHYSICAL_ADDRESS \|\|
163	thisMsgCode == MSGCODE_SET_DEVICE_TYPE \|\|
164	thisMsgCode == MSGCODE_SET_HDMI_VERSION \|\|
165	thisMsgCode == MSGCODE_SET_OSD_NAME \|\|
166	thisMsgCode == MSGCODE_WRITE_EEPROM \|\|
167	thisMsgCode == MSGCODE_TRANSMIT_IDLETIME \|\|
168	thisMsgCode == MSGCODE_SET_ACTIVE_SOURCE)
169	return thisMsgCode == msgResponse;
170
171	if (!m_message->IsTranmission())
172	return false;
173
174	return ((msgCode == MSGCODE_COMMAND_ACCEPTED \|\| msgCode == MSGCODE_COMMAND_REJECTED) &&
175	(msgResponse == MSGCODE_TRANSMIT_ACK_POLARITY \|\| msgResponse == MSGCODE_TRANSMIT \|\| msgResponse == MSGCODE_TRANSMIT_EOM)) \|\|
176	msgCode == MSGCODE_TIMEOUT_ERROR \|\|
177	msgCode == MSGCODE_RECEIVE_FAILED \|\|
178	msgCode == MSGCODE_TRANSMIT_FAILED_ACK \|\|
179	msgCode == MSGCODE_TRANSMIT_FAILED_TIMEOUT_DATA \|\|
180	msgCode == MSGCODE_TRANSMIT_FAILED_TIMEOUT_LINE \|\|
181	msgCode == MSGCODE_TRANSMIT_SUCCEEDED;
182	}
183
184	const char *CCECAdapterMessageQueueEntry::ToString(void) const
185	{
186	/* CEC transmissions got the 'set ack polarity' msgcode, which doesn't look nice */
187	if (m_message->IsTranmission())
188	return "CEC transmission";
189	else
190	return CCECAdapterMessage::ToString(m_message->Message());
191	}
192
193	bool CCECAdapterMessageQueueEntry::MessageReceivedCommandAccepted(const CCECAdapterMessage &message)
194	{
195	bool bSendSignal(false);
196	bool bHandled(false);
197	{
198	CLockObject lock(m_mutex);
199	if (m_iPacketsLeft > 0)
200	{
201	/* decrease by 1 */
202	m_iPacketsLeft--;
203
204	#ifdef CEC_DEBUGGING
205	/* log this message */
206	CStdString strLog;
207	strLog.Format("%s - command accepted", ToString());
208	if (m_iPacketsLeft > 0)
209	strLog.AppendFormat(" - waiting for %d more", m_iPacketsLeft);
210	m_queue->m_com->m_callback->GetLib()->AddLog(CEC_LOG_DEBUG, strLog);
211	#endif
212
213	/* no more packets left and not a transmission, so we're done */
214	if (!m_message->IsTranmission() && m_iPacketsLeft == 0)
215	{
216	m_message->state = ADAPTER_MESSAGE_STATE_SENT_ACKED;
217	m_message->response = message.packet;
218	bSendSignal = true;
219	}
220	bHandled = true;
221	}
222	}
223
224	if (bSendSignal)
225	Signal();
226
227	return bHandled;
228	}
229
230	bool CCECAdapterMessageQueueEntry::MessageReceivedTransmitSucceeded(const CCECAdapterMessage &message)
231	{
232	{
233	CLockObject lock(m_mutex);
234	if (m_iPacketsLeft == 0)
235	{
236	/* transmission succeeded, so we're done */
237	#ifdef CEC_DEBUGGING
238	m_queue->m_com->m_callback->GetLib()->AddLog(CEC_LOG_DEBUG, "%s - transmit succeeded", m_message->ToString().c_str());
239	#endif
240	m_message->state = ADAPTER_MESSAGE_STATE_SENT_ACKED;
241	m_message->response = message.packet;
242	}
243	else
244	{
245	/* error, we expected more acks
246	since the messages are processed in order, this should not happen, so this is an error situation */
247	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);
248	m_message->state = ADAPTER_MESSAGE_STATE_ERROR;
249	}
250	}
251
252	Signal();
253
254	return true;
255	}
256
257	bool CCECAdapterMessageQueueEntry::MessageReceivedResponse(const CCECAdapterMessage &message)
258	{
259	{
260	CLockObject lock(m_mutex);
261	#ifdef CEC_DEBUGGING
262	m_queue->m_com->m_callback->GetLib()->AddLog(CEC_LOG_DEBUG, "%s - received response - %s", ToString(), message.ToString().c_str());
263	#else
264	if (message.IsError())
265	m_queue->m_com->m_callback->GetLib()->AddLog(CEC_LOG_DEBUG, "%s - received response - %s", ToString(), message.ToString().c_str());
266	#endif
267	m_message->response = message.packet;
268	if (m_message->IsTranmission())
269	m_message->state = message.Message() == MSGCODE_TRANSMIT_SUCCEEDED ? ADAPTER_MESSAGE_STATE_SENT_ACKED : ADAPTER_MESSAGE_STATE_SENT_NOT_ACKED;
270	else
271	m_message->state = ADAPTER_MESSAGE_STATE_SENT_ACKED;
272	}
273
274	Signal();
275
276	return true;
277	}
278
279	bool CCECAdapterMessageQueueEntry::ProvidesExtendedResponse(void)
280	{
281	return m_queue && m_queue->ProvidesExtendedResponse();
282	}
283
284	bool CCECAdapterMessageQueueEntry::TimedOutOrSucceeded(void) const
285	{
286	return m_message->bFireAndForget && (m_bSucceeded \|\| m_queueTimeout.TimeLeft() == 0);
287	}
288
289	CCECAdapterMessageQueue::CCECAdapterMessageQueue(CUSBCECAdapterCommunication *com) :
290	PLATFORM::CThread(),
291	m_com(com),
292	m_iNextMessage(0)
293	{
294	m_incomingAdapterMessage = new CCECAdapterMessage;
295	m_currentCECFrame.Clear();
296	}
297
298	CCECAdapterMessageQueue::~CCECAdapterMessageQueue(void)
299	{
300	StopThread(-1);
301	Clear();
302	StopThread();
303	delete m_incomingAdapterMessage;
304	}
305
306	void CCECAdapterMessageQueue::Clear(void)
307	{
308	StopThread(5);
309	CLockObject lock(m_mutex);
310	m_writeQueue.Clear();
311	m_messages.clear();
312	}
313
314	void *CCECAdapterMessageQueue::Process(void)
315	{
316	CCECAdapterMessageQueueEntry *message(NULL);
317	while (!IsStopped())
318	{
319	/* wait for a new message */
320	if (m_writeQueue.Pop(message, MESSAGE_QUEUE_SIGNAL_WAIT_TIME) && message)
321	{
322	/* write this message */
323	{
324	CLockObject lock(m_mutex);
325	m_com->WriteToDevice(message->m_message);
326	}
327	if (message->m_message->state == ADAPTER_MESSAGE_STATE_ERROR \|\|
328	message->m_message->Message() == MSGCODE_START_BOOTLOADER)
329	{
330	message->Signal();
331	Clear();
332	break;
333	}
334	}
335
336	CheckTimedOutMessages();
337	}
338	return NULL;
339	}
340
341	void CCECAdapterMessageQueue::CheckTimedOutMessages(void)
342	{
343	CLockObject lock(m_mutex);
344	vector<uint64_t> timedOut;
345	for (map<uint64_t, CCECAdapterMessageQueueEntry *>::iterator it = m_messages.begin(); it != m_messages.end(); it++)
346	{
347	if (it->second->TimedOutOrSucceeded())
348	{
349	timedOut.push_back(it->first);
350	if (!it->second->m_bSucceeded)
351	m_com->m_callback->GetLib()->AddLog(CEC_LOG_DEBUG, "command '%s' was not acked by the controller", CCECAdapterMessage::ToString(it->second->m_message->Message()));
352	delete it->second->m_message;
353	delete it->second;
354	}
355	}
356
357	for (vector<uint64_t>::iterator it = timedOut.begin(); it != timedOut.end(); it++)
358	{
359	uint64_t iEntryId = *it;
360	m_messages.erase(iEntryId);
361	}
362	}
363
364	void CCECAdapterMessageQueue::MessageReceived(const CCECAdapterMessage &msg)
365	{
366	bool bHandled(false);
367	CLockObject lock(m_mutex);
368	/* send the received message to each entry in the queue until it is handled */
369	for (map<uint64_t, CCECAdapterMessageQueueEntry *>::iterator it = m_messages.begin(); !bHandled && it != m_messages.end(); it++)
370	bHandled = it->second->MessageReceived(msg);
371
372	if (!bHandled)
373	{
374	/* the message wasn't handled */
375	bool bIsError(m_com->HandlePoll(msg));
376	#ifdef CEC_DEBUGGING
377	m_com->m_callback->GetLib()->AddLog(bIsError ? CEC_LOG_WARNING : CEC_LOG_DEBUG, msg.ToString().c_str());
378	#else
379	if (bIsError)
380	m_com->m_callback->GetLib()->AddLog(CEC_LOG_WARNING, msg.ToString().c_str());
381	#endif
382
383	/* push this message to the current frame */
384	if (!bIsError && msg.PushToCecCommand(m_currentCECFrame))
385	{
386	/* and push the current frame back over the callback method when a full command was received */
387	if (m_com->IsInitialised())
388	m_com->m_callback->OnCommandReceived(m_currentCECFrame);
389
390	/* clear the current frame */
391	m_currentCECFrame.Clear();
392	}
393	}
394	}
395
396	void CCECAdapterMessageQueue::AddData(uint8_t *data, size_t iLen)
397	{
398	for (size_t iPtr = 0; iPtr < iLen; iPtr++)
399	{
400	bool bFullMessage(false);
401	{
402	CLockObject lock(m_mutex);
403	bFullMessage = m_incomingAdapterMessage->PushReceivedByte(data[iPtr]);
404	}
405
406	if (bFullMessage)
407	{
408	/* a full message was received */
409	CCECAdapterMessage newMessage;
410	newMessage.packet = m_incomingAdapterMessage->packet;
411	MessageReceived(newMessage);
412
413	/* clear the current message */
414	CLockObject lock(m_mutex);
415	m_incomingAdapterMessage->Clear();
416	}
417	}
418	}
419
420	bool CCECAdapterMessageQueue::Write(CCECAdapterMessage *msg)
421	{
422	msg->state = ADAPTER_MESSAGE_STATE_WAITING_TO_BE_SENT;
423
424	/* set the correct line timeout */
425	if (msg->IsTranmission())
426	{
427	m_com->SetLineTimeout(msg->lineTimeout);
428	}
429
430	CCECAdapterMessageQueueEntry *entry = new CCECAdapterMessageQueueEntry(this, msg);
431	if (!entry)
432	{
433	m_com->m_callback->GetLib()->AddLog(CEC_LOG_ERROR, "couldn't create queue entry for '%s'", CCECAdapterMessage::ToString(msg->Message()));
434	msg->state = ADAPTER_MESSAGE_STATE_ERROR;
435	return false;
436	}
437
438	uint64_t iEntryId(0);
439	/* add to the wait for ack queue */
440	if (msg->Message() != MSGCODE_START_BOOTLOADER)
441	{
442	CLockObject lock(m_mutex);
443	iEntryId = m_iNextMessage++;
444	m_messages.insert(make_pair(iEntryId, entry));
445	}
446
447	/* add the message to the write queue */
448	m_writeQueue.Push(entry);
449
450	bool bReturn(true);
451	if (!msg->bFireAndForget)
452	{
453	if (!entry->Wait(msg->transmit_timeout <= 5 ? CEC_DEFAULT_TRANSMIT_WAIT : msg->transmit_timeout))
454	{
455	m_com->m_callback->GetLib()->AddLog(CEC_LOG_DEBUG, "command '%s' was not acked by the controller", CCECAdapterMessage::ToString(msg->Message()));
456	msg->state = ADAPTER_MESSAGE_STATE_SENT_NOT_ACKED;
457	bReturn = false;
458	}
459
460	if (msg->Message() != MSGCODE_START_BOOTLOADER)
461	{
462	CLockObject lock(m_mutex);
463	m_messages.erase(iEntryId);
464	}
465
466	if (msg->ReplyIsError() && msg->state != ADAPTER_MESSAGE_STATE_SENT_NOT_ACKED)
467	msg->state = ADAPTER_MESSAGE_STATE_ERROR;
468
469	delete entry;
470	}
471
472	return bReturn;
473	}
474
475	bool CCECAdapterMessageQueue::ProvidesExtendedResponse(void)
476	{
477	return m_com && m_com->ProvidesExtendedResponse();
478	}