Function ReceiveStream

core/transfer/streaming/stream.go:143–205

(ctx context.Context, stream streaming.Stream)

}

func ReceiveStream(ctx context.Context, stream streaming.Stream) io.Reader {
	r, w := io.Pipe()
	go func() {
		defer stream.Close()
		var window int32
		for {
			var werr error
			if window < windowSize {
				update := &transferapi.WindowUpdate{
					Update: windowSize,
				}
				anyType, err := typeurl.MarshalAny(update)
				if err != nil {
					w.CloseWithError(fmt.Errorf("failed to marshal window update: %w", err))
					return
				}
				// check window update error after recv, stream may be complete
				if werr = stream.Send(anyType); werr == nil {
					window += windowSize
				} else if errors.Is(werr, io.EOF) {
					// TODO: Why does send return EOF here
					werr = nil
				}
			}
			anyType, err := stream.Recv()
			if err != nil {
				if errors.Is(err, io.EOF) || errors.Is(err, context.Canceled) {
					err = nil
				} else {
					err = fmt.Errorf("received failed: %w", err)
				}
				w.CloseWithError(err)
				return
			} else if werr != nil {
				// Try receive before erroring out
				w.CloseWithError(fmt.Errorf("failed to send window update: %w", werr))
				return
			}
			i, err := typeurl.UnmarshalAny(anyType)
			if err != nil {
				w.CloseWithError(fmt.Errorf("failed to unmarshal received object: %w", err))
				return
			}
			switch v := i.(type) {
			case *transferapi.Data:
				n, err := w.Write(v.Data)
				if err != nil {
					w.CloseWithError(fmt.Errorf("failed to unmarshal received object: %w", err))
					// Close will error out sender
					return
				}
				window = window - int32(n)
			// TODO: Handle error case
			default:
				log.G(ctx).Warnf("Ignoring unknown stream object of type %T", i)
				continue
			}
		}