// createBlock
//
// Create the actual GNU Radio Block to that will perform the work method. The resulting
// block object is assigned to gr_stpr
//
// Add property change callbacks for getter/setter methods
//
void iir_filter_ffd_i::createBlock()
{
  //
  // gr_sptr = xxx_make_xxx( args );
  //
    this->gr_sptr = gr::filter::iir_filter_ffd::make( this->fftaps, this->fbtaps );
    this->setPropertyChangeListener("fftaps", this, &iir_filter_ffd_i::setTaps);
    this->setPropertyChangeListener("fbtaps", this, &iir_filter_ffd_i::setTaps);

  // 
  // Use setThrottle method to enable the throttling of consumption/production of data by the
  // service function. The affect of the throttle will try to pause the execution of the 
  // service function for a specified duration.  This duration is calculated using the getTargetDuration() method
  // and the actual processing time to perform the work method.
  //
  // This is turned ON by default for "output" only components
  //
  // setThrottle( bool onoff );

  // 
  // Use maintainTimeStamp to enable proper data flow of timestamp with input and output data. 
  // if turned on (true)
  //  The timestamp from the input source will be used and maintained based on the output rate and
  //  the number of samples produced
  // if turned off
  //   then the timestamp from the input source is passed through if available or the time of day
  // 
  // maintainTimestamp( bool onoff );

   setupIOMappings();
} 

void pll_carriertracking_cc_i::createBlock()
{
    //
    // gr_sptr = xxx_make_xxx( args );
    //

    gr_sptr = gr_make_pll_carriertracking_cc( this->loop_bw, this->max_freq, this->min_freq );

    this->registerGetValue("d_locksig", this, &pll_carriertracking_cc_i::lock_detector);
    this->setPropertyChangeListener("d_squelch_enable", this, &pll_carriertracking_cc_i::squelch_enable);
    this->setPropertyChangeListener("d_lock_threshold", this, &pll_carriertracking_cc_i::set_lock_threshold);

    // 
    // Use setThrottle method to enable the throttling of consumption/production of data by the
    // service function. The affect of the throttle will try to pause the execution of the 
    // service function for a specified duration.  This duration is calculated using the getTargetDuration() method
    // and the actual processing time to perform the work method.
    //
    // This is turned ON by default for "output" only components
    //
    // setThrottle( bool onoff );

    // 
    // Use maintainTimeStamp to enable proper data flow of timestamp with input and output data. 
    // if turned on (true)
    //  The timestamp from the input source will be used and maintained based on the output rate and
    //  the number of samples produced
    // if turned off
    //   then the timestamp from the input source is passed through if available or the time of day
    // 
    // maintainTimestamp( bool onoff );

    setupIOMappings();
}

// createBlock
//
// Create the actual GNU Radio Block to that will perform the work method. The resulting
// block object is assigned to gr_stpr
//
// Add property change callbacks for getter/setter methods
//
void pfb_clock_sync_ccf_4o_i::createBlock()
{
  //
  //
  gr_sptr = digital_make_pfb_clock_sync_ccf(sps, loop_bandwidth, taps, filter_size, init_phase, max_rate_deviation, osps);
   this->registerGetterSetter("loop_bandwidth", &digital_pfb_clock_sync_ccf::get_loop_bandwidth, &digital_pfb_clock_sync_ccf::set_loop_bandwidth);
   this->registerGetterSetter("damping_factor", &digital_pfb_clock_sync_ccf::get_damping_factor, &digital_pfb_clock_sync_ccf::set_damping_factor);
   this->registerGetterSetter("alpha", &digital_pfb_clock_sync_ccf::get_alpha, &digital_pfb_clock_sync_ccf::set_alpha);
   this->registerGetterSetter("beta", &digital_pfb_clock_sync_ccf::get_beta, &digital_pfb_clock_sync_ccf::set_beta);
   this->registerGetter("clock_rate", &digital_pfb_clock_sync_ccf::get_clock_rate);

  // 
  // Use setThrottle method to enable the throttling of consumption/production of data by the
  // service function. The affect of the throttle will try to pause the execution of the 
  // service function for a specified duration.  This duration is calculated using the getTargetDuration() method
  // and the actual processing time to perform the work method.
  //
  // This is turned ON by default for "output" only components
  //
  // setThrottle( bool onoff );

  // 
  // Use maintainTimeStamp to enable proper data flow of timestamp with input and output data. 
  // if turned on (true)
  //  The timestamp from the input source will be used and maintained based on the output rate and
  //  the number of samples produced
  // if turned off
  //   then the timestamp from the input source is passed through if available or the time of day
  // 
  // maintainTimestamp( bool onoff );

   setupIOMappings();
} 

