Add Pull Consumer

[Jarema]

No description provided.

[caspervonb]

Solid base for iteration, first batch of comments.

[caspervonb]

Variants that use skip serialize, would probably be better off as overloads.

Suggested change

	pub no_wait: bool,
	pub no_wait: bool,

[Jarema]

I think this will become unnecesarily messy. Skipping serialization is basically a way of saving a few bytes over the wire, nothing else.

[caspervonb]

Would like to not have this options bag, and instead expose it as different overloads/iterators. Always nicer t call flat consumer.batch_<variant>() calls.

[Jarema]

Keep in mind we have different behaviors depending on passed values:

• expires > 0 + no wait: true -> waits for first message (even if not available) then returns No Messages as soon as there are no no messages for a Consumer
• expires : 0 + no wait: true -> does not wait for first messages. Returns No Messages whenever there are no messages
• expires: whatever, no_wait: false -> waits until expires or fulfills batch.

Add to that max_bytes and heartbeat ... and we would end up with A LOT of variants. To many to have ergonomic API. Keep in mind this is not only an option bag, but an actual serialized struct as a json going via the API that we don't want to expose.

I would instead create low-level API that exposes those behaviors. But that is A LOT of overloads if we don't have arguments passed as Option or config bag or builder.

[caspervonb]

Keep in mind we have different behaviors depending on passed values:

Sounds like an argument FOR named variants that give us good documentation blocks? no waits sounds like a try variant for instance.

[Jarema]

We already have two variants - fetch and process (one with no wait true, on with false.).
We're missing just the third variant and you just found a nice name for it.

We could actually hide BatchConfig from the user, but we said that the low level API should be enabled for power users.

The only question is how to enable passing hearbet/max_bytes/fetch to those 3 methods.

[Jarema]

I would have first pass with current approach and extend while writing docs as a follow up PR.
@caspervonb WDYT?

[caspervonb]

Hmm, shared state? can't this live on Batch?

Suggested change

	pub(crate) subscription: Option<Subscriber>,
	pub(crate) subscription: Option<Subscriber>,

[Jarema]

It should not. Creating new subscription for every batch is pretty bad for performance.

We can though switch to such approach as soon as muxing request subject is possible for Pull Consumers.
Would remove state and be more even more performant.

[caspervonb]

Wouldn't be for every batch, this was partially why I proposed having iterators of iterators: to have somewhere that is not the consumer that can have proper mutability semantics while owning the state, performing the requests as needed etc.

Right now consumer acts like its an iterator semantically, it's just not implementing the traits. You'll run into borrow issues going this way too.

Quick abstract, the basic lowest level layer can semantically be something along the lines of the following:

// A fetch operation here is an iterator fetches batches (naming is hard).
pub struct Fetch {
  // owns the subscriber
  subscriber: Subscriber,
  // other relevant state.
}

impl Stream for Fetch {
    // Result to deal with errors coming from requests
    type Item = Result<Batch, Error>;
}

pub struct Batch {
  // take a reference etc, if you need it.
  subscriber: &'a Subscriber,
}

impl Stream for Batch {
  // Can't fail as far as I'm aware at this point, so no result.
  type Item = Message;

  fn poll_next(
         mut self: pin::Pin<&mut Self>,
         cx: &mut task::Context<'_>,
  ) -> task::Poll<Option<Self::Item>> {
    // Pull from the batch as needed.
  }
}

impl Consumer {
  pub fn fetch() -> Result<Fetch, Error> {
    Ok(Fetch {})
  }

  fn poll_next(
         mut self: pin::Pin<&mut Self>,
         cx: &mut task::Context<'_>,
  ) -> task::Poll<Option<Self::Item>> {
    // Dispatch requests as needed etc.
  }
}

fn main() {
  let consumer = ...;
  
  // Built-in combinators can flatten this again.  
  for batch in consumer.fetch().next().await? {
    for message in batch.next().await? {
      println!("{:?}", message);
    }
  }

[Jarema]

Just for clarity: we seem to find reasonable compromise with current implementation.

[Jarema]

simplified process