func (p *localProducerConsumer) PushBatchCommit() error {
	return nil
}
PushBatchCommit sends all stashed messages to pipe
package pipe
import (
	"database/sql"
	"fmt"
	"sync"
	"github.com/uber/storagetapper/config"
	"golang.org/x/net/context"
)
type localPipe struct {
	mutex sync.Mutex
	ch    map[string](chan interface{})
	cfg   config.PipeConfig
}
type localProducerConsumer struct {
	baseConsumer
	ch chan interface{}
}
func init() {
	registerPlugin("local", initLocalPipe)
}
func initLocalPipe(cfg *config.PipeConfig, db *sql.DB) (Pipe, error) {
	return &localPipe{ch: make(map[string](chan interface{})), cfg: *cfg}, nil
}
func (p *localPipe) Type() string {
	return "local"
}
func (p *localPipe) Config() *config.PipeConfig {
	return &p.cfg
}
func (p *localPipe) Close() error {
	return nil
}
func (p *localPipe) registerProducerConsumer(key string, consumer bool) (*localProducerConsumer, error) {
	p.mutex.Lock()
	ch := p.ch[key]
	if ch == nil {
		ch = make(chan interface{}, p.cfg.MaxBatchSize)
		p.ch[key] = ch
	}
	p.mutex.Unlock()
	l := &localProducerConsumer{ch: ch}
	if consumer {
		l.initBaseConsumer(l.fetchNext)
	} else {
		l.ctx, l.cancel = context.WithCancel(context.Background())
	}
	return l, nil
}
func (p *localPipe) NewConsumer(key string) (Consumer, error) {
	return p.registerProducerConsumer(key, true)
}
func (p *localPipe) NewProducer(key string) (Producer, error) {
	return p.registerProducerConsumer(key, false)
}
func (p *localProducerConsumer) pushLow(b interface{}) error {
	select {
	case p.ch <- b:
		return nil
	case <-p.ctx.Done():
	}
	return fmt.Errorf("context canceled")
}
func (p *localProducerConsumer) Push(data interface{}) error {
	return p.pushLow(data)
}
func (p *localProducerConsumer) PushBatch(key string, data interface{}) error {
	return p.pushLow(data)
}