void regenerate_bb_i::createBlock()
{
    //
    // gr_sptr = xxx_make_xxx( args );
    //

    gr_sptr = gr_make_regenerate_bb( this->period, this->max_regen );

    this->registerSetter("period", &gr_regenerate_bb::set_period);
    this->registerSetter("max_regen", &gr_regenerate_bb::set_max_regen);

    // 
    // Use setThrottle method to enable the throttling of consumption/production of data by the
    // service function. The affect of the throttle will try to pause the execution of the 
    // service function for a specified duration.  This duration is calculated using the getTargetDuration() method
    // and the actual processing time to perform the work method.
    //
    // This is turned ON by default for "output" only components
    //
    // setThrottle( bool onoff );

    // 
    // Use maintainTimeStamp to enable proper data flow of timestamp with input and output data. 
    // if turned on (true)
    //  The timestamp from the input source will be used and maintained based on the output rate and
    //  the number of samples produced
    // if turned off
    //   then the timestamp from the input source is passed through if available or the time of day
    // 
    // maintainTimestamp( bool onoff );

    setupIOMappings();
}

// createBlock
//
// Create the actual GNU Radio Block to that will perform the work method. The resulting
// block object is assigned to gr_stpr
//
// Add property change callbacks for getter/setter methods
//
void dc_blocker_cc_i::createBlock()
{
    //
    // gr_sptr = xxx_make_xxx( args );
    //
    gr_sptr = gr::filter::dc_blocker_cc::make( this->delay_length, this->long_form );
    this->registerGetValue("group_delay", this, &dc_blocker_cc_i::getGroupDelay);


    //
    // Use setThrottle method to enable the throttling of consumption/production of data by the
    // service function. The affect of the throttle will try to pause the execution of the
    // service function for a specified duration.  This duration is calculated using the getTargetDuration() method
    // and the actual processing time to perform the work method.
    //
    // This is turned ON by default for "output" only components
    //
    // setThrottle( bool onoff );

    //
    // Use maintainTimeStamp to enable proper data flow of timestamp with input and output data.
    // if turned on (true)
    //  The timestamp from the input source will be used and maintained based on the output rate and
    //  the number of samples produced
    // if turned off
    //   then the timestamp from the input source is passed through if available or the time of day
    //
    // maintainTimestamp( bool onoff );

    setupIOMappings();

}

void streams_to_stream_ii_4i_i::createBlock()
{
    //
    // gr_sptr = xxx_make_xxx( args );
    //

    try{
    	this->gr_sptr = gr_make_streams_to_stream(this->item_size, this->nstreams);
    }
    catch(...){
    	return;
    }

    // 
    // Use setThrottle method to enable the throttling of consumption/production of data by the
    // service function. The affect of the throttle will try to pause the execution of the 
    // service function for a specified duration.  This duration is calculated using the getTargetDuration() method
    // and the actual processing time to perform the work method.
    //
    // This is turned ON by default for "output" only components
    //
    // setThrottle( bool onoff );

    // 
    // Use maintainTimeStamp to enable proper data flow of timestamp with input and output data. 
    // if turned on (true)
    //  The timestamp from the input source will be used and maintained based on the output rate and
    //  the number of samples produced
    // if turned off
    //   then the timestamp from the input source is passed through if available or the time of day
    // 
    // maintainTimestamp( bool onoff );

    setupIOMappings();
}

