utils-shell/peco.peco
1
0
Fork 0
mirror of https://github.com/peco/peco.git synced 2026-09-10 07:16:29 -04:00
Code Issues Actions Packages Projects Releases Wiki Activity

Integrate ExternalCommand into the new format

Browse source
This commit is contained in:
Daisuke Maki 2016-12-14 14:59:38 +09:00
parent 80c7cde206
commit 47be5c7c85
13 changed files with 228 additions and 201 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
buffer.go
View file

@ -97,7 +97,7 @@ func (mb *MemoryBuffer) Done() <-chan struct{} {
return mb.done
}
func (mb *MemoryBuffer) Accept(ctx context.Context, in chan interface{}, _ pipeline.OutputChannel) {
func (mb *MemoryBuffer) Accept(ctx context.Context, in chan interface{}, _ pipeline.ChanOutput) {
if pdebug.Enabled {
g := pdebug.Marker("MemoryBuffer.Accept")
defer g.End()

44
filter.go
View file

@ -21,13 +21,13 @@ func newFilterProcessor(f filter.Filter, q string) *filterProcessor {
}
}
func (fp *filterProcessor) Accept(ctx context.Context, in chan interface{}, out pipeline.OutputChannel) {
func (fp *filterProcessor) Accept(ctx context.Context, in chan interface{}, out pipeline.ChanOutput) {
acceptAndFilter(ctx, fp.filter, in, out)
}
// This flusher is run in a separate goroutine so that the filter can
// run separately from accepting incoming messages
func flusher(ctx context.Context, f filter.Filter, incoming chan []line.Line, done chan struct{}, out pipeline.OutputChannel) {
func flusher(ctx context.Context, f filter.Filter, incoming chan []line.Line, done chan struct{}, out pipeline.ChanOutput) {
if pdebug.Enabled {
g := pdebug.Marker("flusher goroutine")
defer g.End()
@ -35,23 +35,32 @@ func flusher(ctx context.Context, f filter.Filter, incoming chan []line.Line, do
defer close(done)
defer out.SendEndMark("end of filter")
for buf := range incoming {
for _, in := range buf {
if l, err := f.Apply(ctx, in); err == nil {
out.Send(l)
for {
select {
case <-ctx.Done():
return
case buf, ok := <-incoming:
if !ok {
return
}
pdebug.Printf("flusher: %#v", buf)
f.Apply(ctx, buf, out)
buffer.ReleaseLineListBuf(buf)
}
buffer.ReleaseLineListBuf(buf)
}
}
func acceptAndFilter(ctx context.Context, f filter.Filter, in chan interface{}, out pipeline.OutputChannel) {
func acceptAndFilter(ctx context.Context, f filter.Filter, in chan interface{}, out pipeline.ChanOutput) {
flush := make(chan []line.Line)
flushDone := make(chan struct{})
go flusher(ctx, f, flush, flushDone, out)
buf := buffer.GetLineListBuf()
defer buffer.ReleaseLineListBuf(buf)
bufsiz := f.BufSize()
if bufsiz <= 0 {
bufsiz = cap(buf)
}
defer func() { <-flushDone }() // Wait till the flush goroutine is done
defer close(flush) // Kill the flush goroutine
@ -67,6 +76,9 @@ func acceptAndFilter(ctx context.Context, f filter.Filter, in chan interface{},
pdebug.Printf("filter received done")
}
return
case <-flushTicker.C:
flush <- buf
buf = buffer.GetLineListBuf()
case v := <-in:
switch v.(type) {
case error:
@ -82,6 +94,7 @@ func acceptAndFilter(ctx context.Context, f filter.Filter, in chan interface{},
return
case line.Line:
if pdebug.Enabled {
pdebug.Printf("incoming line")
lines++
}
// We buffer the lines so that we can receive more lines to
@ -89,15 +102,9 @@ func acceptAndFilter(ctx context.Context, f filter.Filter, in chan interface{},
// size is fairly big, because this really only makes a
// difference if we have a lot of lines to process.
buf = append(buf, v.(line.Line))
select {
case <-flushTicker.C:
if len(buf) >= bufsiz {
flush <- buf
buf = buffer.GetLineListBuf()
default:
if len(buf) >= cap(buf) {
flush <- buf
buf = buffer.GetLineListBuf()
}
}
}
}
@ -139,7 +146,9 @@ func (f *Filter) Work(ctx context.Context, q hub.Payload) {
p.SetSource(state.Source())
// Wraps the actual filter
p.Add(newFilterProcessor(state.Filters().Current(), query))
selectedFilter := state.Filters().Current()
ctx = selectedFilter.NewContext(ctx, query)
p.Add(newFilterProcessor(selectedFilter, query))
buf := NewMemoryBuffer()
p.SetDestination(buf)
@ -147,7 +156,6 @@ func (f *Filter) Work(ctx context.Context, q hub.Payload) {
go func() {
defer state.Hub().SendDraw(&DrawOptions{RunningQuery: true})
ctx = filter.NewContext(ctx, query)
if err := p.Run(ctx); err != nil {
state.Hub().SendStatusMsg(err.Error())
}

102
filter/external.go
View file

@ -12,6 +12,8 @@ import (
"github.com/pkg/errors"
)
// NewExternalCmd creates a new filter that uses an external
// command to filter the input
func NewExternalCmd(name string, cmd string, args []string, threshold int, idgen line.IDGenerator, enableSep bool) *ExternalCmd {
if len(args) == 0 {
args = []string{"$QUERY"}
@ -27,77 +29,33 @@ func NewExternalCmd(name string, cmd string, args []string, threshold int, idgen
enableSep: enableSep,
idgen: idgen,
name: name,
outCh: pipeline.OutputChannel(make(chan interface{})),
outCh: pipeline.ChanOutput(make(chan interface{})),
thresholdBufsiz: threshold,
}
}
func (ecf *ExternalCmd) Verify() error {
if ecf.cmd == "" {
return errors.Errorf("no executable specified for custom matcher '%s'", ecf.name)
}
if _, err := exec.LookPath(ecf.cmd); err != nil {
return errors.Wrap(err, "failed to locate command")
}
return nil
func (ecf ExternalCmd) BufSize() int {
return ecf.thresholdBufsiz
}
func (ecf *ExternalCmd) Apply(ctx context.Context, l line.Line) (line.Line, error) {
return nil, nil
}
func (ecf *ExternalCmd) Accept(ctx context.Context, in chan interface{}, out pipeline.OutputChannel) {
if pdebug.Enabled {
g := pdebug.Marker("ExternalCmd.Accept")
defer g.End()
}
defer out.SendEndMark("end of ExternalCmd")
buf := make([]line.Line, 0, ecf.thresholdBufsiz)
for {
select {
case <-ctx.Done():
if pdebug.Enabled {
pdebug.Printf("ExternalCmd received done")
}
return
case v := <-in:
switch v.(type) {
case error:
if pipeline.IsEndMark(v.(error)) {
if pdebug.Enabled {
pdebug.Printf("ExternalCmd received end mark")
}
if len(buf) > 0 {
ecf.launchExternalCmd(ctx, buf, out)
}
}
return
case line.Line:
if pdebug.Enabled {
pdebug.Printf("ExternalCmd received new line")
}
buf = append(buf, v.(line.Line))
if len(buf) < ecf.thresholdBufsiz {
continue
}
ecf.launchExternalCmd(ctx, buf, out)
buf = buf[0:0]
}
}
}
func (ecf *ExternalCmd) NewContext(ctx context.Context, query string) context.Context {
return newContext(ctx, query)
}
func (ecf ExternalCmd) String() string {
return ecf.name
}
func (ecf *ExternalCmd) launchExternalCmd(ctx context.Context, buf []line.Line, out pipeline.OutputChannel) {
defer func() { recover() }() // ignore errors
func (ecf *ExternalCmd) Apply(ctx context.Context, buf []line.Line, out pipeline.ChanOutput) (err error) {
defer func() {
if err := recover(); err != nil {
if pdebug.Enabled {
pdebug.Printf("err: %s", err)
}
}
}() // ignore errors
if pdebug.Enabled {
g := pdebug.Marker("ExternalCmd.launchExternalCmd")
g := pdebug.Marker("ExternalCmd.Apply").BindError(&err)
defer g.End()
}
@ -108,6 +66,7 @@ func (ecf *ExternalCmd) launchExternalCmd(ctx context.Context, buf []line.Line,
args[i] = query
}
}
cmd := exec.Command(ecf.cmd, args...)
if pdebug.Enabled {
pdebug.Printf("Executing command %s %v", cmd.Path, cmd.Args)
@ -121,34 +80,44 @@ func (ecf *ExternalCmd) launchExternalCmd(ctx context.Context, buf []line.Line,
cmd.Stdin = inbuf
r, err := cmd.StdoutPipe()
if err != nil {
return
return errors.Wrap(err, `failed to get stdout pipe`)
}
err = cmd.Start()
if err != nil {
return
return errors.Wrap(err, `failed to start command`)
}
go cmd.Wait()
cmdCh := make(chan line.Line)
go func(cmdCh chan line.Line, rdr *bufio.Reader) {
go func(ctx context.Context, cmdCh chan line.Line, rdr *bufio.Reader) {
defer func() { recover() }()
defer close(cmdCh)
for {
select {
case <-ctx.Done():
return
default:
}
b, _, err := rdr.ReadLine()
if len(b) > 0 {
// TODO: need to redo the spec for custom matchers
// This is the ONLY location where we need to actually
// RECREATE a Raw, and thus the only place where
// ctx.enableSep is required.
cmdCh <- line.NewMatched(line.NewRaw(ecf.idgen.Next(), string(b), ecf.enableSep), nil)
select {
case cmdCh <- line.NewRaw(ecf.idgen.Next(), string(b), ecf.enableSep):
case <-ctx.Done():
return
}
}
if err != nil {
break
return
}
}
}(cmdCh, bufio.NewReader(r))
}(ctx, cmdCh, bufio.NewReader(r))
defer func() {
if p := cmd.Process; p != nil {
@ -159,12 +128,13 @@ func (ecf *ExternalCmd) launchExternalCmd(ctx context.Context, buf []line.Line,
for {
select {
case <-ctx.Done():
return
return nil
case l, ok := <-cmdCh:
if l == nil || !ok {
return
return nil
}
out.Send(l)
}
}
return nil
}

8
filter/filter.go
View file

@ -1,5 +1,13 @@
package filter
import "context"
// newContext initializes the context so that it is suitable
// to be passed to `Run()`
func newContext(ctx context.Context, query string) context.Context {
return context.WithValue(ctx, queryKey, query)
}
// sort related stuff
type byMatchStart [][]int

41
filter/filter_test.go
View file

@ -4,8 +4,10 @@ import (
"context"
"fmt"
"testing"
"time"
"github.com/peco/peco/line"
"github.com/peco/peco/pipeline"
"github.com/stretchr/testify/assert"
)
@ -15,7 +17,8 @@ type indexer interface {
// TestFuzzy tests a fuzzy filter against various inputs
func TestFuzzy(t *testing.T) {
ctx := context.Background()
octx, ocancel := context.WithCancel(context.Background())
defer ocancel()
testValues := []struct {
input string
@ -38,32 +41,32 @@ func TestFuzzy(t *testing.T) {
filter := NewFuzzy()
for i, v := range testValues {
t.Run(fmt.Sprintf(`"%s" against "%s", expect "%t"`, v.input, v.query, v.selected), func(t *testing.T) {
ctx = NewContext(ctx, v.query)
ctx, cancel := context.WithTimeout(filter.NewContext(octx, v.query), 10*time.Second)
defer cancel()
ch := make(chan interface{}, 1)
l := line.NewRaw(uint64(i), v.input, false)
res, err := filter.Apply(ctx, l)
err := filter.Apply(ctx, []line.Line{l}, pipeline.ChanOutput(ch))
if !assert.NoError(t, err, `filter.Apply should succeeed`) {
return
}
if !v.selected {
if !assert.Error(t, err, "filter should fail") {
select {
case l, ok := <-ch:
if !assert.True(t, ok, `channel read should succeed`) {
return
}
if !assert.Nil(t, res, "return value should be nil") {
if !assert.Implements(t, (*line.Line)(nil), l, "result is a line") {
return
}
return
}
if !assert.NoError(t, err, "filtering failed") {
return
t.Logf("%#v", l.(indexer).Indices())
case <-ctx.Done():
if !assert.False(t, v.selected, "did NOT expect to timeout") { // shouldn't happen if we're expecting a result
return
}
}
if !assert.NotNil(t, res, "return value should NOT be nil") {
return
}
if !assert.Implements(t, (*indexer)(nil), res, "can call Indices()") {
return
}
t.Logf("%#v", res.(indexer).Indices())
})
}
}

71
filter/fuzzy.go
View file

@ -2,12 +2,12 @@ package filter
import (
"context"
"errors"
"strings"
"unicode/utf8"
"github.com/peco/peco/internal/util"
"github.com/peco/peco/line"
"github.com/peco/peco/pipeline"
)
// NewFuzzy builds a fuzzy-finder type of filter.
@ -17,41 +17,52 @@ func NewFuzzy() *Fuzzy {
return &Fuzzy{}
}
func (ff Fuzzy) BufSize() int {
return 0
}
func (ff *Fuzzy) NewContext(ctx context.Context, query string) context.Context {
return newContext(ctx, query)
}
func (ff Fuzzy) String() string {
return "Fuzzy"
}
func (ff *Fuzzy) Apply(ctx context.Context, l line.Line) (line.Line, error) {
query := ctx.Value(queryKey).(string)
base := 0
txt := l.DisplayString()
matches := [][]int{}
func (ff *Fuzzy) Apply(ctx context.Context, lines []line.Line, out pipeline.ChanOutput) error {
originalQuery := ctx.Value(queryKey).(string)
hasUpper := util.ContainsUpper(originalQuery)
hasUpper := util.ContainsUpper(query)
for _, l := range lines {
base := 0
matches := [][]int{}
txt := l.DisplayString()
query := originalQuery
for len(query) > 0 {
r, n := utf8.DecodeRuneInString(query)
query = query[n:]
if r == utf8.RuneError {
// "Silently" ignore
continue
}
for len(query) > 0 {
r, n := utf8.DecodeRuneInString(query)
if r == utf8.RuneError {
// "Silently" ignore (just return a no match)
return nil, errors.New("failed to decode input string")
var i int
if hasUpper { // explicit match
i = strings.IndexRune(txt, r)
} else {
i = strings.IndexFunc(txt, util.CaseInsensitiveIndexFunc(r))
}
if i == -1 {
continue
}
// otherwise we have a match, but the next match must match against
// something AFTER the current match
txt = txt[i+n:]
matches = append(matches, []int{base + i, base + i + n})
base = base + i + n
}
query = query[n:]
var i int
if hasUpper { // explicit match
i = strings.IndexRune(txt, r)
} else {
i = strings.IndexFunc(txt, util.CaseInsensitiveIndexFunc(r))
}
if i == -1 {
return nil, errors.New("filter did not match against given line")
}
// otherwise we have a match, but the next match must match against
// something AFTER the current match
txt = txt[i+n:]
matches = append(matches, []int{base + i, base + i + n})
base = base + i + n
out.Send(line.NewMatched(l, matches))
}
return line.NewMatched(l, matches), nil
return nil
}

11
filter/interface.go
View file

@ -15,7 +15,8 @@ var ErrFilterNotFound = errors.New("specified filter was not found")
var ignoreCaseFlags = regexpFlagList([]string{"i"})
var defaultFlags = regexpFlagList{}
var queryKey = struct{}{}
var queryKey = &struct{}{}
var incomingBufferKey = &struct{}{}
// DefaultCustomFilterBufferThreshold is the default value
// for BufferThreshold setting on CustomFilters.
@ -56,7 +57,7 @@ type Regexp struct {
mutex sync.Mutex
name string
onEnd func()
outCh pipeline.OutputChannel
outCh pipeline.ChanOutput
}
type ExternalCmd struct {
@ -64,12 +65,14 @@ type ExternalCmd struct {
cmd string
enableSep bool
idgen line.IDGenerator
outCh pipeline.OutputChannel
outCh pipeline.ChanOutput
name string
thresholdBufsiz int
}
type Filter interface {
Apply(context.Context, line.Line) (line.Line, error)
Apply(context.Context, []line.Line, pipeline.ChanOutput) error
BufSize() int
NewContext(context.Context, string) context.Context
String() string
}

98
filter/regexp.go
View file

@ -54,10 +54,8 @@ func queryToRegexps(query string, flags regexpFlags, quotemeta bool) ([]*regexp.
return regexps, nil
}
// NewContext initializes the context so that it is suitable
// to be passed to `Run()`
func NewContext(ctx context.Context, query string) context.Context {
return context.WithValue(ctx, queryKey, query)
func (rf *Regexp) NewContext(ctx context.Context, query string) context.Context {
return newContext(ctx, query)
}
// NewRegexp creates a new regexp based filter
@ -70,10 +68,14 @@ func NewRegexp() *Regexp {
flags: regexpFlagList(defaultFlags),
quotemeta: false,
name: "Regexp",
outCh: pipeline.OutputChannel(make(chan interface{})),
outCh: pipeline.ChanOutput(make(chan interface{})),
}
}
func (rf Regexp) BufSize() int {
return 0
}
func (rf *Regexp) OutCh() <-chan interface{} {
rf.mutex.Lock()
defer rf.mutex.Unlock()
@ -103,58 +105,62 @@ func (f *regexpQueryFactory) Compile(s string, flags regexpFlags, quotemeta bool
return rxs, nil
}
func (rf *Regexp) Apply(ctx context.Context, l line.Line) (line.Line, error) {
func (rf *Regexp) Apply(ctx context.Context, lines []line.Line, out pipeline.ChanOutput) error {
query := ctx.Value(queryKey).(string)
regexps, err := rf.factory.Compile(query, rf.flags, rf.quotemeta)
if err != nil {
return nil, errors.Wrap(err, "failed to compile queries as regular expression")
return errors.Wrap(err, "failed to compile queries as regular expression")
}
v := l.DisplayString()
allMatched := true
matches := [][]int{}
TryRegexps:
for _, rx := range regexps {
match := rx.FindAllStringSubmatchIndex(v, -1)
if match == nil {
allMatched = false
break TryRegexps
for _, l := range lines {
v := l.DisplayString()
allMatched := true
matches := [][]int{}
TryRegexps:
for _, rx := range regexps {
match := rx.FindAllStringSubmatchIndex(v, -1)
if match == nil {
allMatched = false
break TryRegexps
}
matches = append(matches, match...)
}
matches = append(matches, match...)
}
if !allMatched {
return nil, errors.New("filter did not match against given line")
}
sort.Sort(byMatchStart(matches))
// We need to "dedupe" the results. For example, if we matched the
// same region twice, we don't want that to be drawn
deduped := make([][]int, 0, len(matches))
for i, m := range matches {
// Always push the first one
if i == 0 {
deduped = append(deduped, m)
if !allMatched {
continue
}
prev := deduped[len(deduped)-1]
switch {
case matchContains(prev, m):
// If the previous match contains this one, then
// don't do anything
continue
case matchOverlaps(prev, m):
// If the previous match overlaps with this one,
// merge the results and make it a bigger one
deduped[len(deduped)-1] = mergeMatches(prev, m)
default:
deduped = append(deduped, m)
sort.Sort(byMatchStart(matches))
// We need to "dedupe" the results. For example, if we matched the
// same region twice, we don't want that to be drawn
deduped := make([][]int, 0, len(matches))
for i, m := range matches {
// Always push the first one
if i == 0 {
deduped = append(deduped, m)
continue
}
prev := deduped[len(deduped)-1]
switch {
case matchContains(prev, m):
// If the previous match contains this one, then
// don't do anything
continue
case matchOverlaps(prev, m):
// If the previous match overlaps with this one,
// merge the results and make it a bigger one
deduped[len(deduped)-1] = mergeMatches(prev, m)
default:
deduped = append(deduped, m)
}
}
out.Send(line.NewMatched(l, deduped))
}
return line.NewMatched(l, deduped), nil
return nil
}
func (rf Regexp) String() string {

2
interface.go
View file

@ -369,7 +369,7 @@ type FilterQuery Query
// Source implements pipeline.Source, and is the buffer for the input
type Source struct {
pipeline.OutputChannel
pipeline.ChanOutput
done chan struct{}
capacity int

2
layout.go
View file

@ -441,8 +441,6 @@ func (l *ListArea) Draw(state *Peco, parent Layout, perPage int, options *DrawOp
xOffset := loc.Column()
line := target.DisplayString()
pdebug.Printf("state.SingleKeyJumpMode = %t", state.SingleKeyJumpMode())
pdebug.Printf("state.SingleKeyJumpShowPrefix = %t", state.SingleKeyJumpShowPrefix())
if state.SingleKeyJumpMode() || state.SingleKeyJumpShowPrefix() {
prefixes := state.SingleKeyJumpPrefixes()
if n < int(len(prefixes)) {

12
pipeline/interface.go
View file

@ -19,7 +19,7 @@ type EndMark struct{}
type Source interface {
// Start should be able to be called repeatedly, producing the
// same data to be consumed by the chained Acceptors
Start(context.Context, OutputChannel)
Start(context.Context, ChanOutput)
Reset()
}
@ -27,7 +27,7 @@ type Source interface {
// Acceptor is an object that can accept input, and send to
// an optional output
type Acceptor interface {
Accept(context.Context, chan interface{}, OutputChannel)
Accept(context.Context, chan interface{}, ChanOutput)
}
// Destination is a special case Acceptor that has no more Acceptors
@ -47,5 +47,9 @@ type Pipeline struct {
dst Destination
}
// OutputChannel is an alias to `chan interface{}`
type OutputChannel chan interface{}
type Output interface {
Send(interface{}) error
}
// ChanOutput is an alias to `chan interface{}`
type ChanOutput chan interface{}

30
pipeline/pipeline.go
View file

@ -4,9 +4,10 @@ package pipeline
import (
"time"
"context"
pdebug "github.com/lestrrat/go-pdebug"
"github.com/pkg/errors"
"context"
)
// EndMark returns true
@ -28,13 +29,28 @@ func IsEndMark(err error) bool {
return false
}
func NilOutput(ctx context.Context) ChanOutput {
ch := make(chan interface{})
go func() {
for {
select {
case <-ctx.Done():
return
case <-ch:
}
}
}()
return ChanOutput(ch)
}
// OutCh returns the channel that acceptors can listen to
func (oc OutputChannel) OutCh() <-chan interface{} {
func (oc ChanOutput) OutCh() <-chan interface{} {
return oc
}
// Send sends the data `v` through this channel
func (oc OutputChannel) Send(v interface{}) (err error) {
func (oc ChanOutput) Send(v interface{}) (err error) {
if oc == nil {
return errors.New("nil channel")
}
@ -52,7 +68,7 @@ func (oc OutputChannel) Send(v interface{}) (err error) {
}
// SendEndMark sends an end mark
func (oc OutputChannel) SendEndMark(s string) error {
func (oc ChanOutput) SendEndMark(s string) error {
return errors.Wrap(oc.Send(errors.Wrap(EndMark{}, s)), "failed to send end mark")
}
@ -118,14 +134,14 @@ func (p *Pipeline) Run(ctx context.Context) (err error) {
// Setup the Acceptors, effectively chaining all nodes
// starting from the destination, working all the way
// up to the Source
var prevCh OutputChannel = OutputChannel(make(chan interface{}))
var prevCh ChanOutput = ChanOutput(make(chan interface{}))
go p.dst.Accept(ctx, prevCh, nil)
for i := len(p.nodes) - 1; i >= 0; i-- {
cur := p.nodes[i]
ch := make(chan interface{}) //
ch := make(chan interface{}) //
go cur.Accept(ctx, ch, prevCh)
prevCh = OutputChannel(ch)
prevCh = ChanOutput(ch)
}
// And now tell the Source to send the values so data chugs

6
source.go
View file

@ -25,7 +25,7 @@ func NewSource(in io.Reader, idgen line.IDGenerator, capacity int, enableSep boo
ready: make(chan struct{}),
setupDone: make(chan struct{}),
setupOnce: sync.Once{},
OutputChannel: pipeline.OutputChannel(make(chan interface{})),
ChanOutput: pipeline.ChanOutput(make(chan interface{})),
}
s.Reset()
return s
@ -128,7 +128,7 @@ func (s *Source) Setup(ctx context.Context, state *Peco) {
}
// Start starts
func (s *Source) Start(ctx context.Context, out pipeline.OutputChannel) {
func (s *Source) Start(ctx context.Context, out pipeline.ChanOutput) {
// I should be the only one running this method until I bail out
if pdebug.Enabled {
g := pdebug.Marker("Source.Start")
@ -157,7 +157,7 @@ func (s *Source) Reset() {
g := pdebug.Marker("Source.Reset")
defer g.End()
}
s.OutputChannel = pipeline.OutputChannel(make(chan interface{}))
s.ChanOutput = pipeline.ChanOutput(make(chan interface{}))
}
// Ready returns the "input ready" channel. It will be closed as soon as