Function NewStreamPipe

internal/stream/pipe.go:64–98

class function which creates a StreamPipe with default configuration.

(
	ctx context.Context,
	client *gotgproto.Client,
	location tg.InputFileLocationClass,
	start, end int64,
	log *zap.Logger,
)

// class function which creates a StreamPipe with default configuration.
func NewStreamPipe(
	ctx context.Context,
	client *gotgproto.Client,
	location tg.InputFileLocationClass,
	start, end int64,
	log *zap.Logger,
) (io.ReadCloser, error) {

	if start > end {
		return nil, fmt.Errorf("invalid range: start (%d) > end (%d)", start, end)
	}

	ctx, cancel := context.WithCancel(ctx)

	totalBytes := end - start + 1
	blockSize := calculateBlockSize(start, end)

	p := &StreamPipe{
		ctx:        ctx,
		cancel:     cancel,
		log:        log.Named("StreamPipe"),
		client:     client,
		location:   location,
		start:      start,
		end:        end,
		blockSize:  blockSize,
		totalBytes: totalBytes,
		blockQueue: make(chan []byte, config.ValueOf.StreamBufferCount),
	}

	// start prefetching in background
	go p.prefetch()

	return p, nil
}

// Read implements io.Reader
func (p *StreamPipe) Read(buf []byte) (n int, err error) {