847115abcb1e6f58e3ce875ded4453c98c2906a0
[deb_libcec.git] / 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 (m_message->IsTranmission() && 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 }