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
109
110
111
112
113
114
115
116
117
118
119
120
121
|
// Copyright (C) 2016 Nippon Telegraph and Telephone Corporation.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
// implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package server
import (
api "github.com/osrg/gobgp/api"
"github.com/osrg/gobgp/packet"
"github.com/osrg/gobgp/table"
"gopkg.in/tomb.v2"
)
type grpcIncomingWatcher struct {
t tomb.Tomb
ch chan watcherEvent
ctlCh chan *GrpcRequest
reqs []*GrpcRequest
}
func (w *grpcIncomingWatcher) notify(t watcherEventType) chan watcherEvent {
if t == WATCHER_EVENT_UPDATE_MSG || t == WATCHER_EVENT_POST_POLICY_UPDATE_MSG {
return w.ch
}
return nil
}
func (w *grpcIncomingWatcher) stop() {
w.t.Kill(nil)
}
func (w *grpcIncomingWatcher) watchingEventTypes() []watcherEventType {
pre := false
post := false
for _, req := range w.reqs {
if req.Data.(*api.Table).PostPolicy {
post = true
} else {
pre = true
}
}
types := make([]watcherEventType, 0, 2)
if pre {
types = append(types, WATCHER_EVENT_UPDATE_MSG)
}
if post {
types = append(types, WATCHER_EVENT_POST_POLICY_UPDATE_MSG)
}
return types
}
func (w *grpcIncomingWatcher) loop() error {
for {
select {
case <-w.t.Dying():
for _, req := range w.reqs {
close(req.ResponseCh)
}
return nil
case req := <-w.ctlCh:
w.reqs = append(w.reqs, req)
case ev := <-w.ch:
msg := ev.(*watcherEventUpdateMsg)
for _, path := range msg.pathList {
remains := make([]*GrpcRequest, 0, len(w.reqs))
result := &GrpcResponse{
Data: &api.Destination{
Prefix: path.GetNlri().String(),
Paths: []*api.Path{path.ToApiStruct(table.GLOBAL_RIB_NAME)},
},
}
for _, req := range w.reqs {
select {
case <-req.EndCh:
continue
default:
}
remains = append(remains, req)
if req.RouteFamily != bgp.RouteFamily(0) && req.RouteFamily != path.GetRouteFamily() {
continue
}
if req.Name != "" && req.Name != path.GetSource().Address.String() {
continue
}
req.ResponseCh <- result
}
w.reqs = remains
}
}
}
}
func (w *grpcIncomingWatcher) restart(string) error {
return nil
}
func (w *grpcIncomingWatcher) addRequest(req *GrpcRequest) error {
w.ctlCh <- req
return nil
}
func newGrpcIncomingWatcher() (*grpcIncomingWatcher, error) {
w := &grpcIncomingWatcher{
ch: make(chan watcherEvent),
ctlCh: make(chan *GrpcRequest),
reqs: make([]*GrpcRequest, 0, 16),
}
w.t.Go(w.loop)
return w, nil
}
|
