close pipes at end of stream

1 files changed, 28 insertions, 14 deletions

diff --git a/pkg/server/hook/hook.go b/pkg/server/hook/hook.go
index e4497ff..1702afd 100644
--- a/pkg/server/hook/hook.go
+++ b/pkg/server/hook/hook.go
@@ -14,12 +14,18 @@ type hookInitFunc func(opts.Opts, env.Env) (hook, error)
 
 var hooks = map[string]hookInitFunc{}
 
+type Pipe struct {
+	priv interface{}
+	Send Func
+	Recv Func
+}
+
 type hook interface {
 	Open(env env.Env) (interface{}, error)
 }
 
 type H interface {
-	hook
+	Open(env env.Env) (*Pipe, error)
 	String() string
 }
 
@@ -31,6 +37,10 @@ type Recver interface {
 	Recv(rq, wq queue.Q) error
 }
 
+type Closer interface {
+	Close()
+}
+
 type Func func(rq, wq queue.Q) error
 
 func (f Func) Send(rq, wq queue.Q) error {
@@ -51,29 +61,33 @@ func (w *wrapper) String() string {
 	return fmt.Sprintf("hook:%s", w.name)
 }
 
-func Open(h H, env env.Env) (Func, Func, error) {
-	var send, recv Func
-
-	w := h.(*wrapper)
-
-	it, err := h.Open(env)
+func (w *wrapper) Open(env env.Env) (*Pipe, error) {
+	it, err := w.hook.Open(env)
 	if err != nil {
-		return nil, nil, err
+		return nil, err
 	}
 
-	if sender, ok := it.(Sender); ok {
-		send = sender.Send
+	pipe := &Pipe{priv: it}
+
+	if s, ok := it.(Sender); ok {
+		pipe.Send = s.Send
 	}
 
-	if recver, ok := it.(Recver); ok {
-		recv = recver.Recv
+	if r, ok := it.(Recver); ok {
+		pipe.Recv = r.Recv
 	}
 
 	if w.reverse {
-		send, recv = recv, send
+		pipe.Send, pipe.Recv = pipe.Recv, pipe.Send
 	}
 
-	return send, recv, nil
+	return pipe, nil
+}
+
+func (p *Pipe) Close() {
+	if c, ok := p.priv.(Closer); ok {
+		c.Close()
+	}
 }
 
 func New(desc string, env env.Env) (H, error) {