This repository has been archived by the owner on May 31, 2023. It is now read-only.
-
Notifications
You must be signed in to change notification settings - Fork 9
/
procs.go
108 lines (87 loc) · 1.62 KB
/
procs.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
103
104
105
106
107
108
package procs
import (
"errors"
"fmt"
"github.com/Roman2K/scat"
"github.com/Roman2K/scat/concur"
)
var (
ErrShort = errors.New("missing final chunks")
ErrUnreturnedSlots = errors.New("unreturned slots left")
)
var Nop Proc
func init() {
Nop = InplaceFunc(func(*scat.Chunk) error { return nil })
}
type MissingDataError struct {
Err error
}
var _ error = MissingDataError{}
func (err MissingDataError) Error() string {
return fmt.Sprintf("missing data: %v", err.Err)
}
type Proc interface {
Process(*scat.Chunk) <-chan Res
Finish() error
}
type EndProc interface {
ProcessFinal(*scat.Chunk, *scat.Chunk) error
ProcessEnd(*scat.Chunk) error
}
type ErrProc interface {
ProcessErr(*scat.Chunk, error) <-chan Res
}
type WrapperProc interface {
Proc
Underlying() Proc
}
func underlying(p Proc) Proc {
for {
w, ok := p.(WrapperProc)
if !ok {
break
}
p = w.Underlying()
}
return p
}
type Res struct {
Chunk *scat.Chunk
Err error
}
type Procer interface {
Proc() Proc
}
type Unprocer interface {
Unproc() Proc
}
type ProcUnprocer interface {
Procer
Unprocer
}
type DynProcer interface {
Procs(*scat.Chunk) ([]Proc, error)
Finish() error
}
func Process(proc Proc, chunk *scat.Chunk) error {
defer proc.Finish()
for res := range proc.Process(chunk) {
if res.Err != nil {
return res.Err
}
}
return proc.Finish()
}
func SingleRes(c *scat.Chunk, err error) <-chan Res {
ch := make(chan Res, 1)
defer close(ch)
ch <- Res{c, err}
return ch
}
func finishFuncs(procs []Proc) (fns concur.Funcs) {
fns = make(concur.Funcs, len(procs))
for i, p := range procs {
fns[i] = p.Finish
}
return
}