-
Notifications
You must be signed in to change notification settings - Fork 1
/
output.go
83 lines (72 loc) · 1.73 KB
/
output.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
package output
import (
"fmt"
"github.com/gfleury/gstreamtop/conf"
"github.com/gfleury/gstreamtop/profiling"
"github.com/gfleury/gstreamtop/tablestream"
"time"
)
type Outputer interface {
Loop()
Configure() error
Stream() *tablestream.Stream
ErrorChan() *chan error
CreateStreamFromConfigurationMapping(mapping *conf.Mapping, createNamedQueries []string) error
InputExists() *bool
SetInputExists(func() *bool)
EnableProfile()
Profile() bool
}
type StreamOutput struct {
Outputer
stream *tablestream.Stream
errors chan error
inputExists func() *bool
profile bool
}
func (o *StreamOutput) CreateStreamFromConfigurationMapping(mapping *conf.Mapping, createNamedQueries []string) error {
o.stream = &tablestream.Stream{}
for _, tableDDL := range mapping.Tables {
err := o.Stream().Query(tableDDL)
if err != nil {
return err
}
}
for _, createNamedQuerie := range createNamedQueries {
found := false
for _, query := range mapping.Queries {
if query.Name == createNamedQuerie {
err := o.Stream().Query(query.Query)
if err != nil {
return err
}
found = true
}
}
if !found {
return fmt.Errorf("No query named %s found", createNamedQuerie)
}
}
return nil
}
func (o *StreamOutput) Stream() *tablestream.Stream {
return o.stream
}
func (o *StreamOutput) ErrorChan() *chan error {
return &o.errors
}
func (o *StreamOutput) SetInputExists(inputExists func() *bool) {
o.inputExists = inputExists
}
func (o *StreamOutput) InputExists() *bool {
if o.Profile() {
profiling.DumpMemory(fmt.Sprintf("memdump-%d.prof", time.Now().Unix()))
}
return o.inputExists()
}
func (o *StreamOutput) EnableProfile() {
o.profile = true
}
func (o *StreamOutput) Profile() bool {
return o.profile
}