pub fn fetch_messages<'a, I, J>(&mut self, input: I) -> Result<Vec<fetch::Response>>
        where J: AsRef<FetchPartition<'a>>,
              I: IntoIterator<Item = J>
    {
        let state = &mut self.state;
        let config = &self.config;

        let correlation = state.next_correlation_id();

        // Map topic and partition to the corresponding broker
        let mut reqs: HashMap<&str, protocol::FetchRequest> = HashMap::new();
        for inp in input {
            let inp = inp.as_ref();
            if let Some(broker) = state.find_broker(inp.topic, inp.partition) {
                reqs.entry(broker)
                    .or_insert_with(|| {
                        protocol::FetchRequest::new(correlation,
                                                    &config.client_id,
                                                    config.fetch_max_wait_time,
                                                    config.fetch_min_bytes)
                    })
                    .add(inp.topic,
                         inp.partition,
                         inp.offset,
                         if inp.max_bytes > 0 {
                             inp.max_bytes
                         } else {
                             config.fetch_max_bytes_per_partition
                         });
            }
        }

        __fetch_messages(&mut self.conn_pool, config, reqs)
    }
/// Fetch messages from Kafka (multiple topic, partitions).
///
/// It takes a vector specifying the topic partitions and their
/// offsets as of which to fetch messages.  Additionally, the
/// default "max fetch size per partition" can be explicitely
/// overriden if it is "defined" - this is, if `max_bytes` is
/// greater than zero.
///
/// The result is exposed in a raw, complicated manner but allows
/// for very efficient consumption possibilities. All of the data
/// available through the returned fetch responses is bound to
/// their lifetime as that data is merely a "view" into parts of
/// the response structs.  If you need to keep individual messages
/// for a longer time than the whole fetch responses, you'll need
/// to make a copy of the message data.
///
/// * This method transparently uncompresses messages (while Kafka
/// might sent them in compressed format.)
///
/// * This method ensures to skip messages with a lower offset
/// than requested (while Kafka might for efficiency reasons sent
/// messages with a lower offset.)
///
/// Note: before using this method consider using
/// `kafka::consumer::Consumer` instead which provides an easier
/// to use API for the regular use-case of fetching messesage from
/// Kafka.
///
/// # Example
///
/// This example demonstrates iterating all fetched messages from
/// two topic partitions.  From one partition we allow Kafka to
/// deliver to us the default number bytes as defined by
/// `KafkaClient::set_fetch_max_bytes_per_partition`, from the
/// other partition we allow Kafka to deliver up to 1MiB of
/// messages.
///
/// ```no_run
/// use kafka::client::{KafkaClient, FetchPartition};
///
/// let mut client = KafkaClient::new(vec!("localhost:9092".to_owned()));
/// client.load_metadata_all().unwrap();
/// let reqs = &[FetchPartition::new("my-topic", 0, 0),
///              FetchPartition::new("my-topic-2", 0, 0).with_max_bytes(1024*1024)];
/// let resps = client.fetch_messages(reqs).unwrap();
/// for resp in resps {
///   for t in resp.topics() {
///     for p in t.partitions() {
///       match p.data() {
///         &Err(ref e) => {
///           println!("partition error: {}:{}: {}", t.topic(), p.partition(), e)
///         }
///         &Ok(ref data) => {
///           println!("topic: {} / partition: {} / latest available message offset: {}",
///                    t.topic(), p.partition(), data.highwatermark_offset());
///           for msg in data.messages() {
///             println!("topic: {} / partition: {} / message.offset: {} / message.len: {}",
///                      t.topic(), p.partition(), msg.offset, msg.value.len());
///           }
///         }
///       }
///     }
///   }
/// }
/// ```
/// See also `kafka::consumer`.
/// See also `KafkaClient::set_fetch_max_bytes_per_partition`.
Fetch messages from Kafka (multiple topic, partitions).
It takes a vector specifying the topic partitions and their
offsets as of which to fetch messages.  Additionally, the
default "max fetch size per partition" can be explicitely
overriden if it is "defined" - this is, if `max_bytes` is
greater than zero.

The result is exposed in a raw, complicated manner but allows
for very efficient consumption possibilities. All of the data
available through the returned fetch responses is bound to
their lifetime as that data is merely a "view" into parts of
the response structs.  If you need to keep individual messages
for a longer time than the whole fetch responses, you'll need
to make a copy of the message data.

This method transparently uncompresses messages (while Kafka
might sent them in compressed format.)

This method ensures to skip messages with a lower offset
than requested (while Kafka might for efficiency reasons sent
messages with a lower offset.)