void mpsk_snr_est_cc_i::createBlock()
{
    //
    // gr_sptr = xxx_make_xxx( args );
    //

    snr_est_type_t curr_type = SNR_EST_SIMPLE;
    if (this->type == 0){
        curr_type = SNR_EST_SIMPLE;
    } else if (this->type == 1){
        curr_type = SNR_EST_SKEW;
    } else if (this->type == 2){
        curr_type = SNR_EST_M2M4;
    } else if (this->type == 3){
        curr_type = SNR_EST_SVR;
    }

    gr_sptr = digital_make_mpsk_snr_est_cc( curr_type, this->tag_nsamples, this->alpha);

    this->registerGetterSetter("tag_nsamples",
                               &digital_mpsk_snr_est_cc::tag_nsample,
                               &digital_mpsk_snr_est_cc::set_tag_nsample);

    this->registerGetterSetter("alpha",
                               &digital_mpsk_snr_est_cc::alpha,
                               &digital_mpsk_snr_est_cc::set_alpha);

    this->setPropertyChangeListener("type",
                                    this,
                                    &mpsk_snr_est_cc_i::set_type);

    this->registerGetValue("snr",
                           this,
                           &mpsk_snr_est_cc_i::get_snr);

    // 
    // Use setThrottle method to enable the throttling of consumption/production of data by the
    // service function. The affect of the throttle will try to pause the execution of the 
    // service function for a specified duration.  This duration is calculated using the getTargetDuration() method
    // and the actual processing time to perform the work method.
    //
    // This is turned ON by default for "output" only components
    //
    // setThrottle( bool onoff );

    // 
    // Use maintainTimeStamp to enable proper data flow of timestamp with input and output data. 
    // if turned on (true)
    //  The timestamp from the input source will be used and maintained based on the output rate and
    //  the number of samples produced
    // if turned off
    //   then the timestamp from the input source is passed through if available or the time of day
    // 
    // maintainTimestamp( bool onoff );

    setupIOMappings();
}

void mpsk_receiver_cc_i::createBlock()
{
    //
    // gr_sptr = xxx_make_xxx( args );
    //

    gr_sptr = digital_make_mpsk_receiver_cc( this->M, this->theta, this->loop_bw, this->fmin, this->fmax,
                                             this->mu, this->gain_mu, this->omega, this->gain_omega, this->omega_rel);

    this->registerSetter("M",
                         &digital_mpsk_receiver_cc::set_modulation_order);
    this->registerGetValue("M",
                            this,
                            &mpsk_receiver_cc_i::modulation_order);
    this->registerGetterSetter("theta",
                               &digital_mpsk_receiver_cc::theta,
                               &digital_mpsk_receiver_cc::set_theta);
    this->registerGetterSetter("mu",
                               &digital_mpsk_receiver_cc::mu,
                               &digital_mpsk_receiver_cc::set_mu);
    this->registerGetterSetter("omega",
                               &digital_mpsk_receiver_cc::omega,
                               &digital_mpsk_receiver_cc::set_omega);
    this->registerGetterSetter("gain_mu",
                               &digital_mpsk_receiver_cc::gain_mu,
                               &digital_mpsk_receiver_cc::set_gain_mu);
    this->registerGetterSetter("gain_omega",
                               &digital_mpsk_receiver_cc::gain_omega,
                               &digital_mpsk_receiver_cc::set_gain_omega);
    this->registerGetterSetter("omega_rel",
                               &digital_mpsk_receiver_cc::gain_omega_rel,
                               &digital_mpsk_receiver_cc::set_gain_omega_rel);

    // 
    // Use setThrottle method to enable the throttling of consumption/production of data by the
    // service function. The affect of the throttle will try to pause the execution of the 
    // service function for a specified duration.  This duration is calculated using the getTargetDuration() method
    // and the actual processing time to perform the work method.
    //
    // This is turned ON by default for "output" only components
    //
    // setThrottle( bool onoff );

    // 
    // Use maintainTimeStamp to enable proper data flow of timestamp with input and output data. 
    // if turned on (true)
    //  The timestamp from the input source will be used and maintained based on the output rate and
    //  the number of samples produced
    // if turned off
    //   then the timestamp from the input source is passed through if available or the time of day
    // 
    // maintainTimestamp( bool onoff );

    setupIOMappings();
}

