a270a78235792a39a81a6fd746088c7155cea175
[deb_libcec.git] / 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) && message)
236 {
237 /* write this message */
238 {
239 CLockObject lock(m_mutex);
240 m_com->WriteToDevice(message->m_message);
241 }
242 if (message->m_message->state == ADAPTER_MESSAGE_STATE_ERROR ||
243 message->m_message->Message() == MSGCODE_START_BOOTLOADER)
244 {
245 message->Signal();
246 Clear();
247 break;
248 }
249 }
250 }
251 return NULL;
252 }
253
254 void CCECAdapterMessageQueue::MessageReceived(const CCECAdapterMessage &msg)
255 {
256 bool bHandled(false);
257 CLockObject lock(m_mutex);
258 /* send the received message to each entry in the queue until it is handled */
259 for (map<uint64_t, CCECAdapterMessageQueueEntry *>::iterator it = m_messages.begin(); !bHandled && it != m_messages.end(); it++)
260 bHandled = it->second->MessageReceived(msg);
261
262 if (!bHandled)
263 {
264 /* the message wasn't handled */
265 bool bIsError(m_com->HandlePoll(msg));
266 CLibCEC::AddLog(bIsError ? CEC_LOG_WARNING : CEC_LOG_DEBUG, msg.ToString());
267
268 /* push this message to the current frame */
269 if (!bIsError && msg.PushToCecCommand(m_currentCECFrame))
270 {
271 /* and push the current frame back over the callback method when a full command was received */
272 if (m_com->IsInitialised())
273 m_com->m_callback->OnCommandReceived(m_currentCECFrame);
274
275 /* clear the current frame */
276 m_currentCECFrame.Clear();
277 }
278 }
279 }
280
281 void CCECAdapterMessageQueue::AddData(uint8_t *data, size_t iLen)
282 {
283 for (size_t iPtr = 0; iPtr < iLen; iPtr++)
284 {
285 bool bFullMessage(false);
286 {
287 CLockObject lock(m_mutex);
288 bFullMessage = m_incomingAdapterMessage.PushReceivedByte(data[iPtr]);
289 }
290
291 if (bFullMessage)
292 {
293 /* a full message was received */
294 CCECAdapterMessage newMessage;
295 newMessage.packet = m_incomingAdapterMessage.packet;
296 MessageReceived(newMessage);
297
298 /* clear the current message */
299 CLockObject lock(m_mutex);
300 m_incomingAdapterMessage.Clear();
301 }
302 }
303 }
304
305 bool CCECAdapterMessageQueue::Write(CCECAdapterMessage *msg)
306 {
307 msg->state = ADAPTER_MESSAGE_STATE_WAITING_TO_BE_SENT;
308
309 /* set the correct line timeout */
310 if (msg->IsTranmission())
311 {
312 m_com->SetLineTimeout(msg->lineTimeout);
313 }
314
315 CCECAdapterMessageQueueEntry *entry = new CCECAdapterMessageQueueEntry(msg);
316 uint64_t iEntryId(0);
317 /* add to the wait for ack queue */
318 if (msg->Message() != MSGCODE_START_BOOTLOADER)
319 {
320 CLockObject lock(m_mutex);
321 iEntryId = m_iNextMessage++;
322 m_messages.insert(make_pair(iEntryId, entry));
323 }
324
325 /* add the message to the write queue */
326 m_writeQueue.Push(entry);
327
328 bool bReturn(true);
329 if (entry)
330 {
331 if (!entry->Wait(msg->transmit_timeout <= 5 ? CEC_DEFAULT_TRANSMIT_WAIT : msg->transmit_timeout))
332 {
333 CLibCEC::AddLog(CEC_LOG_DEBUG, "command '%s' was not acked by the controller", CCECAdapterMessage::ToString(msg->Message()));
334 msg->state = ADAPTER_MESSAGE_STATE_SENT_NOT_ACKED;
335 bReturn = false;
336 }
337
338 if (msg->Message() != MSGCODE_START_BOOTLOADER)
339 {
340 CLockObject lock(m_mutex);
341 m_messages.erase(iEntryId);
342 }
343 delete entry;
344 }
345
346 return bReturn;
347 }