func (p *fileProducer) PushBatch(key string, in interface{}) error {
	return p.push(key, in, true)
}
PushBatch stashes a keyed message into batch which will be send to File by
PushBatchCommit
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"
}
func (p *filePipe) Config() *config.PipeConfig {
	return &p.cfg
}
func (p *filePipe) Close() error {
	return nil
}
func (p *filePipe) NewProducer(topic string) (Producer, error) {
	m := metrics.NewFilePipeMetrics("pipe_producer", map[string]string{"topic": topic, "pipeType": "file"})
	return &fileProducer{filePipe: p, topic: topic, files: make(map[string]*file), fs: &fileFS{}, metrics: m, stats: make(map[string]*stat)}, nil
}
func (p *filePipe) initConsumer(c *fileConsumer, fn fetchFunc) (Consumer, error) {
	fname, offset, err := c.seek(c.topic, InitialOffset)
	if log.E(err) {
		return nil, err
	}
	if fname != "" {
		if strings.HasSuffix(fname, ".open") {
			c.offset = offset
		} else {
			c.openFile(fname, offset)
		}
	}
	c.initBaseConsumer(fn)
	return c, nil
}
func (p *filePipe) NewConsumer(topic string) (Consumer, error) {
	w, err := fsnotify.NewWatcher()
	if err != nil {
		return nil, err
	}
	m := metrics.NewFilePipeMetrics("pipe_consumer", map[string]string{"topic": topic, "pipeType": "file"})
	c := &fileConsumer{filePipe: p, topic: topic, fs: &fileFS{}, metrics: m, watcher: w}
	return p.initConsumer(c, c.fetchNext)
}
func topicPath(datadir string, topic string) string {
	var r string
	if datadir != "" {
		r += datadir + "/"
	}
	if topic != "" {
		r += topic
	}
	return r
}
func (p *fileProducer) topicPath(topic string) string {
	return topicPath(p.datadir, topic)
}
func (p *fileConsumer) topicPath(topic string) string {
	return topicPath(p.datadir, topic)
}
func (p *fileConsumer) nextFile(topic string, curFile string) (string, error) {
	tp := p.topicPath(topic)
	dir := filepath.Dir(tp)
	if curFile == "" {
		curFile = tp
	}
	files, err := p.fs.ReadDir(dir, curFile)
	if err != nil {
		if os.IsNotExist(err) {
			return "", nil
		}
		return "", err
	}
	if len(files) == 0 {
		return "", nil
	}
	i := sort.Search(len(files), func(i int) bool {
		fn := dir + "/" + files[i].Name()
		return fn > curFile
	})
	if i < len(files) {
		fn := dir + "/" + files[i].Name()
		if strings.HasPrefix(fn, tp) && !files[i].IsDir() {
			log.Debugf("%v NextFile: %v,  CurFile: %v", topic, files[i].Name(), curFile)
			return files[i].Name(), nil
		}
	}
	return "", nil
}
func (p *fileConsumer) seek(topic string, offset int64) (string, int64, error) {
	tp := p.topicPath(topic)
	dir := filepath.Dir(tp)
	files, err := p.fs.ReadDir(dir, tp)
	if err != nil {
		if os.IsNotExist(err) {
			return "", 0, nil
		}
		return "", 0, err
	}
	if len(files) == 0 {
		return "", 0, nil
	}
	if offset == OffsetOldest {
		for _, f := range files {
			if strings.HasPrefix(dir+"/"+f.Name(), tp) && !f.IsDir() {
				if strings.HasSuffix(f.Name(), ".open") {
					return "", 0, nil
				}
				return f.Name(), 0, nil
			}
		}
		return "", 0, nil
	}
	if offset == OffsetNewest {
		for i := len(files) - 1; i >= 0; i-- {
			if strings.HasPrefix(dir+"/"+files[i].Name(), tp) && !files[i].IsDir() {
				return files[i].Name(), files[i].Size(), nil
			}
		}
		return "", 0, nil
	}
	return "", 0, fmt.Errorf("arbitrary offsets not supported, only OffsetOldest and OffsetNewest offsets supported")
}
func (p *fileProducer) newFileName(key string) string {
	p.seqno++  
	format := "%s%010d.%03d.%s"
	if p.cfg.Compression {
		format += ".gz"
	}
	if p.cfg.Encryption.Enabled {
		format += ".gpg"
	}
	return fmt.Sprintf(format+".open", p.topicPath(p.topic), time.Now().Unix(), p.seqno, key)
}
func (p *fileProducer) initCrypterWriter(filename string, writer io.WriteCloser) (io.WriteCloser, error) {
	if len(p.cfg.Encryption.PublicKey) == 0 {
		return nil, fmt.Errorf("public key is empty. required for producing encrypted stream")
	}
	block, err := armor.Decode(bytes.NewReader([]byte(p.cfg.Encryption.PublicKey)))
	if log.E(err) {
		return nil, err
	}
	if block.Type != openpgp.PublicKeyType {
		return nil, fmt.Errorf("expected public key. got:%s", block.Type)
	}
	encEntity, err := openpgp.ReadEntity(packet.NewReader(block.Body))
	if log.E(err) {
		return nil, err
	}
	block, err = armor.Decode(bytes.NewReader([]byte(p.cfg.Encryption.SigningKey)))
	if log.E(err) {
		return nil, err
	}
	if block.Type != openpgp.PrivateKeyType {
		return nil, fmt.Errorf("expected private key. got:%s", block.Type)
	}
	signEntity, err := openpgp.ReadEntity(packet.NewReader(block.Body))
	if log.E(err) {
		return nil, err
	}
	err = signEntity.PrivateKey.Decrypt([]byte(signKeyPw))
	log.Debugf("signKeyPw %v", err)
	if log.E(err) {
		return nil, err
	}
	log.Debugf("pubKey:  %x", encEntity.PrimaryKey.Fingerprint)
	log.Debugf("signKey: %x", signEntity.PrimaryKey.Fingerprint)
	hints := &openpgp.FileHints{
		IsBinary: true,
		FileName: filename,
		ModTime:  time.Now(),
	}
	writer, err = openpgp.Encrypt(writer, []*openpgp.Entity{encEntity}, signEntity, hints, nil)
	if log.E(err) {
		return nil, err
	}
	return writer, nil
}
func listInsert(p *fileProducer, n *file) {
	if p.flast != nil {
		p.flast.next = n
	}
	if p.ffirst == nil {
		p.ffirst = n
	}
	p.flast = n
}
func listRemove(p *fileProducer, d *file) {
	if d.prev != nil {
		d.prev.next = d.next
	} else {
		p.ffirst = d.next
	}
	if d.next != nil {
		d.next.prev = d.prev
	} else {
		p.flast = d.prev
	}
}
func (p *fileProducer) newFile(key string) error {
	if err := p.fs.MkdirAll(filepath.Dir(p.topicPath(p.topic)), dirPerm); err != nil {
		return err
	}
	n := p.newFileName(key)
	w, seeker, err := p.fs.OpenWrite(n)
	if err != nil {
		return err
	}
	var offset int64
	if seeker != nil {
		offset, err = seeker.Seek(0, io.SeekEnd)
		if err != nil {
			return err
		}
	}
	h := sha256.New()
	hw := &hashWriter{w, h, p.metrics, nil}
	var writer flushWriteCloser = hw
	if p.cfg.Encryption.Enabled {
		w, err := p.initCrypterWriter(n, writer)
		if err != nil {
			return err
		}
		writer = &chainer{&noopFlusher{w}, writer}
	}
	writer = &chainer{&flushClose{bufio.NewWriter(writer)}, writer}
	if p.cfg.Compression {
		writer = &chainer{gzip.NewWriter(writer), writer}
	}
	_ = p.closeFile(p.files[key], true)
	log.Debugf("Opened: %v, %v compression: %v", key, n, p.cfg.Compression)
	f := &file{n, key, w, seeker, h, offset, 0, writer, p.flast, nil, offset}
	hw.f = f
	listInsert(p, f)
	p.files[key] = f
	p.metrics.FilesOpened.Inc(1)
	return nil
}
func (p *fileProducer) getFile(key string) (*file, error) {
	f := p.files[key]
	if f == nil {
		if err := p.newFile(key); err != nil {
			return nil, err
		}
		f = p.files[key]
	}
	return f, nil
}
func (p *fileProducer) cancel(f *file) {
	err := p.fs.Remove(strings.TrimSuffix(f.name, ".open"))
	if err != nil && !os.IsNotExist(err) {
		log.E(err)
	}
	err = p.fs.Remove(f.name)
	if err != nil && !os.IsNotExist(err) {
		log.E(err)
	}
	log.E(p.fs.Cancel(f.file))
}
func syncFsMetadata() error {
	_, _, errno := unix.Syscall(unix.SYS_SYNC, 0, 0, 0)
	if errno != 0 {
		return errno
	}
	return nil
}
func (p *fileProducer) closeFile(f *file, graceful bool) error {
	log.Debugf("Closed: %v", f)
	if f == nil {
		return nil
	}
	var rerr error
	defer func() {
		listRemove(p, f)
		delete(p.files, f.key)
		if rerr != nil || !graceful {
			p.cancel(f)
		}
	}()
	if err := f.writer.Close(); log.E(err) {
		rerr = err
	}
	fn := strings.TrimSuffix(f.name, ".open")
	if graceful && rerr == nil {
		if err := p.fs.Rename(f.name, fn); log.E(err) {
			rerr = err
		}
	}
	p.stats[fn] = &stat{NumRecs: f.nRecs, Hash: fmt.Sprintf("%x", f.hash.Sum(nil)), FileName: fn}
	p.metrics.FilesClosed.Inc(1)
	log.E(syncFsMetadata())
	log.Debugf("Closed: %v", f.name)
	return rerr
}
func (p *fileProducer) writeBinaryMsgLength(f *file, len int) error {
	if atomic.LoadInt64(&p.text) == 1 || !p.cfg.FileDelimited {
		return nil
	}
	sz := make([]byte, 4)
	binary.LittleEndian.PutUint32(sz, uint32(len))
	_, err := f.writer.Write(sz)
	return err
}
func (p *fileProducer) writeTextMsgDelimiter(f *file) error {
	if atomic.LoadInt64(&p.text) == 0 || !p.cfg.FileDelimited {
		return nil
	}
	o := make([]byte, 0)
	o = append(o, delimiter)
	_, err := f.writer.Write(o)
	return err
}
func (p *fileProducer) rotateOnSizeLimit(key string, f *file) {
	if (p.cfg.MaxFileDataSize != 0 && f.offset >= p.cfg.MaxFileDataSize) || (p.cfg.MaxFileSize != 0 && f.compressedSize > p.cfg.MaxFileSize) {
		_ = p.closeFile(p.files[key], true)
	}
}
func (p *fileProducer) push(key string, in interface{}, batch bool) error {
	var bytes []byte
	switch b := in.(type) {
	case []byte:
		bytes = b
	default:
		return fmt.Errorf("file pipe can handle binary arrays only")
	}
	f, err := p.getFile(key)
	if err != nil {
		return err
	}
	defer func() {
		if err != nil {
			p.cancel(f)
		}
	}()
	if err = p.writeBinaryMsgLength(f, len(bytes)); err != nil {
		return err
	}
	if _, err = f.writer.Write(bytes); err != nil {
		return err
	}
	if err = p.writeTextMsgDelimiter(f); err != nil {
		return err
	}
	f.offset += int64(len(bytes)) + 1
	f.nRecs++
	if !batch {
		if err = f.writer.Flush(); err != nil {
			return err
		}
		p.rotateOnSizeLimit(key, f)
	}
	return nil
}
func (p *fileProducer) PushK(key string, in interface{}) error {
	return p.push(key, in, false)
}
func (p *fileProducer) Push(in interface{}) error {
	return p.push("default", in, false)
}