void cvsd_encode_sb_i::createBlock()
{
    //
    // gr_sptr = xxx_make_xxx( args );
    //

    try {
        gr_sptr = vocoder_make_cvsd_encode_sb( min_step, max_step, step_decay, accum_decay, K, J, pos_accum_max, neg_accum_max);
    } catch (...) {
        return;
    }

    this->registerGetValue("min_step", this, &cvsd_encode_sb_i::get_min_step);
    this->registerGetValue("max_step", this, &cvsd_encode_sb_i::get_max_step);
    this->registerGetValue("step_decay", this, &cvsd_encode_sb_i::get_step_decay);
    this->registerGetValue("accum_decay", this, &cvsd_encode_sb_i::get_accum_decay);
    this->registerGetValue("K", this, &cvsd_encode_sb_i::get_K);
    this->registerGetValue("J", this, &cvsd_encode_sb_i::get_J);
    this->registerGetValue("pos_accum_max", this, &cvsd_encode_sb_i::get_pos_accum_max);
    this->registerGetValue("neg_accum_max", this, &cvsd_encode_sb_i::get_neg_accum_max);

    // 
    // Use setThrottle method to enable the throttling of consumption/production of data by the
    // service function. The affect of the throttle will try to pause the execution of the 
    // service function for a specified duration.  This duration is calculated using the getTargetDuration() method
    // and the actual processing time to perform the work method.
    //
    // This is turned ON by default for "output" only components
    //
    // setThrottle( bool onoff );

    // 
    // Use maintainTimeStamp to enable proper data flow of timestamp with input and output data. 
    // if turned on (true)
    //  The timestamp from the input source will be used and maintained based on the output rate and
    //  the number of samples produced
    // if turned off
    //   then the timestamp from the input source is passed through if available or the time of day
    // 
    // maintainTimestamp( bool onoff );

    setupIOMappings();
}

void ofdm_frame_acquisition_i::createBlock()
{
    //
    // gr_sptr = xxx_make_xxx( args );
    //

    // Need to have at least 2 known symbols or else GR constructor goes into infinite loop because of unsigned arithmetic in for loop
    if (known_symbol.size() <= 2) {
        LOG_ERROR(ofdm_frame_acquisition_i, "'known_symbol' must contain at least 2 elements");
        return;
    }

    try {
        gr_sptr = digital_make_ofdm_frame_acquisition(occupied_carriers, fft_length, cplen, known_symbol, max_fft_shift_len);
    } catch (...) {
        return;
    }

    this->registerGetValue("snr", this, &ofdm_frame_acquisition_i::get_snr);

    // 
    // Use setThrottle method to enable the throttling of consumption/production of data by the
    // service function. The affect of the throttle will try to pause the execution of the 
    // service function for a specified duration.  This duration is calculated using the getTargetDuration() method
    // and the actual processing time to perform the work method.
    //
    // This is turned ON by default for "output" only components
    //
    // setThrottle( bool onoff );

    // 
    // Use maintainTimeStamp to enable proper data flow of timestamp with input and output data. 
    // if turned on (true)
    //  The timestamp from the input source will be used and maintained based on the output rate and
    //  the number of samples produced
    // if turned off
    //   then the timestamp from the input source is passed through if available or the time of day
    // 
    // maintainTimestamp( bool onoff );

    setupIOMappings();
}

