pub fn fetch_offsets<T: AsRef<str>>(&mut self,
                                        topics: &[T],
                                        offset: FetchOffset)
                                        -> Result<HashMap<String, Vec<PartitionOffset>>> {
        let time = offset.to_kafka_value();
        let n_topics = topics.len();

        let state = &mut self.state;
        let correlation = state.next_correlation_id();

        // Map topic and partition to the corresponding broker
        let config = &self.config;
        let mut reqs: HashMap<&str, protocol::OffsetRequest> = HashMap::with_capacity(n_topics);
        for topic in topics {
            let topic = topic.as_ref();
            if let Some(ps) = state.partitions_for(topic) {
                for (id, host) in ps.iter()
                    .filter_map(|(id, p)| p.broker(&state).map(|b| (id, b.host()))) {
                    let entry = reqs.entry(host)
                        .or_insert_with(|| {
                            protocol::OffsetRequest::new(correlation, &config.client_id)
                        });
                    entry.add(topic, id, time);
                }
            }
        }

        // Call each broker with the request formed earlier
        let now = Instant::now();
        let mut res: HashMap<String, Vec<PartitionOffset>> = HashMap::with_capacity(n_topics);
        for (host, req) in reqs {
            let resp = try!(__send_receive::<_, protocol::OffsetResponse>(&mut self.conn_pool,
                                                                          &host,
                                                                          now,
                                                                          req));
            for tp in resp.topic_partitions {
                let e = res.entry(tp.topic).or_insert(vec![]);
                for p in tp.partitions {
                    e.push(p.into_offset());
                }
            }
        }
        Ok(res)
    }
/// Fetch offsets for a list of topics
///
/// # Examples
///
/// ```no_run
/// use kafka::client::KafkaClient;
///
/// let mut client = KafkaClient::new(vec!["localhost:9092".to_owned()]);
/// client.load_metadata_all().unwrap();
/// let topics: Vec<String> = client.topics().names().map(ToOwned::to_owned).collect();
/// let offsets = client.fetch_offsets(&topics, kafka::client::FetchOffset::Latest).unwrap();
/// ```
///
/// Returns a mapping of topic name to `PartitionOffset`s for each
/// currently available partition of the corresponding topic.
Fetch offsets for a list of topics
Fetch offsets for a list of topics
pub fn fetch_offsets<T: AsRef<str>>(&mut self,
                                        topics: &[T],
                                        offset: FetchOffset)
                                        -> Result<HashMap<String, Vec<PartitionOffset>>> {
        let time = offset.to_kafka_value();
        let n_topics = topics.len();
        let state = &mut self.state;
        let correlation = state.next_correlation_id();
        // Map topic and partition to the corresponding broker
        let config = &self.config;
        let mut reqs: HashMap<&str, protocol::OffsetRequest> = HashMap::with_capacity(n_topics);
        for topic in topics {
            let topic = topic.as_ref();
            if let Some(ps) = state.partitions_for(topic) {
                for (id, host) in ps.iter()
                    .filter_map(|(id, p)| p.broker(&state).map(|b| (id, b.host()))) {
                    let entry = reqs.entry(host)
                        .or_insert_with(|| {
                            protocol::OffsetRequest::new(correlation, &config.client_id)
                        });
                    entry.add(topic, id, time);
                }
            }
        }
        // Call each broker with the request formed earlier
        let now = Instant::now();
        let mut res: HashMap<String, Vec<PartitionOffset>> = HashMap::with_capacity(n_topics);
        for (host, req) in reqs {
            let resp = try!(__send_receive::<_, protocol::OffsetResponse>(&mut self.conn_pool,
                                                                          &host,
                                                                          now,
                                                                          req));
            for tp in resp.topic_partitions {
                let e = res.entry(tp.topic).or_insert(vec![]);
                for p in tp.partitions {
                    e.push(p.into_offset());
                }
            }
        }
        Ok(res)
    }
pub fn fetch_offsets < T : AsRef < str > > ( & mut self , topics : & [ T ] , offset : FetchOffset ) -> Result < HashMap < String , Vec < PartitionOffset > > > { let time = offset . to_kafka_value ( ) ; let n_topics = topics . len ( ) ; let state = & mut self . state ; let correlation = state . next_correlation_id ( ) ; // Map topic and partition to the corresponding broker let config = & self . config ; let mut reqs : HashMap < & str , protocol :: OffsetRequest > = HashMap :: with_capacity ( n_topics ) ; for topic in topics { let topic = topic . as_ref ( ) ; if let Some ( ps ) = state . partitions_for ( topic ) { for ( id , host ) in ps . iter ( ) . filter_map ( | ( id , p ) | p . broker ( & state ) . map ( | b | ( id , b . host ( ) ) ) ) { let entry = reqs . entry ( host ) . or_insert_with ( | | { protocol :: OffsetRequest :: new ( correlation , & config . client_id ) } ) ; entry . add ( topic , id , time ) ; } } } // Call each broker with the request formed earlier let now = Instant :: now ( ) ; let mut res : HashMap < String , Vec < PartitionOffset > > = HashMap :: with_capacity ( n_topics ) ; for ( host , req ) in reqs { let resp = try ! ( __send_receive protocol OffsetResponse ( mut self conn_pool host now req ) ) ; for tp in resp . topic_partitions { let e = res . entry ( tp . topic ) . or_insert ( vec ! [ ] ) ; for p in tp . partitions { e . push ( p . into_offset ( ) ) ; } } } Ok ( res ) }
Fetch offsets for a list of topics
Fetch offsets for a list of topics
&mut self
topics
&[T]
offset
FetchOffset
examples
```no_run
use kafka::client::KafkaClient.

let mut client = KafkaClient::new(vec!["localhost:9092".to_owned()]);
client.load_metadata_all().unwrap();
let topics: Vec = client.topics().names().map(ToOwned::to_owned).collect();
let offsets = client.fetch_offsets(&topics, kafka::client::FetchOffset::Latest).unwrap();
```

Returns a mapping of topic name to `PartitionOffset`s for each
currently available partition of the corresponding topic.
` ` ` no_run use kafka :: client :: KafkaClient . let mut client = KafkaClient :: new ( vec! [ " localhost : 9092 " . to_owned () ] ) ; client . load_metadata_all () . unwrap () ; let topics : Vec = client . topics () . names () . map ( ToOwned :: to_owned ) . collect () ; let offsets = client . fetch_offsets ( &topics kafka :: client :: FetchOffset :: Latest ) . unwrap () ; ` ` ` Returns a mapping of topic name to ` PartitionOffset ` s for each currently available partition of the corresponding topic .