| 476 | } |
| 477 | |
| 478 | void ReactorTask::on_data_available(InternalDataReader_rch reader) |
| 479 | { |
| 480 | OpenDDS::DCPS::ConfigReader::SampleSequence samples; |
| 481 | OpenDDS::DCPS::InternalSampleInfoSequence infos; |
| 482 | reader->read(samples, infos, DDS::LENGTH_UNLIMITED, |
| 483 | DDS::NOT_READ_SAMPLE_STATE, DDS::ANY_VIEW_STATE, DDS::ANY_INSTANCE_STATE); |
| 484 | for (size_t idx = 0; idx != samples.size(); ++idx) { |
| 485 | if (infos[idx].valid_data && samples[idx].key() == COMMON_DCPS_THREAD_STATUS_INTERVAL) { |
| 486 | const TimeDuration per(std::atoi(samples[idx].value().c_str())); |
| 487 | if (per == thread_status_period_) { |
| 488 | continue; |
| 489 | } |
| 490 | thread_status_period_ = per; |
| 491 | if (thread_status_timer_ != ReactorWrapper::InvalidTimerId) { |
| 492 | reactor_wrapper_.cancel(thread_status_timer_); |
| 493 | } |
| 494 | if (per) { |
| 495 | if (!tsm_updater_handler_) { |
| 496 | tsm_updater_handler_ = make_rch<ThreadStatusManager::Updater>(); |
| 497 | } |
| 498 | |
| 499 | thread_status_timer_ = reactor_wrapper_.schedule(*tsm_updater_handler_, thread_status_manager_, per, per); |
| 500 | if (thread_status_timer_ == ReactorWrapper::InvalidTimerId && log_level >= LogLevel::Notice) { |
| 501 | ACE_ERROR((LM_NOTICE, "(%P|%t) NOTICE: ReactorTask::on_data_available: failed to " |
| 502 | "schedule timer for ThreadStatusManager::Updater\n")); |
| 503 | } |
| 504 | } |
| 505 | } |
| 506 | } |
| 507 | } |
| 508 | |
| 509 | bool ReactorWrapper::open(ACE_Reactor* reactor) |
| 510 | { |