void probe_avg_mag_sqrd_f_i::createBlock()
{
    //
    // gr_sptr = xxx_make_xxx( args );
    //

    try {
        gr_sptr = gr_make_probe_avg_mag_sqrd_f( this->threshold_db, this->alpha );
    } catch (...) {
        return;
    }

    this->registerGetterSetter("threshold_db", &gr_probe_avg_mag_sqrd_f::threshold, &gr_probe_avg_mag_sqrd_f::set_threshold);
    this->registerSetter("alpha", &gr_probe_avg_mag_sqrd_f::set_alpha);

    this->registerGetter("level", &gr_probe_avg_mag_sqrd_f::level);
    this->registerGetter("unmuted", &gr_probe_avg_mag_sqrd_f::unmuted);

    // 
    // Use setThrottle method to enable the throttling of consumption/production of data by the
    // service function. The affect of the throttle will try to pause the execution of the 
    // service function for a specified duration.  This duration is calculated using the getTargetDuration() method
    // and the actual processing time to perform the work method.
    //
    // This is turned ON by default for "output" only components
    //
    // setThrottle( bool onoff );

    // 
    // Use maintainTimeStamp to enable proper data flow of timestamp with input and output data. 
    // if turned on (true)
    //  The timestamp from the input source will be used and maintained based on the output rate and
    //  the number of samples produced
    // if turned off
    //   then the timestamp from the input source is passed through if available or the time of day
    // 
    // maintainTimestamp( bool onoff );

    setupIOMappings();
}

void pfb_synthesizer_ccf_5i_i::createBlock()
{
    //
    // gr_sptr = xxx_make_xxx( args );
    //

    try {
        gr_sptr = gr::filter::pfb_synthesizer_ccf::make( this->numchans, this->taps, this->twox );
    } catch (...) {
        return;
    }

    this->registerGetter("taps", &gr::filter::pfb_synthesizer_ccf::taps);
    this->registerSetter("taps", &gr::filter::pfb_synthesizer_ccf::set_taps);
    this->registerGetter("channel_map", &gr::filter::pfb_synthesizer_ccf::channel_map);
    this->registerSetter("channel_map", &gr::filter::pfb_synthesizer_ccf::set_channel_map);

    // 
    // Use setThrottle method to enable the throttling of consumption/production of data by the
    // service function. The affect of the throttle will try to pause the execution of the 
    // service function for a specified duration.  This duration is calculated using the getTargetDuration() method
    // and the actual processing time to perform the work method.
    //
    // This is turned ON by default for "output" only components
    //
    // setThrottle( bool onoff );

    // 
    // Use maintainTimeStamp to enable proper data flow of timestamp with input and output data. 
    // if turned on (true)
    //  The timestamp from the input source will be used and maintained based on the output rate and
    //  the number of samples produced
    // if turned off
    //   then the timestamp from the input source is passed through if available or the time of day
    // 
    // maintainTimestamp( bool onoff );

    setupIOMappings();
}

// createBlock
//
// Create the actual GNU Radio Block to that will perform the work method. The resulting
// block object is assigned to gr_stpr
//
// Add property change callbacks for getter/setter methods
//
void multiply_const_vss_i::createBlock()
{

  // need to have at least 1 item in the list... bad things happend if not
  if ( k.size() < 1 )  return;

  try {
    gr_sptr = gr_make_multiply_const_vss(k);
  }
  catch(...) {
    return;
  }

  // 
  // Use setThrottle method to enable the throttling of consumption/production of data by the
  // service function. The affect of the throttle will try to pause the execution of the 
  // service function for a specified duration.  This duration is calculated using the getTargetDuration() method
  // and the actual processing time to perform the work method.
  //
  // This is turned ON by default for "output" only components
  //
  // setThrottle( bool onoff );

  // 
  // Use maintainTimeStamp to enable proper data flow of timestamp with input and output data. 
  // if turned on (true)
  //  The timestamp from the input source will be used and maintained based on the output rate and
  //  the number of samples produced
  // if turned off
  //   then the timestamp from the input source is passed through if available or the time of day
  // 
  // maintainTimestamp( bool onoff );

   setupIOMappings();
   this->registerGetterSetter("k", &gr_multiply_const_vss::k, &gr_multiply_const_vss::set_k);

} 

