| 77 | |
| 78 | template<typename Sample, typename TypedDataReader> |
| 79 | bool |
| 80 | MultiTopicDataReader_T<Sample, TypedDataReader>::join( |
| 81 | SampleVec& resulting, const SampleWithInfo& prototype, |
| 82 | const std::vector<OPENDDS_STRING>& key_names, const void* key_data, |
| 83 | DDS::DataReader_ptr other_dr, const MetaStruct& other_meta) |
| 84 | { |
| 85 | using namespace DDS; |
| 86 | DataReaderImpl* other_dri = dynamic_cast<DataReaderImpl*>(other_dr); |
| 87 | if (!other_dri) { |
| 88 | ACE_ERROR((LM_ERROR, ACE_TEXT("(%P|%t) ERROR: MultiTopicDataReader_T::join: ") |
| 89 | ACE_TEXT("Failed to get DataReaderImpl.\n"))); |
| 90 | return false; |
| 91 | } |
| 92 | |
| 93 | TopicDescription_var other_td = other_dri->get_topicdescription(); |
| 94 | CORBA::String_var other_topic = other_td->get_name(); |
| 95 | const QueryPlan& other_qp = query_plans_[other_topic.in()]; |
| 96 | const size_t n_keys = key_names.size(); |
| 97 | |
| 98 | if (n_keys > 0 && other_meta.numDcpsKeys() == n_keys) { // complete key |
| 99 | InstanceHandle_t ih = other_dri->lookup_instance_generic(key_data); |
| 100 | if (ih != HANDLE_NIL) { |
| 101 | GenericData other_data(other_meta, false); |
| 102 | SampleInfo info; |
| 103 | const ReturnCode_t ret = other_dri->read_instance_generic(other_data.ptr_, |
| 104 | info, ih, READ_SAMPLE_STATE, ANY_VIEW_STATE, ALIVE_INSTANCE_STATE); |
| 105 | if (ret != DDS::RETCODE_OK && ret != DDS::RETCODE_NO_DATA) { |
| 106 | if (log_level >= LogLevel::Notice) { |
| 107 | ACE_ERROR((LM_NOTICE, "(%P|%t) NOTICE: MultiTopicDataReader_T::join: read_instance_generic" |
| 108 | " for topic %C returns %C\n", other_topic.in(), retcode_to_string(ret))); |
| 109 | } |
| 110 | return false; |
| 111 | } |
| 112 | if (ret == DDS::RETCODE_NO_DATA || !info.valid_data) { |
| 113 | return false; |
| 114 | } |
| 115 | |
| 116 | resulting.push_back(prototype); |
| 117 | resulting.back().combine(SampleWithInfo(other_topic.in(), info)); |
| 118 | assign_fields(resulting.back().sample_, other_data.ptr_, other_qp, other_meta); |
| 119 | } |
| 120 | } else { // incomplete key or cross-join (0 key fields) |
| 121 | ReturnCode_t ret = RETCODE_OK; |
| 122 | for (InstanceHandle_t ih = HANDLE_NIL; ret != RETCODE_NO_DATA;) { |
| 123 | GenericData other_data(other_meta, false); |
| 124 | SampleInfo info; |
| 125 | ret = other_dri->read_next_instance_generic(other_data.ptr_, |
| 126 | info, ih, READ_SAMPLE_STATE, ANY_VIEW_STATE, ALIVE_INSTANCE_STATE); |
| 127 | if (ret != RETCODE_OK && ret != RETCODE_NO_DATA) { |
| 128 | if (log_level >= LogLevel::Notice) { |
| 129 | ACE_ERROR((LM_NOTICE, "(%P|%t) NOTICE: MultiTopicDataReader_T::join:" |
| 130 | " read_next_instance_generic for topic %C returns %C\n", |
| 131 | other_topic.in(), retcode_to_string(ret))); |
| 132 | } |
| 133 | return false; |
| 134 | } |
| 135 | if (ret == RETCODE_NO_DATA || !info.valid_data) { |
| 136 | break; |
nothing calls this directly
no test coverage detected