c++/EEDB/SPStreams/FederatedSourceStream.cpp
/***** source_stream

  Description: set the input or source stream feeding objects into this level of the stream stack.
  Returntype : either an MQdb::DBStream or subclass of EEDB::SPStream object or undef if not set
  Exceptions : none 

*****/
source_stream
Description: set the input or source stream feeding objects into this level of the stream stack.
source_stream Description : set the input or source stream feeding objects into this level of the stream stack .
EEDB::SPStream*  EEDB::SPStreams::FederatedSourceStream::_build_source_stream() {
  if(_source_stream) { _source_stream->release(); _source_stream=NULL; }
  EEDB::SPStreams::MultiMergeStream *stream = _build_unfiltered_stream();
  stream->clear_sourcestream_filters();  
  map<string, bool>::iterator  it1;
  for(it1 = _filter_source_ids.begin(); it1 != _filter_source_ids.end(); it1++) {
    stream->add_source_id_filter((*it1).first);
  }
  if(!_filter_source_names.empty() or !_filter_experiment_searchlogic.empty()) {
    map<string, bool>  modified_filter_ids;
    if(!_filter_source_names.empty()) {
      stream->stream_data_sources("FeatureSource");
      while(EEDB::DataSource *obj = (EEDB::DataSource*)stream->next_in_stream()) {
        if(obj->classname() != EEDB::FeatureSource::class_name) { obj->release(); continue; }
        if(_filter_source_names[obj->name()]) {
          modified_filter_ids[obj->db_id()] = true;
        }
        obj->release();
      }
    }
    if(!_filter_experiment_searchlogic.empty()) {
      stream->stream_data_sources("Experiment", _filter_experiment_searchlogic);
      while(EEDB::DataSource *obj = (EEDB::DataSource*)stream->next_in_stream()) {
        if(obj->classname() != EEDB::Experiment::class_name) { obj->release(); continue; }
        modified_filter_ids[obj->db_id()] = true;
        obj->release();
      }
    }
    if(modified_filter_ids.empty()) {
      stream->release();
      _source_stream = new EEDB::SPStreams::StreamBuffer();
      return _source_stream;
    } else {
      stream->clear_sourcestream_filters();
      for(it1 = modified_filter_ids.begin(); it1 != modified_filter_ids.end(); it1++) {
        stream->add_source_id_filter((*it1).first);
      }
    }
  }
  stream->set_sourcestream_output(_sourcestream_output);
  map<string, EEDB::Datatype*>::iterator  it3;
  for(it3 = _expression_datatypes.begin(); it3 != _expression_datatypes.end(); it3++) {
    stream->add_expression_datatype_filter((*it3).second);
  }
  _source_stream = stream;
  return _source_stream;
}
EEDB :: SPStream * EEDB :: SPStreams :: FederatedSourceStream :: _build_source_stream ( ) { if ( _source_stream ) { _source_stream -> release ( ) ; _source_stream = NULL ; } EEDB :: SPStreams :: MultiMergeStream * stream = _build_unfiltered_stream ( ) ; stream -> clear_sourcestream_filters ( ) ; map < string , bool > :: iterator it1 ; for ( it1 = _filter_source_ids . begin ( ) ; it1 != _filter_source_ids . end ( ) ; it1 ++ ) { stream -> add_source_id_filter ( ( * it1 ) . first ) ; } if ( ! _filter_source_names . empty ( ) or ! _filter_experiment_searchlogic . empty ( ) ) { map < string , bool > modified_filter_ids ; if ( ! _filter_source_names . empty ( ) ) { stream -> stream_data_sources ( " " ) ; while ( EEDB :: DataSource * obj = ( EEDB :: DataSource * ) stream -> next_in_stream ( ) ) { if ( obj -> classname ( ) != EEDB :: FeatureSource :: class_name ) { obj -> release ( ) ; continue ; } if ( _filter_source_names [ obj -> name ( ) ] ) { modified_filter_ids [ obj -> db_id ( ) ] = true ; } obj -> release ( ) ; } } if ( ! _filter_experiment_searchlogic . empty ( ) ) { stream -> stream_data_sources ( " " , _filter_experiment_searchlogic ) ; while ( EEDB :: DataSource * obj = ( EEDB :: DataSource * ) stream -> next_in_stream ( ) ) { if ( obj -> classname ( ) != EEDB :: Experiment :: class_name ) { obj -> release ( ) ; continue ; } modified_filter_ids [ obj -> db_id ( ) ] = true ; obj -> release ( ) ; } } if ( modified_filter_ids . empty ( ) ) { stream -> release ( ) ; _source_stream = new EEDB :: SPStreams :: StreamBuffer ( ) ; return _source_stream ; } else { stream -> clear_sourcestream_filters ( ) ; for ( it1 = modified_filter_ids . begin ( ) ; it1 != modified_filter_ids . end ( ) ; it1 ++ ) { stream -> add_source_id_filter ( ( * it1 ) . first ) ; } } } stream -> set_sourcestream_output ( _sourcestream_output ) ; map < string , EEDB :: Datatype * > :: iterator it3 ; for ( it3 = _expression_datatypes . begin ( ) ; it3 != _expression_datatypes . end ( ) ; it3 ++ ) { stream -> add_expression_datatype_filter ( ( * it3 ) . second ) ; } _source_stream = stream ; return _source_stream ; }
source_stream
Description: set the input or source stream feeding objects into this level of the stream stack.
source_stream Description : set the input or source stream feeding objects into this level of the stream stack .
//for each peer, resets any source_id_filter() from previous queries //first add explicit source_ids filters //if _filter_source_names OR _filter_experiment_searchlogic is set //  then we need to stream_data_sources(), post filter, //  and then extend the add_source_id_filter() with additional ids //  these filters work to refine the previous explicit filters (peer_ids, source_ids)  //failed to find matches so need to return an empty stream //need to rebuild filter with the new modified_filter_ids