void packed_to_unpacked_ss_i::createBlock()
{
    //
    // gr_sptr = xxx_make_xxx( args );
    //

    gr_endianness_t end;
    if (this->endianness == 1) {
        end = GR_MSB_FIRST;
    } else if (this->endianness == 2) {
        end = GR_LSB_FIRST;
    }

    gr_sptr = gr_make_packed_to_unpacked_ss(this->bits_per_chunk, end);

    // 
    // Use setThrottle method to enable the throttling of consumption/production of data by the
    // service function. The affect of the throttle will try to pause the execution of the 
    // service function for a specified duration.  This duration is calculated using the getTargetDuration() method
    // and the actual processing time to perform the work method.
    //
    // This is turned ON by default for "output" only components
    //
    // setThrottle( bool onoff );

    // 
    // Use maintainTimeStamp to enable proper data flow of timestamp with input and output data. 
    // if turned on (true)
    //  The timestamp from the input source will be used and maintained based on the output rate and
    //  the number of samples produced
    // if turned off
    //   then the timestamp from the input source is passed through if available or the time of day
    // 
    // maintainTimestamp( bool onoff );

    setupIOMappings();
}

// createBlock
//
// Create the actual GNU Radio Block to that will perform the work method. The resulting
// block object is assigned to gr_stpr
//
// Add property change callbacks for getter/setter methods
//
void sig_source_f_i::createBlock()
{
  //
  // gr_sptr = xxx_make_xxx( args );
  //
	this->gr_sptr = gr_make_sig_source_f( this->sampling_freq, (gr_waveform_t)this->waveform, this->frequency, this->amplitude, this->offset );
	this->setPropertyChangeListener("sampling_freq", this, &sig_source_f_i::changedSamplingFreq);
	this->registerGetter("sampling_freq", &gr_sig_source_f::sampling_freq);
	this->registerGetterSetter("waveform", &gr_sig_source_f::waveform, &gr_sig_source_f::set_waveform);
	this->registerGetterSetter("frequency", &gr_sig_source_f::frequency, &gr_sig_source_f::set_frequency);
	this->registerGetterSetter("amplitude", &gr_sig_source_f::amplitude, &gr_sig_source_f::set_amplitude);
	this->registerGetterSetter("offset", &gr_sig_source_f::offset, &gr_sig_source_f::set_offset);

  // 
  // Use setThrottle method to enable the throttling of consumption/production of data by the
  // service function. The affect of the throttle will try to pause the execution of the 
  // service function for a specified duration.  This duration is calculated using the getTargetDuration() method
  // and the actual processing time to perform the work method.
  //
  // This is turned ON by default for "output" only components
  //
  // setThrottle( bool onoff );

  // 
  // Use maintainTimeStamp to enable proper data flow of timestamp with input and output data. 
  // if turned on (true)
  //  The timestamp from the input source will be used and maintained based on the output rate and
  //  the number of samples produced
  // if turned off
  //   then the timestamp from the input source is passed through if available or the time of day
  // 
  // maintainTimestamp( bool onoff );

   setupIOMappings();

} 

void vector_sink_c_i::createBlock()
{
    //
    // gr_sptr = xxx_make_xxx( args );
    //
    try {
        gr_sptr = gr_make_vector_sink_c( this->vlen );
    } catch (...) {
        return;
    }

    this->registerGetter("data", &gr_vector_sink_c::data);
    this->setPropertyChangeListener("reset", this, &vector_sink_c_i::on_reset);

    //
    // Use setThrottle method to enable the throttling of consumption/production of data by the
    // service function. The affect of the throttle will try to pause the execution of the
    // service function for a specified duration.  This duration is calculated using the getTargetDuration() method
    // and the actual processing time to perform the work method.
    //
    // This is turned ON by default for "output" only components
    //
    // setThrottle( bool onoff );

    //
    // Use maintainTimeStamp to enable proper data flow of timestamp with input and output data.
    // if turned on (true)
    //  The timestamp from the input source will be used and maintained based on the output rate and
    //  the number of samples produced
    // if turned off
    //   then the timestamp from the input source is passed through if available or the time of day
    //
    // maintainTimestamp( bool onoff );

    setupIOMappings();
}

