diff options
| -rw-r--r-- | server/peer.go | 21 | ||||
| -rw-r--r-- | server/server.go | 35 |
2 files changed, 49 insertions, 7 deletions
diff --git a/server/peer.go b/server/peer.go index cf712b57..52a8205f 100644 --- a/server/peer.go +++ b/server/peer.go @@ -31,16 +31,20 @@ type Peer struct { acceptedConnCh chan *net.TCPConn incoming chan *bgp.BGPMessage outgoing chan *bgp.BGPMessage + inEventCh chan *message + outEventCh chan *message fsm *FSM } -func NewPeer(g config.GlobalType, peer config.NeighborType) *Peer { +func NewPeer(g config.GlobalType, peer config.NeighborType, outEventCh chan *message) *Peer { p := &Peer{ globalConfig: g, peerConfig: peer, acceptedConnCh: make(chan *net.TCPConn), incoming: make(chan *bgp.BGPMessage, 4096), outgoing: make(chan *bgp.BGPMessage, 4096), + inEventCh: make(chan *message, 4096), + outEventCh: outEventCh, } p.fsm = NewFSM(&g, &peer, p.acceptedConnCh, p.incoming, p.outgoing) p.t.Go(p.loop) @@ -70,6 +74,8 @@ func (peer *Peer) loop() error { j, _ := json.Marshal(m) fmt.Println(string(j)) } + case m := <-peer.inEventCh: + fmt.Println(m) } } } @@ -83,3 +89,16 @@ func (peer *Peer) Stop() error { func (peer *Peer) PassConn(conn *net.TCPConn) { peer.acceptedConnCh <- conn } + +func (peer *Peer) SendMessage(msg *message) { + peer.inEventCh <- msg +} + +func (peer *Peer) sendToHub(destination string, event int, data interface{}) { + peer.outEventCh <- &message{ + src: peer.peerConfig.NeighborAddress.String(), + dst: destination, + event: event, + data: data, + } +} diff --git a/server/server.go b/server/server.go index 56fe5960..3f7885a6 100644 --- a/server/server.go +++ b/server/server.go @@ -24,12 +24,20 @@ import ( "strings" ) +type message struct { + src string + dst string + event int + data interface{} +} + type BgpServer struct { bgpConfig config.BgpType globalTypeCh chan config.GlobalType addedPeerCh chan config.NeighborType deletedPeerCh chan config.NeighborType listenPort int + peerMap map[string]*Peer } func NewBgpServer(port int) *BgpServer { @@ -65,14 +73,15 @@ func (server *BgpServer) Serve() { } }() - peerMap := make(map[string]*Peer) + server.peerMap = make(map[string]*Peer) + broadcastCh := make(chan *message) for { f, _ := l.File() select { case conn := <-acceptCh: fmt.Println(conn) remoteAddr := strings.Split(conn.RemoteAddr().String(), ":")[0] - peer, found := peerMap[remoteAddr] + peer, found := server.peerMap[remoteAddr] if found { fmt.Println("found neighbor", remoteAddr) peer.PassConn(conn) @@ -84,20 +93,22 @@ func (server *BgpServer) Serve() { fmt.Println(peer) addr := peer.NeighborAddress.String() SetTcpMD5SigSockopts(int(f.Fd()), addr, peer.AuthPassword) - p := NewPeer(server.bgpConfig.Global, peer) - peerMap[peer.NeighborAddress.String()] = p + p := NewPeer(server.bgpConfig.Global, peer, broadcastCh) + server.peerMap[peer.NeighborAddress.String()] = p case peer := <-server.deletedPeerCh: fmt.Println(peer) addr := peer.NeighborAddress.String() SetTcpMD5SigSockopts(int(f.Fd()), addr, "") - p, found := peerMap[addr] + p, found := server.peerMap[addr] if found { fmt.Println("found neighbor", addr) p.Stop() - delete(peerMap, addr) + delete(server.peerMap, addr) } else { fmt.Println("can't found neighbor", addr) } + case msg := <-broadcastCh: + server.broadcast(msg) } } } @@ -113,3 +124,15 @@ func (server *BgpServer) PeerAdd(peer config.NeighborType) { func (server *BgpServer) PeerDelete(peer config.NeighborType) { server.deletedPeerCh <- peer } + +func (server *BgpServer) broadcast(msg *message) { + for key := range server.peerMap { + if key == msg.src { + continue + } + if msg.dst == "" || msg.dst == key { + peer := server.peerMap[key] + peer.SendMessage(msg) + } + } +} |