func (p *filePipe) Config() *config.PipeConfig {
	return &p.cfg
}
Config returns pipe configuration
package pipe
import (
	"bufio"
	"bytes"
	"compress/gzip"
	"crypto"
	"crypto/sha256"
	"database/sql"
	"encoding/binary"
	"encoding/json"
	"fmt"
	"hash"
	"io"
	"io/ioutil"
	"os"
	"path/filepath"
	"sort"
	"strings"
	"sync/atomic"
	"time"
	"golang.org/x/crypto/openpgp"
	"golang.org/x/crypto/openpgp/armor"
	"golang.org/x/crypto/openpgp/packet"  
	"golang.org/x/sys/unix"
	"github.com/fsnotify/fsnotify"
	"github.com/uber/storagetapper/config"
	"github.com/uber/storagetapper/log"
	"github.com/uber/storagetapper/metrics"
)
const (
	dirPerm        = 0775
	delimiter byte = '\n'
)
var signKeyPw = ""
var privateKeyPw = ""
type flushWriteCloser interface {
	Write([]byte) (int, error)
	Flush() error
	Close() error
}
type fs interface {
	MkdirAll(path string, perm os.FileMode) error
	Rename(oldpath, newpath string) error
	ReadDir(dirname string, listFrom string) ([]os.FileInfo, error)
	OpenRead(name string, offset int64) (io.ReadCloser, error)
	OpenWrite(name string) (flushWriteCloser, io.Seeker, error)
	Remove(name string) error
	Cancel(io.Closer) error
}
type filePipe struct {
	datadir string
	cfg     config.PipeConfig
}
type file struct {
	name   string
	key    string
	file   io.WriteCloser  
	seek   io.Seeker
	hash   hash.Hash
	offset int64
	nRecs  int64
	writer flushWriteCloser
	prev *file
	next *file
	compressedSize int64
}
type stat struct {
	NumRecs  int64
	Hash     string
	FileName string
}
type fileProducer struct {
	*filePipe
	header Header
	topic  string
	files  map[string]*file
	ffirst *file
	flast  *file
	seqno  int
	fs   fs
	text int64  
	metrics *metrics.FilePipeMetrics
	stats map[string]*stat
}
type fileConsumer struct {
	*filePipe
	baseConsumer
	topic  string
	file   io.ReadCloser
	name   string
	reader *bufio.Reader
	header Header
	fs     fs
	text   int64  
	msg []byte
	err error
	metrics *metrics.FilePipeMetrics
	pgpMD   *openpgp.MessageDetails
	watcher *fsnotify.Watcher
	offset int64
}
type noopFlusher struct {
	io.WriteCloser
}
func (f *noopFlusher) Flush() error {
	return nil
}
type flushClose struct {
	*bufio.Writer
}
func (f *flushClose) Close() error {
	return f.Writer.Flush()
}
type hashWriter struct {
	flushWriteCloser
	hash    hash.Hash
	metrics *metrics.FilePipeMetrics
	f       *file
}
func (w *hashWriter) Write(b []byte) (int, error) {
	n, err := w.flushWriteCloser.Write(b)
	if w.hash != nil {
		_, _ = w.hash.Write(b)
	}
	w.metrics.BytesWritten.Inc(int64(n))
	if w.f != nil {
		w.f.compressedSize += int64(n)
	}
	return n, err
}
type chainer struct {
	w1 flushWriteCloser
	w2 flushWriteCloser
}
func (c *chainer) Write(b []byte) (int, error) {
	return c.w1.Write(b)  
}
func (c *chainer) Close() error {
	err1 := c.w1.Close()
	err2 := c.w2.Close()
	if err1 != nil {
		return err1
	}
	return err2
}
func (c *chainer) Flush() error {
	if err := c.w1.Flush(); err != nil {
		return err
	}
	return c.w2.Flush()
}
func init() {
	registerPlugin("file", initFilePipe)
}
func initFilePipe(cfg *config.PipeConfig, db *sql.DB) (Pipe, error) {
	return &filePipe{cfg.BaseDir, *cfg}, nil
}
func (p *filePipe) Type() string {
	return "file"
}