Revision 65dc89cb src/urt_subscriber.c
| src/urt_subscriber.c | ||
|---|---|---|
| 41 | 41 |
/* LOCAL FUNCTIONS */ |
| 42 | 42 |
/******************************************************************************/ |
| 43 | 43 |
|
| 44 |
void urtFetchMessage () |
|
| 45 |
{
|
|
| 46 |
//TODO: Update message pointer |
|
| 47 |
//TODO: Copy message origin time |
|
| 48 |
//TODO: Copy message payload |
|
| 49 |
} |
|
| 50 |
|
|
| 51 |
urt_message_t* urtFindOldestMessage(urt_message_t* oldestMessage) |
|
| 52 |
{
|
|
| 53 |
while (oldestMessage->next->originTime < oldestMessage->originTime) |
|
| 54 |
{
|
|
| 55 |
oldestMessage = oldestMessage->next; |
|
| 56 |
} |
|
| 57 |
return oldestMessage; |
|
| 58 |
} |
|
| 59 |
|
|
| 60 |
urt_message_t* urtFindLatestMessage(urt_message_t* latestMessage) |
|
| 61 |
{
|
|
| 62 |
urt_message_t* lastMessage = subscriber->base.lastMessage; |
|
| 63 |
while (lastMessage->next->originTime < lastMessage->originTime) |
|
| 64 |
{
|
|
| 65 |
lastMessage = lastMessage->next; |
|
| 66 |
#if (URT_CFG_PUBSUB_PROFILING == true) |
|
| 67 |
subscriber->base.lastMessage->numConsumersLeft--; |
|
| 68 |
subscriber->base->numMessagesReceived++; |
|
| 69 |
#endif /* URT_CFG_PUBSUB_PROFILING */ |
|
| 70 |
} |
|
| 71 |
} |
|
| 72 |
|
|
| 44 | 73 |
/******************************************************************************/ |
| 45 | 74 |
/* EXPORTED FUNCTIONS */ |
| 46 | 75 |
/******************************************************************************/ |
| ... | ... | |
| 126 | 155 |
* Retruns URT_STATUS_FETCH_NOMESSAGE if there is no new message to fetch. |
| 127 | 156 |
*/ |
| 128 | 157 |
urt_status_t urtNrtSubscriberFetchNextMessage (urt_nrtsubscriber_t* subscriber, void* payload, size_t bytes, urt_delay_t* latency) |
| 129 |
{
|
|
| 158 |
{
|
|
| 130 | 159 |
urtDebugAssert(subscriber); |
| 160 |
|
|
| 161 |
if (!subscriber->base.topic) |
|
| 162 |
return URT_STATUS_FETCH_NOTOPIC; |
|
| 163 |
|
|
| 164 |
urtMutexLock(subscriber->base.topic->lock); |
|
| 165 |
|
|
| 166 |
urt_message_t* messageTemp = subscriber->base.lastMessage; |
|
| 167 |
if(messageTemp->originTime == subscriber->base.lastMessageTime) |
|
| 168 |
{
|
|
| 169 |
if (subscriber->base.lastMessage->next->originTime > subscriber->base.lastMessageTime) |
|
| 170 |
{
|
|
| 171 |
urtMutexUnlock(subscriber->base.topic->lock); |
|
| 172 |
return URT_STATUS_FETCH_NOMESSAGE; |
|
| 173 |
} |
|
| 174 |
messageTemp = messageTemp->next; |
|
| 175 |
} |
|
| 176 |
else |
|
| 177 |
{
|
|
| 178 |
messageTemp = urtFindOldestMessage(messageTemp->next); |
|
| 179 |
} |
|
| 180 |
|
|
| 181 |
urtFetchMessage(); |
|
| 182 |
|
|
| 183 |
uint64_t calculatedLatency = NULL; //TODO: Calculate message latency |
|
| 184 |
#if(URT_CFG_PUBSUB_PROFILING == true) |
|
| 185 |
subscriber->base.sumLatencies += calculatedLatency; |
|
| 186 |
|
|
| 187 |
if (calculatedLatency < subscriber->minLatency) |
|
| 188 |
{
|
|
| 189 |
subscriber->minLatency = calculatedLatency; |
|
| 190 |
} |
|
| 191 |
else if (calculatedLatency > subscriber->maxLatency) |
|
| 192 |
{
|
|
| 193 |
subscriber->maxLatency = calculatedLatency; |
|
| 194 |
} |
|
| 195 |
#endif /* URT_CFG_PUBSUB_PROFILING */ |
|
| 196 |
bool temp = false; |
|
| 197 |
if (temp/*optional latency output argument given*/) |
|
| 198 |
{
|
|
| 199 |
latency = calculatedLatency; |
|
| 200 |
} |
|
| 201 |
|
|
| 202 |
#if (URT_CFG_PUBSUB_PROFILING == true) |
|
| 203 |
subscriber->base.lastMessage->numConsumersLeft--; |
|
| 204 |
subscriber->base->numMessagesReceived++; |
|
| 205 |
#endif /* URT_CFG_PUBSUB_PROFILING */ |
|
| 206 |
|
|
| 207 |
urtMutexUnlock(subscriber->base.topic->lock); |
|
| 131 | 208 |
return URT_STATUS_OK; |
| 132 | 209 |
} |
| 133 | 210 |
|
| ... | ... | |
| 144 | 221 |
* Returns URT_STATUS_FETCH_NOMESSAGE if there is no new message to fetch. |
| 145 | 222 |
*/ |
| 146 | 223 |
urt_status_t urtNrtSubscriberFetchLatestMessage (urt_nrtsubscriber_t* subscriber, void* payload, size_t bytes, urt_delay_t* latency) {
|
| 224 |
urtDebugAssert(subscriber); |
|
| 225 |
|
|
| 226 |
if (!subscriber->base.topic) |
|
| 227 |
return URT_STATUS_FETCH_NOTOPIC; |
|
| 228 |
|
|
| 229 |
urtMutexLock(subscriber->base.topic->lock); |
|
| 230 |
urt_message_t* lastMessage = urtFindLatestMessage(subscriber->base.lastMessage); |
|
| 231 |
|
|
| 232 |
if (lastMessage->originTime == subscriber->base.lastMessageTime) |
|
| 233 |
{
|
|
| 234 |
urtMutexUnlock(subscriber->base.topic->lock); |
|
| 235 |
return URT_STATUS_FETCH_NOMESSAGE; |
|
| 236 |
} |
|
| 237 |
|
|
| 238 |
urtFetchMessage(); |
|
| 239 |
|
|
| 240 |
uint64_t calculatedLatency = NULL; //TODO: Calculate message latency |
|
| 241 |
#if(URT_CFG_PUBSUB_PROFILING == true) |
|
| 242 |
subscriber->base.sumLatencies += calculatedLatency; |
|
| 243 |
|
|
| 244 |
if (calculatedLatency < subscriber->minLatency) |
|
| 245 |
{
|
|
| 246 |
subscriber->minLatency = calculatedLatency; |
|
| 247 |
} |
|
| 248 |
else if (calculatedLatency > subscriber->maxLatency) |
|
| 249 |
{
|
|
| 250 |
subscriber->maxLatency = calculatedLatency; |
|
| 251 |
} |
|
| 252 |
#endif /* URT_CFG_PUBSUB_PROFILING */ |
|
| 253 |
bool temp = false; |
|
| 254 |
if (temp/*optional latency output argument given*/) |
|
| 255 |
{
|
|
| 256 |
latency = calculatedLatency; |
|
| 257 |
} |
|
| 258 |
|
|
| 259 |
urtMutexUnlock(subscriber->base.topic->lock); |
|
| 147 | 260 |
return URT_STATUS_OK; |
| 148 | 261 |
} |
| 149 | 262 |
|
| ... | ... | |
| 276 | 389 |
* Returns URT_STATUS_FETCH_NOMESSAGE if there is no new message to fetch. |
| 277 | 390 |
*/ |
| 278 | 391 |
urt_status_t urtSrtSubscriberFetchNextMessage(urt_srtsubscriber_t* subscriber, void* payload, |
| 279 |
size_t bytes, urt_delay_t* latency){return URT_STATUS_OK;}
|
|
| 392 |
size_t bytes, urt_delay_t* latency) |
|
| 393 |
{
|
|
| 394 |
urtDebugAssert(subscriber); |
|
| 395 |
|
|
| 396 |
if (!subscriber->base.topic) |
|
| 397 |
return URT_STATUS_FETCH_NOTOPIC; |
|
| 398 |
|
|
| 399 |
urtMutexLock(subscriber->base.topic->lock); |
|
| 400 |
|
|
| 401 |
urt_message_t* messageTemp = subscriber->base.lastMessage; |
|
| 402 |
if(messageTemp->originTime == subscriber->base.lastMessageTime) |
|
| 403 |
{
|
|
| 404 |
if (subscriber->base.lastMessage->next->originTime > subscriber->base.lastMessageTime) |
|
| 405 |
{
|
|
| 406 |
urtMutexUnlock(subscriber->base.topic->lock); |
|
| 407 |
return URT_STATUS_FETCH_NOMESSAGE; |
|
| 408 |
} |
|
| 409 |
messageTemp = messageTemp->next; |
|
| 410 |
} |
|
| 411 |
else |
|
| 412 |
{
|
|
| 413 |
messageTemp = urtFindOldestMessage(messageTemp->next); |
|
| 414 |
} |
|
| 415 |
|
|
| 416 |
urtFetchMessage(); |
|
| 417 |
|
|
| 418 |
uint64_t calculatedLatency = NULL; //TODO: Calculate message latency |
|
| 419 |
#if(URT_CFG_PUBSUB_PROFILING == true) |
|
| 420 |
subscriber->base.sumLatencies += calculatedLatency; |
|
| 421 |
|
|
| 422 |
if (calculatedLatency < subscriber->minLatency) |
|
| 423 |
{
|
|
| 424 |
subscriber->minLatency = calculatedLatency; |
|
| 425 |
} |
|
| 426 |
else if (calculatedLatency > subscriber->maxLatency) |
|
| 427 |
{
|
|
| 428 |
subscriber->maxLatency = calculatedLatency; |
|
| 429 |
} |
|
| 430 |
#endif /* URT_CFG_PUBSUB_PROFILING */ |
|
| 431 |
bool temp = false; |
|
| 432 |
if (temp/*optional latency output argument given*/) |
|
| 433 |
{
|
|
| 434 |
latency = calculatedLatency; |
|
| 435 |
} |
|
| 436 |
|
|
| 437 |
#if (URT_CFG_PUBSUB_PROFILING == true) |
|
| 438 |
subscriber->base.lastMessage->numConsumersLeft--; |
|
| 439 |
subscriber->base->numMessagesReceived++; |
|
| 440 |
#endif /* URT_CFG_PUBSUB_PROFILING */ |
|
| 441 |
|
|
| 442 |
urtMutexUnlock(subscriber->base.topic->lock); |
|
| 443 |
return URT_STATUS_OK; |
|
| 444 |
} |
|
| 280 | 445 |
|
| 281 | 446 |
/** |
| 282 | 447 |
* @brief Fetches the latest message. |
| ... | ... | |
| 291 | 456 |
* Returns URT_STATUS_FETCH_NOMESSAGE if there is no new message to fetch. |
| 292 | 457 |
*/ |
| 293 | 458 |
urt_status_t urtSrtSubscriberFetchLatestMessage(urt_srtsubscriber_t* subscriber, void* payload, |
| 294 |
size_t bytes, urt_delay_t* latency){return URT_STATUS_OK;}
|
|
| 459 |
size_t bytes, urt_delay_t* latency) |
|
| 460 |
{
|
|
| 461 |
urtDebugAssert(subscriber); |
|
| 462 |
|
|
| 463 |
if (!subscriber->base.topic) |
|
| 464 |
return URT_STATUS_FETCH_NOTOPIC; |
|
| 465 |
|
|
| 466 |
urtMutexLock(subscriber->base.topic->lock); |
|
| 467 |
urt_message_t* lastMessage = urtFindLatestMessage(subscriber->base.lastMessage); |
|
| 468 |
|
|
| 469 |
if (lastMessage->originTime == subscriber->base.lastMessageTime) |
|
| 470 |
{
|
|
| 471 |
urtMutexUnlock(subscriber->base.topic->lock); |
|
| 472 |
return URT_STATUS_FETCH_NOMESSAGE; |
|
| 473 |
} |
|
| 474 |
|
|
| 475 |
urtFetchMessage(); |
|
| 476 |
|
|
| 477 |
uint64_t calculatedLatency = NULL; //TODO: Calculate message latency |
|
| 478 |
#if(URT_CFG_PUBSUB_PROFILING == true) |
|
| 479 |
subscriber->base.sumLatencies += calculatedLatency; |
|
| 480 |
|
|
| 481 |
if (calculatedLatency < subscriber->minLatency) |
|
| 482 |
{
|
|
| 483 |
subscriber->minLatency = calculatedLatency; |
|
| 484 |
} |
|
| 485 |
else if (calculatedLatency > subscriber->maxLatency) |
|
| 486 |
{
|
|
| 487 |
subscriber->maxLatency = calculatedLatency; |
|
| 488 |
} |
|
| 489 |
#endif /* URT_CFG_PUBSUB_PROFILING */ |
|
| 490 |
bool temp = false; |
|
| 491 |
if (temp/*optional latency output argument given*/) |
|
| 492 |
{
|
|
| 493 |
latency = calculatedLatency; |
|
| 494 |
} |
|
| 495 |
|
|
| 496 |
urtMutexUnlock(subscriber->base.topic->lock); |
|
| 497 |
return URT_STATUS_OK; |
|
| 498 |
} |
|
| 295 | 499 |
|
| 296 | 500 |
/** |
| 297 | 501 |
* @brief Calculates the usefulness of the subscriber. |
| ... | ... | |
| 456 | 660 |
* Returns URT_STATUS_FETCH_NOMESSAGE if there is no new message to fetch. |
| 457 | 661 |
*/ |
| 458 | 662 |
urt_status_t urtFrtSubscriberFetchNextMessage(urt_frtsubscriber_t* subscriber, void* payload, |
| 459 |
size_t bytes, urt_delay_t* latency){return URT_STATUS_OK;}
|
|
| 663 |
size_t bytes, urt_delay_t* latency) |
|
| 664 |
{
|
|
| 665 |
urtDebugAssert(subscriber); |
|
| 666 |
|
|
| 667 |
if (!subscriber->base.topic) |
|
| 668 |
return URT_STATUS_FETCH_NOTOPIC; |
|
| 669 |
|
|
| 670 |
urtMutexLock(subscriber->base.topic->lock); |
|
| 671 |
|
|
| 672 |
urt_message_t* messageTemp = subscriber->base.lastMessage; |
|
| 673 |
if(messageTemp->originTime == subscriber->base.lastMessageTime) |
|
| 674 |
{
|
|
| 675 |
if (subscriber->base.lastMessage->next->originTime > subscriber->base.lastMessageTime) |
|
| 676 |
{
|
|
| 677 |
urtMutexUnlock(subscriber->base.topic->lock); |
|
| 678 |
return URT_STATUS_FETCH_NOMESSAGE; |
|
| 679 |
} |
|
| 680 |
messageTemp = messageTemp->next; |
|
| 681 |
} |
|
| 682 |
else |
|
| 683 |
{
|
|
| 684 |
messageTemp = urtFindOldestMessage(messageTemp->next); |
|
| 685 |
} |
|
| 686 |
|
|
| 687 |
urtFetchMessage(); |
|
| 688 |
|
|
| 689 |
uint64_t calculatedLatency = NULL; //TODO: Calculate message latency |
|
| 690 |
#if(URT_CFG_PUBSUB_PROFILING == true) |
|
| 691 |
subscriber->base.sumLatencies += calculatedLatency; |
|
| 692 |
#endif /* URT_CFG_PUBSUB_PROFILING */ |
|
| 693 |
|
|
| 694 |
#if (URT_CFG_PUBSUB_PROFILING == true && URT_CFG_PUBSUB_QOS_JITTERCHECKS == true) //TODO: Both must be true? otherwise jitter does not exist |
|
| 695 |
if (calculatedLatency < subscriber->minLatency && calculatedLatency > subscriber->minLatency-subscriber->maxJitter) |
|
| 696 |
{
|
|
| 697 |
subscriber->minLatency = calculatedLatency; |
|
| 698 |
} |
|
| 699 |
else if (calculatedLatency > subscriber->maxLatency && calculatedLatency < subscriber->maxLatency+subscriber->maxJitter) |
|
| 700 |
{
|
|
| 701 |
subscriber->maxLatency = calculatedLatency; |
|
| 702 |
} |
|
| 703 |
#endif /* URT_CFG_PUBSUB_PROFILING && URT_CFG_PUBSUB_QOS_JITTERCHECKS */ |
|
| 704 |
|
|
| 705 |
bool temp = false; |
|
| 706 |
if (temp/*optional latency output argument given*/) |
|
| 707 |
{
|
|
| 708 |
latency = calculatedLatency; |
|
| 709 |
} |
|
| 710 |
|
|
| 711 |
#if (URT_CFG_PUBSUB_PROFILING == true) |
|
| 712 |
subscriber->base.lastMessage->numConsumersLeft--; |
|
| 713 |
subscriber->base->numMessagesReceived++; |
|
| 714 |
#endif /* URT_CFG_PUBSUB_PROFILING */ |
|
| 715 |
|
|
| 716 |
urtMutexUnlock(subscriber->base.topic->lock); |
|
| 717 |
return URT_STATUS_OK; //TODO: or urt_status_jitterviolation |
|
| 718 |
} |
|
| 460 | 719 |
|
| 461 | 720 |
/** |
| 462 | 721 |
* @brief Fetches the latest message. |
| ... | ... | |
| 471 | 730 |
* Returns URT_STATUS_FETCH_NOMESSAGE if there is no new message to fetch. |
| 472 | 731 |
*/ |
| 473 | 732 |
urt_status_t urtFrtSubscriberFetchLatestMessage(urt_frtsubscriber_t* subscriber, void* payload, |
| 474 |
size_t bytes, urt_delay_t* latency){return URT_STATUS_OK;}
|
|
| 733 |
size_t bytes, urt_delay_t* latency) |
|
| 734 |
{
|
|
| 735 |
urtDebugAssert(subscriber); |
|
| 736 |
|
|
| 737 |
if (!subscriber->base.topic) |
|
| 738 |
return URT_STATUS_FETCH_NOTOPIC; |
|
| 739 |
|
|
| 740 |
urtMutexLock(subscriber->base.topic->lock); |
|
| 741 |
urt_message_t* lastMessage = urtFindLatestMessage(subscriber->base.lastMessage); |
|
| 742 |
|
|
| 743 |
if (lastMessage->originTime == subscriber->base.lastMessageTime) |
|
| 744 |
{
|
|
| 745 |
urtMutexUnlock(subscriber->base.topic->lock); |
|
| 746 |
return URT_STATUS_FETCH_NOMESSAGE; |
|
| 747 |
} |
|
| 748 |
|
|
| 749 |
urtFetchMessage(); |
|
| 750 |
|
|
| 751 |
uint64_t calculatedLatency = NULL; //TODO: Calculate message latency |
|
| 752 |
#if(URT_CFG_PUBSUB_PROFILING == true) |
|
| 753 |
subscriber->base.sumLatencies += calculatedLatency; |
|
| 754 |
#endif /* URT_CFG_PUBSUB_PROFILING */ |
|
| 755 |
|
|
| 756 |
#if (URT_CFG_PUBSUB_PROFILING == true && URT_CFG_PUBSUB_QOS_JITTERCHECKS == true) //TODO: Both must be true? otherwise jitter does not exist |
|
| 757 |
if (calculatedLatency < subscriber->minLatency && calculatedLatency > subscriber->minLatency-subscriber->maxJitter) |
|
| 758 |
{
|
|
| 759 |
subscriber->minLatency = calculatedLatency; |
|
| 760 |
} |
|
| 761 |
else if (calculatedLatency > subscriber->maxLatency && calculatedLatency < subscriber->maxLatency+subscriber->maxJitter) |
|
| 762 |
{
|
|
| 763 |
subscriber->maxLatency = calculatedLatency; |
|
| 764 |
} |
|
| 765 |
#endif /* URT_CFG_PUBSUB_PROFILING && URT_CFG_PUBSUB_QOS_JITTERCHECKS */ |
|
| 766 |
|
|
| 767 |
bool temp = false; |
|
| 768 |
if (temp/*optional latency output argument given*/) |
|
| 769 |
{
|
|
| 770 |
latency = calculatedLatency; |
|
| 771 |
} |
|
| 772 |
|
|
| 773 |
urtMutexUnlock(subscriber->base.topic->lock); |
|
| 774 |
return URT_STATUS_OK; |
|
| 775 |
} |
|
| 475 | 776 |
|
| 476 | 777 |
/** |
| 477 | 778 |
* @brief Calculates the validity from the subscriber. |
| ... | ... | |
| 668 | 969 |
* Returns URT_STATUS_FETCH_NOMESSAGE if there is no new message to fetch. |
| 669 | 970 |
*/ |
| 670 | 971 |
urt_status_t urtHrtSubscriberFetchNextMessage(urt_hrtsubscriber_t* subscriber, void* payload, |
| 671 |
size_t bytes, urt_delay_t* latency){return URT_STATUS_OK;}
|
|
| 972 |
size_t bytes, urt_delay_t* latency) |
|
| 973 |
{
|
|
| 974 |
urtDebugAssert(subscriber); |
|
| 975 |
|
|
| 976 |
if (!subscriber->base.topic) |
|
| 977 |
return URT_STATUS_FETCH_NOTOPIC; |
|
| 978 |
|
|
| 979 |
urtMutexLock(subscriber->base.topic->lock); |
|
| 980 |
urt_message_t* messageTemp = subscriber->base.lastMessage; |
|
| 981 |
if (messageTemp->next->originTime > messageTemp.originTime) |
|
| 982 |
{
|
|
| 983 |
urtMutexUnlock(subscriber->base.topic->lock); |
|
| 984 |
return URT_STATUS_FETCH_NOMESSAGE; |
|
| 985 |
} |
|
| 986 |
messageTemp = messageTemp->next; |
|
| 987 |
|
|
| 988 |
uint64_t calculatedLatency = NULL; //TODO: Calculate message latency |
|
| 989 |
#if (URT_CFG_PUBSUB_PROFILING == true) |
|
| 990 |
subscriber->base.sumLatencies += calculatedLatency; |
|
| 991 |
#endif /* URT_CFG_PUBSUB_PROFILING */ |
|
| 992 |
bool temp = false; |
|
| 993 |
if (temp /* optional latency output argument given */) |
|
| 994 |
{
|
|
| 995 |
latency = calculatedLatency |
|
| 996 |
} |
|
| 997 |
|
|
| 998 |
subscriber->base.lastMessage->numHrtConsumersLeft--; |
|
| 999 |
if (subscriber->base.lastMessage->numHrtConsumersLeft != 0) |
|
| 1000 |
{
|
|
| 1001 |
urtCondvarSignal(subscriber->base.topic->hrtReleased); |
|
| 1002 |
} |
|
| 1003 |
|
|
| 1004 |
#if (URT_CFG_PUBSUB_PROFILING == true) |
|
| 1005 |
subscriber->base.lastMessage->numConsumersLeft--; |
|
| 1006 |
subscriber->base->numMessagesReceived++; |
|
| 1007 |
#endif /* URT_CFG_PUBSUB_PROFILING */ |
|
| 1008 |
|
|
| 1009 |
#if (URT_CFG_PUBSUB_QOS_JITTERCHECKS == true || URT_CFG_PUBSUB_PROFILING == true) |
|
| 1010 |
if (temp /*latency is within allowed jitter range*/) |
|
| 1011 |
{
|
|
| 1012 |
if (calculatedLatency < subscriber->minLatency) |
|
| 1013 |
{
|
|
| 1014 |
subscriber->minLatency = calculatedLatency; |
|
| 1015 |
} |
|
| 1016 |
else if (calculatedLatency > subscriber->maxLatency) |
|
| 1017 |
{
|
|
| 1018 |
subscriber->maxLatency = calculatedLatency; |
|
| 1019 |
} |
|
| 1020 |
} |
|
| 1021 |
else |
|
| 1022 |
{
|
|
| 1023 |
#if (URT_CFG_PUBSUB_QOS_JITTERCHECKS == true) |
|
| 1024 |
urtMutexUnlock(subscriber->base.topic->lock); |
|
| 1025 |
urtCoreStopNodes(URT_STATUS_JITTERVIOLATION); |
|
| 1026 |
return URT_STATUS_JITTERVIOLATION; |
|
| 1027 |
#endif /* URT_CFG_PUBSUB_QOS_JITTERCHECKS */ |
|
| 1028 |
} |
|
| 1029 |
#endif /* URT_CFG_PUBSUB_QOS_JITTERCHECKS || URT_CFG_PUBSUB_PROFILING */ |
|
| 1030 |
|
|
| 1031 |
urtFetchMessage(); |
|
| 1032 |
|
|
| 1033 |
#if (URT_CFG_PUBSUB_QOS_DEADLINECHECKS == true) |
|
| 1034 |
if (messageTemp->next->originTime < messageTemp->originTime) |
|
| 1035 |
{
|
|
| 1036 |
//TODO: update qos deadliner timer wrt. next message |
|
| 1037 |
} |
|
| 1038 |
else |
|
| 1039 |
{
|
|
| 1040 |
//TODO: reset qos deadline timer |
|
| 1041 |
} |
|
| 1042 |
#endif /* URT_CFG_PUBSUB_QOS_DEADLINECHECKS */ |
|
| 1043 |
|
|
| 1044 |
urtMutexUnlock(subscriber->base.topic->lock); |
|
| 1045 |
return URT_STATUS_OK; |
|
| 1046 |
} |
|
| 672 | 1047 |
|
| 673 | 1048 |
|
| 674 | 1049 |
/** |
Also available in: Unified diff