-
Notifications
You must be signed in to change notification settings - Fork 1
/
standard.go
89 lines (76 loc) · 2.12 KB
/
standard.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
package imparse
import (
"context"
"github.com/txchat/imparse/proto/auth"
)
//
type Exec interface {
Transport(ctx context.Context, id int64, key, from, target string, ch Channel, frameType FrameType, data []byte) error
RevAck(ctx context.Context, id int64, keys []string, data []byte) error
}
type Filter func(ctx context.Context, frame Frame) error
type ActionFilter interface {
FrameFilters(tp FrameType) []Filter
}
//implement
type StandardAnswer struct {
cache Cache
exec Exec
trace Trace
frameFilters map[FrameType][]Filter
}
func NewStandardAnswer(cache Cache, exec Exec, trace Trace, filters map[FrameType][]Filter) *StandardAnswer {
return &StandardAnswer{
cache: cache,
exec: exec,
trace: trace,
frameFilters: filters,
}
}
func (s *StandardAnswer) FrameFilters(tp FrameType) []Filter {
if s.frameFilters == nil {
s.frameFilters = make(map[FrameType][]Filter)
}
return s.frameFilters[tp]
}
func (s *StandardAnswer) Check(ctx context.Context, checker Checker, frame Frame) error {
if s.trace != nil {
finish, _ := s.trace.StartSpanFromContext(ctx, "Check")
defer finish()
}
return checker.CheckFrame(frame)
}
func (s *StandardAnswer) Filter(ctx context.Context, frame Frame) (uint64, error) {
if s.trace != nil {
finish, _ := s.trace.StartSpanFromContext(ctx, "Filter")
defer finish()
}
fs := s.FrameFilters(frame.Type())
return frame.Filter(ctx, s.cache, fs...)
}
func (s *StandardAnswer) Transport(ctx context.Context, frame Frame) error {
if s.trace != nil {
finish, _ := s.trace.StartSpanFromContext(ctx, "Transport")
defer finish()
}
return frame.Transport(ctx, s.exec)
}
func (s *StandardAnswer) Ack(ctx context.Context, frame Frame) (int64, error) {
if s.trace != nil {
finish, _ := s.trace.StartSpanFromContext(ctx, "Ack")
defer finish()
}
return frame.Ack(ctx, s.exec)
}
//
type StandardStorage struct {
db DB
}
func NewStandardStorage(db DB) *StandardStorage {
return &StandardStorage{
db: db,
}
}
func (s *StandardStorage) SaveMsg(ctx context.Context, deviceType auth.Device, frame Frame) error {
return s.db.SaveMsg(deviceType, frame)
}