zebra client uses the new Watch API

Signed-off-by: FUJITA Tomonori <fujita.tomonori@lab.ntt.co.jp>

1 files changed, 20 insertions, 27 deletions

diff --git a/server/server.go b/server/server.go
index e6d78112..4cd71ad6 100644
--- a/server/server.go
+++ b/server/server.go
@@ -106,6 +106,7 @@ type BgpServer struct {
 	shutdown    bool
 	watchers    *watcherManager
 	watcherMap  map[watchType][]*Watcher
+	zclient     *zebraClient
 }
 
 func NewBgpServer() *BgpServer {
@@ -840,20 +841,26 @@ func (server *BgpServer) SetCollector(c config.Collector) error {
 	return nil
 }
 
-func (server *BgpServer) SetZebraConfig(z config.Zebra) error {
-	if !z.Config.Enabled {
-		return nil
-	}
-	ch := make(chan *GrpcResponse)
-	server.GrpcReqCh <- &GrpcRequest{
-		RequestType: REQ_INITIALIZE_ZEBRA,
-		Data:        &z.Config,
-		ResponseCh:  ch,
-	}
-	if err := (<-ch).Err(); err != nil {
-		return err
+func (s *BgpServer) StartZebraClient(x *config.Zebra) (err error) {
+	ch := make(chan struct{})
+	defer func() { <-ch }()
+
+	s.mgmtCh <- func() {
+		defer close(ch)
+
+		if s.zclient != nil {
+			err = fmt.Errorf("already connected to Zebra")
+		} else {
+			c := x.Config
+
+			protos := make([]string, 0, len(c.RedistributeRouteTypeList))
+			for _, p := range c.RedistributeRouteTypeList {
+				protos = append(protos, string(p))
+			}
+			s.zclient, err = newZebraClient(s, c.Url, protos)
+		}
 	}
-	return nil
+	return err
 }
 
 func (server *BgpServer) SetRpkiConfig(c []config.RpkiServer) error {
@@ -1819,20 +1826,6 @@ func (server *BgpServer) handleGrpc(grpcReq *GrpcRequest) {
 		if len(pathList) > 0 {
 			server.propagateUpdate(nil, pathList)
 		}
-	case REQ_INITIALIZE_ZEBRA:
-		c := grpcReq.Data.(*config.ZebraConfig)
-		protos := make([]string, 0, len(c.RedistributeRouteTypeList))
-		for _, p := range c.RedistributeRouteTypeList {
-			protos = append(protos, string(p))
-		}
-		z, err := newZebraWatcher(server, c.Url, protos)
-		if err == nil {
-			server.watchers.addWatcher(WATCHER_ZEBRA, z)
-		}
-		grpcReq.ResponseCh <- &GrpcResponse{
-			ResponseErr: err,
-		}
-		close(grpcReq.ResponseCh)
 	case REQ_INITIALIZE_COLLECTOR:
 		c := grpcReq.Data.(*config.CollectorConfig)
 		collector, err := NewCollector(server.GrpcReqCh, c.Url, c.DbName, c.TableDumpInterval)