forked from calibrae-project/spawn
-
Notifications
You must be signed in to change notification settings - Fork 0
/
pipe.go
102 lines (98 loc) · 2.27 KB
/
pipe.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
package pipe
import (
"github.com/p9c/log"
"io"
"os"
"github.com/p9c/pod/pkg/pipe/stdconn"
"github.com/p9c/pod/pkg/pipe/stdconn/worker"
"github.com/p9c/pod/pkg/util/interrupt"
"github.com/p9c/qu"
)
// Consume listens for messages from a child process over a stdio pipe.
func Consume(quit qu.C, handler func([]byte) error, args ...string) *worker.Worker {
var n int
var e error
D.Ln("spawning worker process", args)
w, _ := worker.Spawn(quit, args...)
data := make([]byte, 8192)
// onBackup := false
go func() {
out:
for {
// D.Ln("readloop")
select {
case <-interrupt.HandlersDone.Wait():
D.Ln("quitting log consumer")
break out
case <-quit.Wait():
D.Ln("breaking on quit signal")
break out
default:
}
n, e = w.StdConn.Read(data)
if n == 0 {
F.Ln("read zero from stdconn", args)
// onBackup = true
log.LogChanDisabled.Store(true)
break out
}
if E.Chk(e) && e != io.EOF {
// Probably the child process has died, so quit
E.Ln("err:", e)
// onBackup = true
break out
} else if n > 0 {
if e = handler(data[:n]); E.Chk(e) {
}
}
// if n, e = w.StdPipe.Read(data); E.Chk(e) {
// }
// // when the child stops sending over RPC, fall back to the also working but not printing stderr
// if n > 0 {
// prefix := "[" + args[len(args)-1] + "]"
// if onBackup {
// prefix += "b"
// }
// printIt := true
// if logi.L.LogChanDisabled {
// printIt = false
// }
// if printIt {
// fmt.Fprint(os.Stderr, prefix+" "+string(data[:n]))
// }
// }
}
}()
return w
}
// Serve runs a goroutine processing the FEC encoded packets, gathering them and
// decoding them to be delivered to a handler function
func Serve(quit qu.C, handler func([]byte) error) *stdconn.StdConn {
var n int
var e error
data := make([]byte, 8192)
go func() {
D.Ln("starting pipe server")
out:
for {
select {
case <-quit.Wait():
// D.Ln(interrupt.GoroutineDump())
break out
default:
}
n, e = os.Stdin.Read(data)
if e != nil && e != io.EOF {
break out
}
if n > 0 {
if e = handler(data[:n]); E.Chk(e) {
break out
}
}
}
// D.Ln(interrupt.GoroutineDump())
D.Ln("pipe server shut down")
}()
return stdconn.New(os.Stdin, os.Stdout, quit)
}