package runtimeapi

import "context"

// clientStreamReceivePump keeps exactly one Recv in flight and exposes its
// completion to the handler's coordinator. Engine-backed client streams must
// observe both sides concurrently: an engine may reject the request before it
// reads the body, while the client may legitimately still be waiting to send.
func clientStreamReceivePump[T any](ctx context.Context, receive func() (*T, error)) <-chan clientStreamReceive[T] {
	received := make(chan clientStreamReceive[T], 1)
	go func() {
		for {
			message, err := receive()
			select {
			case received <- clientStreamReceive[T]{message: message, err: err}:
			case <-ctx.Done():
				return
			}
			if err != nil || ctx.Err() != nil {
				return
			}
		}
	}()
	return received
}

type clientStreamReceive[T any] struct {
	message *T
	err     error
}