Ensure we always have called Add() on the inflight counter before we Wait() for it

[wvanbergen]

@eapache This is not the cause of #449, but fixes another a potential issue when using the waitgroup.

[eapache]

I don't think this is necessary? If you create a wait group and then immediately Wait on it (without ever calling Add) it just returns right away...

[wvanbergen]

I think, this could potentially be an issue:

p := NewAsyncProducer(...)
p.Input() <- msg
p.Close()

if the Close call gets handled before the goroutine has read the first message from the input channel, the inflight counter is still 0.

[eapache]

Oh interesting, you're right that's a legit issue. However, I'm not sure your patch fixes all the cases :)

p := NewAsyncProducer(...)
p.Close()
p.Input() <- msg

With your patch, if the input message is handled before the waiting goroutine wakes up, then we will increment the counter from 0 before immediately decrementing it... so I think we must stop doing the "increment+decrement" trick for messages we return with the shutting-down error as well.

[wvanbergen]

I don't really understand this one? Close() doesn't return until the Wait() is finished?

[eapache]

1. Close() calls Wait()
2. Go switches contexts
3. the message is sent on Input(), is received, causes the wait counter to increment from 0 (the problem case) and decrement again since it is immediately returned as an error

[wvanbergen]

(I assume you means AsyncClose(), otherwise message on the input channel can never happen until the Wait() is finished.)

I don't really see what the problem is: after calling AsyncClose(), no message will ever make it into the processing logic, and we always wait for all the messages being processed to be completed. The worst that can happen is that we wait a bit too long, for msg to be returned as error?

[eapache]

I assume you means AsyncClose()

Ya, sorry

no message will ever make it into the processing logic

But it will still cause the wait counter to be Add(1)ed to, which is the bug

[eapache]

OK, this is an improvement regardless so I'll merge it; I'll also make a PR which should (I hope) fix and demonstrate the other case I'm thinking of.