template <  typename IN_PORT_TYPE, typename OUT_PORT_TYPE > int copy_octet_base::_transformerServiceFunction( typename  std::vector< gr_istream< IN_PORT_TYPE > > &istreams ,
											typename  std::vector< gr_ostream< OUT_PORT_TYPE > > &ostreams  )
{
  typedef typename std::vector< gr_istream< IN_PORT_TYPE > >   _IStreamList;
  typedef typename std::vector< gr_ostream< OUT_PORT_TYPE > >  _OStreamList;

  boost::mutex::scoped_lock lock(serviceThreadLock);

  if ( validGRBlock() == false ) {

    // create our processing block, and setup  property notifiers
    createBlock();

    // create input/output port-stream mapping
    setupIOMappings();

    LOG_DEBUG( copy_octet_base, " FINISHED BUILDING  GNU RADIO BLOCK");
  }
 
  //process any Stream ID changes this could affect number of io streams
  processStreamIdChanges();

  if ( !validGRBlock() || istreams.size() == 0 || ostreams.size() == 0  ) {
    LOG_WARN(copy_octet_base, "NO STREAMS ATTACHED TO BLOCK..." );
    return NOOP;
  }

  _input_ready.resize( istreams.size() );
  _ninput_items_required.resize( istreams.size() );
  _ninput_items.resize( istreams.size() );
  _input_items.resize( istreams.size() );
  _output_items.resize( ostreams.size() );

  //
  // RESOLVE: need to look at forecast strategy, 
  //    1)  see how many read items are necessary for N number of outputs
  //    2)  read input data and see how much output we can produce
  //

  //
  // Grab available data from input streams
  //
  typename _OStreamList::iterator ostream;
  typename _IStreamList::iterator istream = istreams.begin();
  int nitems=0;
  for ( int idx=0 ; istream != istreams.end() && serviceThread->threadRunning() ; idx++, istream++ ) {
    // note this a blocking read that can cause deadlocks
    nitems = istream->read();
    
    if ( istream->overrun() ) {
      LOG_WARN( copy_octet_base, " NOT KEEPING UP WITH STREAM ID:" << istream->streamID );
    }

    if ( istream->sriChanged() ) {
      // RESOLVE - need to look at how SRI changes can affect Gnu Radio BLOCK state
      LOG_DEBUG( copy_octet_base, "SRI CHANGED, STREAMD IDX/ID: " 
               << idx << "/" << istream->pkt->streamID );
      setOutputStreamSRI( idx, istream->pkt->SRI );
    }

  }

  LOG_TRACE( copy_octet_base, "READ NITEMS: "  << nitems );
  if ( nitems <= 0 && !_istreams[0].eos() ) return NOOP;

  bool exitServiceFunction = false;
  bool eos = false;
  int  nout = 0;
  while ( nout > -1 && !exitServiceFunction && serviceThread->threadRunning() ) {

    eos = false;
    nout = _forecastAndProcess( eos, istreams, ostreams );
    if ( nout > -1  ) {

      // we chunked on data so move read pointer..
      istream = istreams.begin();
      for ( ; istream != istreams.end(); istream++ ) {
	int idx=std::distance( istreams.begin(), istream );
	// if we processed data for this stream
	if ( _input_ready[idx] ) {
	  size_t nitems = 0;
	  try {
	    nitems = gr_sptr->nitems_read( idx );
	  }
	  catch(...){}
      
	  istream->consume( nitems );
	  LOG_TRACE( copy_octet_base, " CONSUME READ DATA  ITEMS/REMAIN " << nitems << "/" << istream->nitems());
	}

      }
      gr_sptr->reset_read_index();
    }

    // check for not enough data return
    if ( nout == -1 ) {

      // check for  end of stream
      istream = istreams.begin();
      for ( ; istream != istreams.end() ; istream++) if ( istream->eos() ) eos=true;
//.........这里部分代码省略.........