Function SendStream

core/transfer/streaming/stream.go:45–141

(ctx context.Context, r io.Reader, stream streaming.Stream)

}

func SendStream(ctx context.Context, r io.Reader, stream streaming.Stream) {
	window := make(chan int32)
	go func() {
		defer close(window)
		for {
			select {
			case <-ctx.Done():
				return
			default:
			}

			anyType, err := stream.Recv()
			if err != nil {
				if !errors.Is(err, io.EOF) && !errors.Is(err, context.Canceled) {
					log.G(ctx).WithError(err).Error("send stream ended without EOF")
				}
				return
			}
			i, err := typeurl.UnmarshalAny(anyType)
			if err != nil {
				log.G(ctx).WithError(err).Error("failed to unmarshal stream object")
				continue
			}
			switch v := i.(type) {
			case *transferapi.WindowUpdate:
				select {
				case <-ctx.Done():
					return
				case window <- v.Update:
				}
			default:
				log.G(ctx).Errorf("unexpected stream object of type %T", i)
			}
		}
	}()
	go func() {
		defer stream.Close()

		buf := bufPool.Get().(*[]byte)
		defer bufPool.Put(buf)

		var remaining int32

		for {
			if remaining > 0 {
				// Don't wait for window update since there are remaining
				select {
				case <-ctx.Done():
					// TODO: Send error message on stream before close to allow remote side to return error
					return
				case update := <-window:
					remaining += update
				default:
				}
			} else {
				// Block until window updated
				select {
				case <-ctx.Done():