Method parseFileTyped

input/inpututils/compressedstream.go:255–369

(fn string, comp compressionType)

}

func (s *CompressedInput) parseFileTyped(fn string, comp compressionType) {

	ctx := log.WithFields(log.Fields{"f": "compressedInput.parseFile", "fn": fn})
	stream, sz, lastModified, url, err := s.Opener(fn)

	stream = s.stats.NewStatsReader(stream, sz)
	if err != nil {
		log.WithFields(log.Fields{"f": "compressedInput.parseFile", "fn": fn}).WithError(err).Error("Error while opening stream")
		return
	}
	defer stream.Close()

	var r io.Reader

	switch comp {
	case gzipCompression:
		if sz > 1000000 {
			rgz, err := newFastGzReader(stream)
			if err != nil {
				// Sometimes the fast gz reader fails to initialize due to
				// memory pressure. We'd still like to run so try the
				// slower (and less memory hungry) gzip.
				ctx.WithError(err).Error("error initializing fast gzip, will attempt slow gzip")
				r, err = gzip.NewReader(stream)
				if err != nil {
					ctx.WithError(err).Fatal("both fast and slow gzip readers failed to initialize")
					return
				}
			} else {
				defer rgz.Close()
				r = rgz
			}
		} else {
			rgz, err := gzip.NewReader(stream)
			if err != nil {
				ctx.WithError(err).Fatal("error initializing gzip")
				return
			}
			defer rgz.Close()
			r = rgz
		}
	case zstdCompression:
		rzst := zstd.NewReader(stream)
		defer rzst.Release()
		r = rzst
	default:
		ctx.WithError(err).Fatal("Unknown compression type specified.")
	}

	ctx.Info("begin reading")

	rbuf := bufio.NewReaderSize(r, kChunkBuffer)

	for atomic.LoadInt64(&s.stopping) == 0 {
		bakerData := s.pool.Get().(*baker.Data)
		bakerData.Meta = baker.Metadata{
			MetadataLastModified: lastModified,
			MetadataURL:          url,