before using this method consider using
`kafka::consumer::Consumer` instead which provides an easier
to use API for the regular use-case of fetching messesage from
Kafka.
Fetch messages from Kafka ( multiple topic partitions ) . It takes a vector specifying the topic partitions and their offsets as of which to fetch messages . Additionally the default " max fetch size per partition " can be explicitely overriden if it is " defined " - this is if ` max_bytes ` is greater than zero . The result is exposed in a raw complicated manner but allows for very efficient consumption possibilities . All of the data available through the returned fetch responses is bound to their lifetime as that data is merely a " view " into parts of the response structs . If you need to keep individual messages for a longer time than the whole fetch responses you ' ll need to make a copy of the message data . This method transparently uncompresses messages ( while Kafka might sent them in compressed format . ) This method ensures to skip messages with a lower offset than requested ( while Kafka might for efficiency reasons sent messages with a lower offset . ) before using this method consider using ` kafka :: consumer :: Consumer ` instead which provides an easier to use API for the regular use - case of fetching messesage from Kafka .
pub fn fetch_messages<'a, I, J>(&mut self, input: I) -> Result<Vec<fetch::Response>>
        where J: AsRef<FetchPartition<'a>>,
              I: IntoIterator<Item = J>
    {
        let state = &mut self.state;
        let config = &self.config;
        let correlation = state.next_correlation_id();
        // Map topic and partition to the corresponding broker
        let mut reqs: HashMap<&str, protocol::FetchRequest> = HashMap::new();
        for inp in input {
            let inp = inp.as_ref();
            if let Some(broker) = state.find_broker(inp.topic, inp.partition) {
                reqs.entry(broker)
                    .or_insert_with(|| {
                        protocol::FetchRequest::new(correlation,
                                                    &config.client_id,
                                                    config.fetch_max_wait_time,
                                                    config.fetch_min_bytes)
                    })
                    .add(inp.topic,
                         inp.partition,
                         inp.offset,
                         if inp.max_bytes > 0 {
                             inp.max_bytes
                         } else {
                             config.fetch_max_bytes_per_partition
                         });
            }
        }
        __fetch_messages(&mut self.conn_pool, config, reqs)
    }
pub fn fetch_messages < ' a , I , J > ( & mut self , input : I ) -> Result < Vec < fetch :: Response > > where J : AsRef < FetchPartition < ' a > > , I : IntoIterator < Item = J > { let state = & mut self . state ; let config = & self . config ; let correlation = state . next_correlation_id ( ) ; // Map topic and partition to the corresponding broker let mut reqs : HashMap < & str , protocol :: FetchRequest > = HashMap :: new ( ) ; for inp in input { let inp = inp . as_ref ( ) ; if let Some ( broker ) = state . find_broker ( inp . topic , inp . partition ) { reqs . entry ( broker ) . or_insert_with ( | | { protocol :: FetchRequest :: new ( correlation , & config . client_id , config . fetch_max_wait_time , config . fetch_min_bytes ) } ) . add ( inp . topic , inp . partition , inp . offset , if inp . max_bytes > 0 { inp . max_bytes } else { config . fetch_max_bytes_per_partition } ) ; } } __fetch_messages ( & mut self . conn_pool , config , reqs ) }
Fetch messages from Kafka (multiple topic, partitions).
Fetch messages from Kafka ( multiple topic partitions ) .
&mut self
input
I
example
This example demonstrates iterating all fetched messages from
two topic partitions.  From one partition we allow Kafka to
deliver to us the default number bytes as defined by
`KafkaClient::set_fetch_max_bytes_per_partition`, from the
other partition we allow Kafka to deliver up to 1MiB of
messages.

```no_run
use kafka::client::{KafkaClient, FetchPartition}.

let mut client = KafkaClient::new(vec!("localhost:9092".to_owned()));
client.load_metadata_all().unwrap();
let reqs = &[FetchPartition::new("my-topic", 0, 0),
FetchPartition::new("my-topic-2", 0, 0).with_max_bytes(1024*1024)];
let resps = client.fetch_messages(reqs).unwrap();
for resp in resps {
for t in resp.topics() {
for p in t.partitions() {
match p.data() {
&Err(ref e) => {
println!("partition error: {}:{}: {}", t.topic(), p.partition(), e)
}
&Ok(ref data) => {
println!("topic: {} / partition: {} / latest available message offset: {}",
t.topic(), p.partition(), data.highwatermark_offset());
for msg in data.messages() {
println!("topic: {} / partition: {} / message.offset: {} / message.len: {}",
t.topic(), p.partition(), msg.offset, msg.value.len());
}
}
}
}
}
}
```
See also `kafka::consumer`.
See also `KafkaClient::set_fetch_max_bytes_per_partition`.
This example demonstrates iterating all fetched messages from two topic partitions . From one partition we allow Kafka to deliver to us the default number bytes as defined by ` KafkaClient :: set_fetch_max_bytes_per_partition ` from the other partition we allow Kafka to deliver up to 1MiB of messages . ` ` ` no_run use kafka :: client :: { KafkaClient FetchPartition } . let mut client = KafkaClient :: new ( vec! ( " localhost : 9092 " . to_owned () )) ; client . load_metadata_all () . unwrap () ; let reqs = & [ FetchPartition :: new ( " my - topic " 0 0 ) FetchPartition :: new ( " my - topic - 2 " 0 0 ) . with_max_bytes ( 1024 * 1024 ) ] ; let resps = client . fetch_messages ( reqs ) . unwrap () ; for resp in resps { for t in resp . topics () { for p in t . partitions () { match p . data () { &Err ( ref e ) = > { println! ( " partition error : {} : {} : {} " t . topic () p . partition () e ) } &Ok ( ref data ) = > { println! ( " topic : {} / partition : {} / latest available message offset : {} " t . topic () p . partition () data . highwatermark_offset () ) ; for msg in data . messages () { println! ( " topic : {} / partition : {} / message . offset : {} / message . len : {} " t . topic () p . partition () msg . offset msg . value . len () ) ; } } } } } } ` ` ` See also ` kafka :: consumer ` . See also ` KafkaClient :: set_fetch_max_bytes_per_partition ` .