Add recvClose channel to stream

[dmcgowan]

Prevent panic from closing recv channel, which may be written to after close. Use a separate channel to signal recv has closed and check that channel on read and write.

Alternative to #139 for containerd/containerd#8390

[dmcgowan]

Had to add additional

	select {
	case <-s.recvClose:
		return s.recvErr
	default:
	}

on receive to replace the err nil check. Since Go does not guarantee select order priority, messages could still be sent after a close was called, causing that test flake. If the receive and close happen at the same time, it isn't important which one wins.

[henry118]

LGTM

[Iceber]

Is it necessary to ensure that messages that have been sent to recv are processed?

var msg *streamMessage

select {
...
case <- cs.s.recvClose:
    select {
        case msg <- cs.s.recv:
        default:
            return cs.s.recvErr
    }
case msg <- cs.s.recv:
}
if msg.header.Type == messageTypeResponse {
...

Because there is a check before receiving data, s.recv will not always have data

	select {
	case <-s.recvClose:
		return s.recvErr
	default:
	}

[cpuguy83]

Hmm, good catch.
I think it is a good idea to drain/handle appropriately since this is a buffered channel and the client thinks it has been sent.
At worst we should have only 1 message to process, currently.

[dmcgowan]

Updated, that is a good change. Wish select could have ordering which would make code much cleaner

[cpuguy83]

LGTM

[fuweid]

LGTM