reduce duplicate rdy update requests

[andyxning]

This PR will reduct duplicate rdy update requests.

[andyxning]

/cc @ploxiln @mreiferson

[andyxning]

/cc @ploxiln @mreiferson

[ploxiln]

Yeah, seems safe, I might just merge it, since you at least will use it plenty. But I would like to get some input from the other maintainers.

By the way, did you consider this alternative:

--- a/consumer.go
+++ b/consumer.go
@@ -975,7 +975,7 @@ func (r *Consumer) updateRDY(c *Conn, count int64) error {
 }

 func (r *Consumer) sendRDY(c *Conn, count int64) error {
-       if count == 0 && c.LastRDY() == 0 {
+       if count == c.LastRDY() {
                // no need to send. It's already that RDY count
                return nil
        }

[andyxning]

@ploxiln Thanks for the suggestion. The key point of implementing using the code in the PR is that if the count is not zero, then SetRDY will update lastRdyTimestamp. Since rdy manage is tricky, i think we should not neglect it.

[ploxiln]

Yeah, fair.

Some observations:

• Conn.RDY() and Conn.LastRDY() are the exact same
• Conn.lastRdyTimestamp / Conn.LastRdyTime() are only used in Consumer.redistributeRDY() to check Config.LowRdyTimeout

So this should be fine. Maybe later someone will come up with a good refactor of the whole RDY distribution algorithm (e.g. to address #277).