bmp/bmp-impl/src/test/java/org/opendaylight/protocol/bmp/impl/app/BmpMonitorImplTest.java

   1 /*
   2  * Copyright (c) 2015 Cisco Systems, Inc. and others.  All rights reserved.
   3  *
   4  * This program and the accompanying materials are made available under the
   5  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
   6  * and is available at http://www.eclipse.org/legal/epl-v10.html
   7  */
   8
   9 package org.opendaylight.protocol.bmp.impl.app;
  10
  11 import static org.junit.Assert.assertEquals;
  12 import static org.junit.Assert.assertFalse;
  13 import static org.junit.Assert.assertNotNull;
  14 import static org.junit.Assert.assertNull;
  15 import static org.junit.Assert.assertTrue;
  16 import static org.mockito.Matchers.any;
  17 import static org.mockito.Mockito.doAnswer;
  18 import static org.opendaylight.protocol.bmp.parser.message.TestUtil.createRouteMonMsgWithEndOfRibMarker;
  19 import static org.opendaylight.protocol.bmp.parser.message.TestUtil.createRouteMonitMsg;
  20 import static org.opendaylight.protocol.util.CheckUtil.readDataOperational;
  21 import static org.opendaylight.protocol.util.CheckUtil.waitFutureSuccess;
  22
  23 import com.google.common.net.InetAddresses;
  24 import io.netty.bootstrap.Bootstrap;
  25 import io.netty.channel.Channel;
  26 import io.netty.channel.ChannelFuture;
  27 import io.netty.channel.ChannelInitializer;
  28 import io.netty.channel.ChannelOption;
  29 import io.netty.channel.EventLoopGroup;
  30 import io.netty.channel.epoll.Epoll;
  31 import io.netty.channel.epoll.EpollEventLoopGroup;
  32 import io.netty.channel.epoll.EpollSocketChannel;
  33 import io.netty.channel.nio.NioEventLoopGroup;
  34 import io.netty.channel.socket.SocketChannel;
  35 import io.netty.channel.socket.nio.NioSocketChannel;
  36 import java.net.InetSocketAddress;
  37 import java.util.List;
  38 import org.junit.After;
  39 import org.junit.Before;
  40 import org.junit.Test;
  41 import org.mockito.Mock;
  42 import org.mockito.MockitoAnnotations;
  43 import org.opendaylight.controller.md.sal.binding.impl.BindingToNormalizedNodeCodec;
  44 import org.opendaylight.controller.md.sal.binding.test.AbstractConcurrentDataBrokerTest;
  45 import org.opendaylight.controller.md.sal.binding.test.AbstractDataBrokerTestCustomizer;
  46 import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
  47 import org.opendaylight.controller.md.sal.common.api.data.ReadFailedException;
  48 import org.opendaylight.controller.md.sal.dom.api.DOMDataWriteTransaction;
  49 import org.opendaylight.mdsal.singleton.common.api.ClusterSingletonService;
  50 import org.opendaylight.mdsal.singleton.common.api.ClusterSingletonServiceProvider;
  51 import org.opendaylight.mdsal.singleton.common.api.ClusterSingletonServiceRegistration;
  52 import org.opendaylight.protocol.bgp.inet.RIBActivator;
  53 import org.opendaylight.protocol.bgp.parser.impl.BGPActivator;
  54 import org.opendaylight.protocol.bgp.parser.spi.BGPExtensionProviderContext;
  55 import org.opendaylight.protocol.bgp.parser.spi.pojo.SimpleBGPExtensionProviderContext;
  56 import org.opendaylight.protocol.bgp.rib.spi.RIBExtensionProviderContext;
  57 import org.opendaylight.protocol.bgp.rib.spi.SimpleRIBExtensionProviderContext;
  58 import org.opendaylight.protocol.bmp.api.BmpDispatcher;
  59 import org.opendaylight.protocol.bmp.impl.BmpDispatcherImpl;
  60 import org.opendaylight.protocol.bmp.impl.BmpHandlerFactory;
  61 import org.opendaylight.protocol.bmp.impl.config.BmpDeployerDependencies;
  62 import org.opendaylight.protocol.bmp.impl.session.DefaultBmpSessionFactory;
  63 import org.opendaylight.protocol.bmp.impl.spi.BmpMonitoringStation;
  64 import org.opendaylight.protocol.bmp.parser.BmpActivator;
  65 import org.opendaylight.protocol.bmp.parser.message.TestUtil;
  66 import org.opendaylight.protocol.bmp.spi.registry.BmpMessageRegistry;
  67 import org.opendaylight.protocol.bmp.spi.registry.SimpleBmpExtensionProviderContext;
  68 import org.opendaylight.protocol.util.CheckUtil;
  69 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.inet.types.rev130715.IpAddress;
  70 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.inet.types.rev130715.Ipv4Address;
  71 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.inet.rev180329.bmp.monitor.monitor.router.peer.pre.policy.rib.tables.routes.Ipv4RoutesCase;
  72 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.rib.rev180329.PeerId;
  73 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.rib.rev180329.rib.Tables;
  74 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.types.rev180329.Ipv4AddressFamily;
  75 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.types.rev180329.UnicastSubsequentAddressFamily;
  76 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bmp.message.rev180329.AdjRibInType;
  77 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bmp.message.rev180329.PeerType;
  78 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bmp.message.rev180329.RouteMirroringMessage;
  79 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bmp.message.rev180329.StatsReportsMessage;
  80 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bmp.message.rev180329.stat.Tlvs;
  81 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bmp.monitor.rev180329.BmpMonitor;
  82 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bmp.monitor.rev180329.MonitorId;
  83 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bmp.monitor.rev180329.RouterId;
  84 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bmp.monitor.rev180329.Status;
  85 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bmp.monitor.rev180329.bmp.monitor.Monitor;
  86 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bmp.monitor.rev180329.bmp.monitor.MonitorKey;
  87 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bmp.monitor.rev180329.peers.Peer;
  88 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bmp.monitor.rev180329.peers.PeerKey;
  89 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bmp.monitor.rev180329.peers.peer.Mirrors;
  90 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bmp.monitor.rev180329.peers.peer.PeerSession;
  91 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bmp.monitor.rev180329.peers.peer.PostPolicyRib;
  92 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bmp.monitor.rev180329.peers.peer.PrePolicyRib;
  93 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bmp.monitor.rev180329.peers.peer.Stats;
  94 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bmp.monitor.rev180329.routers.Router;
  95 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bmp.monitor.rev180329.routers.RouterKey;
  96 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
  97 import org.opendaylight.yangtools.yang.binding.KeyedInstanceIdentifier;
  98 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
  99 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier.NodeIdentifier;
 100 import org.opendaylight.yangtools.yang.data.api.schema.ContainerNode;
 101 import org.opendaylight.yangtools.yang.data.impl.schema.Builders;
 102 import org.opendaylight.yangtools.yang.data.impl.schema.ImmutableNodes;
 103
 104 public class BmpMonitorImplTest extends AbstractConcurrentDataBrokerTest {
 105     // the local port and address where the monitor (ODL) will listen for incoming BMP request
 106     private static final int MONITOR_LOCAL_PORT = 12345;
 107     private static final String MONITOR_LOCAL_ADDRESS = "127.0.0.10";
 108     private static final String MONITOR_LOCAL_ADDRESS_2 = "127.0.0.11";
 109     // the router (monitee) address where we are going to simulate a BMP request from
 110     private static final String REMOTE_ROUTER_ADDRESS_1 = "127.0.0.12";
 111     private static final String REMOTE_ROUTER_ADDRESS_2 = "127.0.0.13";
 112     private static final Ipv4Address PEER1 = new Ipv4Address("20.20.20.20");
 113     private static final MonitorId MONITOR_ID = new MonitorId("monitor");
 114     private static final KeyedInstanceIdentifier<Monitor, MonitorKey> MONITOR_IID = InstanceIdentifier
 115         .create(BmpMonitor.class).child(Monitor.class, new MonitorKey(MONITOR_ID));
 116     private static final PeerId PEER_ID = new PeerId(PEER1.getValue());
 117     private static final InstanceIdentifier<BmpMonitor> BMP_II = InstanceIdentifier.create(BmpMonitor.class);
 118     private BindingToNormalizedNodeCodec mappingService;
 119     private RIBActivator ribActivator;
 120     private BGPActivator bgpActivator;
 121     private BmpActivator bmpActivator;
 122     private BmpDispatcher dispatcher;
 123     private BmpMonitoringStation bmpApp;
 124     private BmpMessageRegistry msgRegistry;
 125     private RIBExtensionProviderContext ribExtension;
 126     private ClusterSingletonService singletonService;
 127     private ClusterSingletonService singletonService2;
 128     @Mock
 129     private ClusterSingletonServiceRegistration singletonServiceRegistration;
 130     @Mock
 131     private ClusterSingletonServiceRegistration singletonServiceRegistration2;
 132     @Mock
 133     private ClusterSingletonServiceProvider clusterSSProv;
 134     @Mock
 135     private ClusterSingletonServiceProvider clusterSSProv2;
 136
 137     @Before
 138     public void setUp() throws Exception {
 139         super.setup();
 140
 141         MockitoAnnotations.initMocks(this);
 142
 143         doAnswer(invocationOnMock -> {
 144             BmpMonitorImplTest.this.singletonService = (ClusterSingletonService) invocationOnMock.getArguments()[0];
 145             this.singletonService.instantiateServiceInstance();
 146             return BmpMonitorImplTest.this.singletonServiceRegistration;
 147         }).when(this.clusterSSProv).registerClusterSingletonService(any(ClusterSingletonService.class));
 148
 149         doAnswer(invocationOnMock -> BmpMonitorImplTest.this.singletonService.closeServiceInstance())
 150             .when(this.singletonServiceRegistration).close();
 151
 152         doAnswer(invocationOnMock -> {
 153             this.singletonService2 = (ClusterSingletonService) invocationOnMock.getArguments()[0];
 154             this.singletonService2.instantiateServiceInstance();
 155             return BmpMonitorImplTest.this.singletonServiceRegistration2;
 156         }).when(this.clusterSSProv2).registerClusterSingletonService(any(ClusterSingletonService.class));
 157
 158         doAnswer(invocationOnMock -> BmpMonitorImplTest.this.singletonService2.closeServiceInstance())
 159             .when(this.singletonServiceRegistration2).close();
 160
 161         this.mappingService.onGlobalContextUpdated(getSchemaContext());
 162         this.ribActivator = new RIBActivator();
 163         this.ribExtension = new SimpleRIBExtensionProviderContext();
 164         this.ribActivator.startRIBExtensionProvider(this.ribExtension, this.mappingService);
 165
 166         this.bgpActivator = new BGPActivator();
 167         final BGPExtensionProviderContext context = new SimpleBGPExtensionProviderContext();
 168         this.bgpActivator.start(context);
 169         final SimpleBmpExtensionProviderContext ctx = new SimpleBmpExtensionProviderContext();
 170         this.bmpActivator = new BmpActivator(context);
 171         this.bmpActivator.start(ctx);
 172         this.msgRegistry = ctx.getBmpMessageRegistry();
 173
 174         this.dispatcher = new BmpDispatcherImpl(new NioEventLoopGroup(), new NioEventLoopGroup(),
 175             ctx.getBmpMessageRegistry(), new DefaultBmpSessionFactory());
 176
 177         final InetSocketAddress inetAddress = new InetSocketAddress(InetAddresses.forString(MONITOR_LOCAL_ADDRESS),
 178             MONITOR_LOCAL_PORT);
 179
 180         final DOMDataWriteTransaction wTx = getDomBroker().newWriteOnlyTransaction();
 181         final ContainerNode parentNode = Builders.containerBuilder().withNodeIdentifier(
 182                 new NodeIdentifier(BmpMonitor.QNAME))
 183                 .addChild(ImmutableNodes.mapNodeBuilder(Monitor.QNAME).build()).build();
 184         wTx.merge(LogicalDatastoreType.OPERATIONAL, YangInstanceIdentifier.of(BmpMonitor.QNAME), parentNode);
 185         wTx.commit().get();
 186
 187         final BmpDeployerDependencies bmpDependecies = new BmpDeployerDependencies(getDataBroker(), getDomBroker(),
 188             this.ribExtension, this.mappingService.getCodecFactory(), getSchemaContext(), this.clusterSSProv);
 189         this.bmpApp = new BmpMonitoringStationImpl(bmpDependecies, this.dispatcher, MONITOR_ID, inetAddress, null);
 190         readDataOperational(getDataBroker(), BMP_II, monitor -> {
 191             assertEquals(1, monitor.getMonitor().size());
 192             final Monitor bmpMonitor = monitor.getMonitor().get(0);
 193             assertEquals(MONITOR_ID, bmpMonitor.getMonitorId());
 194             assertEquals(0, bmpMonitor.getRouter().size());
 195             assertEquals(MONITOR_ID, bmpMonitor.getMonitorId());
 196             assertEquals(0, bmpMonitor.getRouter().size());
 197             return monitor;
 198         });
 199     }
 200
 201     @Override
 202     protected final AbstractDataBrokerTestCustomizer createDataBrokerTestCustomizer() {
 203         final AbstractDataBrokerTestCustomizer customizer = super.createDataBrokerTestCustomizer();
 204         this.mappingService = customizer.getBindingToNormalized();
 205         return customizer;
 206     }
 207
 208     @After
 209     public void tearDown() throws Exception {
 210         this.ribActivator.close();
 211         this.bgpActivator.close();
 212         this.bmpActivator.close();
 213         this.dispatcher.close();
 214         this.bmpApp.close();
 215         this.mappingService.close();
 216
 217         readDataOperational(getDataBroker(), BMP_II, monitor -> {
 218             assertTrue(monitor.getMonitor().isEmpty());
 219             return monitor;
 220         });
 221     }
 222
 223     @Test
 224     public void testRouterMonitoring() throws Exception {
 225         // first test if a single router monitoring is working
 226         final Channel channel1 = testMonitoringStation(REMOTE_ROUTER_ADDRESS_1);
 227         readDataOperational(getDataBroker(), MONITOR_IID, monitor -> {
 228             assertEquals(1, monitor.getRouter().size());
 229             return monitor;
 230         });
 231
 232         final Channel channel2 = testMonitoringStation(REMOTE_ROUTER_ADDRESS_2);
 233         readDataOperational(getDataBroker(), MONITOR_IID, monitor -> {
 234             assertEquals(2, monitor.getRouter().size());
 235             return monitor;
 236         });
 237
 238         // initiate another BMP request from router 1, create a redundant connection
 239         // we expect the connection to be closed
 240         final Channel channel3 = connectTestClient(REMOTE_ROUTER_ADDRESS_1, this.msgRegistry);
 241
 242
 243         // channel 1 should still be open, while channel3 should be closed
 244         CheckUtil.checkEquals(() -> assertTrue(channel1.isOpen()));
 245         CheckUtil.checkEquals(() -> assertFalse(channel3.isOpen()));
 246
 247         // now if we close the channel 1 and try it again, it should succeed
 248         waitFutureSuccess(channel1.close());
 249
 250         // channel 2 is still open
 251         readDataOperational(getDataBroker(), MONITOR_IID, monitor -> {
 252             assertEquals(1, monitor.getRouter().size());
 253             return monitor;
 254         });
 255
 256         final Channel channel4 = testMonitoringStation(REMOTE_ROUTER_ADDRESS_1);
 257         readDataOperational(getDataBroker(), MONITOR_IID, monitor -> {
 258             assertEquals(2, monitor.getRouter().size());
 259             return monitor;
 260         });
 261
 262         // close all channel altogether
 263         waitFutureSuccess(channel2.close());
 264         Thread.sleep(100);
 265
 266         // sleep for a while to avoid intermittent InMemoryDataTree modification conflict
 267         waitFutureSuccess(channel4.close());
 268
 269         readDataOperational(getDataBroker(), MONITOR_IID, monitor -> {
 270             assertEquals(0, monitor.getRouter().size());
 271             return monitor;
 272         });
 273     }
 274
 275     private static void waitWriteAndFlushSuccess(final ChannelFuture channelFuture) {
 276         waitFutureSuccess(channelFuture);
 277     }
 278
 279     private Channel testMonitoringStation(final String remoteRouterIpAddr) throws InterruptedException,
 280             ReadFailedException {
 281         final Channel channel = connectTestClient(remoteRouterIpAddr, this.msgRegistry);
 282         final RouterId routerId = getRouterId(remoteRouterIpAddr);
 283
 284         readDataOperational(getDataBroker(), MONITOR_IID, monitor -> {
 285             assertFalse(monitor.getRouter().isEmpty());
 286             // now find the current router instance
 287             Router router = null;
 288             for (final Router r : monitor.getRouter()) {
 289                 if (routerId.equals(r.getRouterId())) {
 290                     router = r;
 291                     break;
 292                 }
 293             }
 294             assertNotNull(router);
 295             assertEquals(Status.Down, router.getStatus());
 296             assertTrue(router.getPeer().isEmpty());
 297             return router;
 298         });
 299
 300         waitWriteAndFlushSuccess(channel.writeAndFlush(TestUtil
 301                 .createInitMsg("description", "name", "some info")));
 302
 303         readDataOperational(getDataBroker(), MONITOR_IID, monitor -> {
 304             assertFalse(monitor.getRouter().isEmpty());
 305             Router retRouter = null;
 306             for (final Router r : monitor.getRouter()) {
 307                 if (routerId.equals(r.getRouterId())) {
 308                     retRouter = r;
 309                     break;
 310                 }
 311             }
 312
 313             assertEquals("some info;", retRouter.getInfo());
 314             assertEquals("name", retRouter.getName());
 315             assertEquals("description", retRouter.getDescription());
 316             assertEquals(routerId, retRouter.getRouterId());
 317             assertTrue(retRouter.getPeer().isEmpty());
 318             assertEquals(Status.Up, retRouter.getStatus());
 319             return retRouter;
 320         });
 321
 322         waitWriteAndFlushSuccess(channel.writeAndFlush(TestUtil.createPeerUpNotification(PEER1, true)));
 323         final KeyedInstanceIdentifier<Router, RouterKey> routerIId =
 324                 MONITOR_IID.child(Router.class, new RouterKey(routerId));
 325
 326         readDataOperational(getDataBroker(), routerIId, router -> {
 327             final List<Peer> peers = router.getPeer();
 328             assertEquals(1, peers.size());
 329             final Peer peer = peers.get(0);
 330             assertEquals(PeerType.Global, peer.getType());
 331             assertEquals(PEER_ID, peer.getPeerId());
 332             assertEquals(PEER1, peer.getBgpId());
 333             assertEquals(TestUtil.IPV4_ADDRESS_10, peer.getAddress().getIpv4Address());
 334             assertEquals(TestUtil.PEER_AS, peer.getAs());
 335             assertNull(peer.getPeerDistinguisher());
 336             assertNull(peer.getStats());
 337
 338             assertNotNull(peer.getPrePolicyRib());
 339             assertEquals(1, peer.getPrePolicyRib().getTables().size());
 340             final Tables prePolicyTable = peer.getPrePolicyRib().getTables().get(0);
 341             assertEquals(Ipv4AddressFamily.class, prePolicyTable.getAfi());
 342             assertEquals(UnicastSubsequentAddressFamily.class, prePolicyTable.getSafi());
 343             assertFalse(prePolicyTable.getAttributes().isUptodate());
 344             assertNotNull(prePolicyTable.getRoutes());
 345
 346             assertNotNull(peer.getPostPolicyRib());
 347             assertEquals(1, peer.getPostPolicyRib().getTables().size());
 348             final Tables postPolicyTable = peer.getPrePolicyRib().getTables().get(0);
 349             assertEquals(Ipv4AddressFamily.class, postPolicyTable.getAfi());
 350             assertEquals(UnicastSubsequentAddressFamily.class, postPolicyTable.getSafi());
 351             assertFalse(postPolicyTable.getAttributes().isUptodate());
 352             assertNotNull(postPolicyTable.getRoutes());
 353
 354             assertNotNull(peer.getPeerSession());
 355             final PeerSession peerSession = peer.getPeerSession();
 356             assertEquals(TestUtil.IPV4_ADDRESS_10, peerSession.getLocalAddress().getIpv4Address());
 357             assertEquals(TestUtil.PEER_LOCAL_PORT, peerSession.getLocalPort());
 358             assertEquals(TestUtil.PEER_REMOTE_PORT, peerSession.getRemotePort());
 359             assertEquals(Status.Up, peerSession.getStatus());
 360             assertNotNull(peerSession.getReceivedOpen());
 361             assertNotNull(peerSession.getSentOpen());
 362             return router;
 363         });
 364
 365
 366         final StatsReportsMessage statsMsg = TestUtil.createStatsReportMsg(PEER1);
 367         waitWriteAndFlushSuccess(channel.writeAndFlush(statsMsg));
 368         final KeyedInstanceIdentifier<Peer, PeerKey> peerIId = routerIId.child(Peer.class, new PeerKey(PEER_ID));
 369
 370         readDataOperational(getDataBroker(), peerIId.child(Stats.class), peerStats -> {
 371             assertNotNull(peerStats.getTimestampSec());
 372             final Tlvs tlvs = statsMsg.getTlvs();
 373             assertEquals(tlvs.getAdjRibsInRoutesTlv().getCount(), peerStats.getAdjRibsInRoutes());
 374             assertEquals(tlvs.getDuplicatePrefixAdvertisementsTlv().getCount(),
 375                     peerStats.getDuplicatePrefixAdvertisements());
 376             assertEquals(tlvs.getDuplicateWithdrawsTlv().getCount(), peerStats.getDuplicateWithdraws());
 377             assertEquals(tlvs.getInvalidatedAsConfedLoopTlv().getCount(), peerStats.getInvalidatedAsConfedLoop());
 378             assertEquals(tlvs.getInvalidatedAsPathLoopTlv().getCount(), peerStats.getInvalidatedAsPathLoop());
 379             assertEquals(tlvs.getInvalidatedClusterListLoopTlv().getCount(),
 380                     peerStats.getInvalidatedClusterListLoop());
 381             assertEquals(tlvs.getInvalidatedOriginatorIdTlv().getCount(), peerStats.getInvalidatedOriginatorId());
 382             assertEquals(tlvs.getLocRibRoutesTlv().getCount(), peerStats.getLocRibRoutes());
 383             assertEquals(tlvs.getRejectedPrefixesTlv().getCount(), peerStats.getRejectedPrefixes());
 384             assertEquals(tlvs.getPerAfiSafiAdjRibInTlv().getCount().toString(),
 385                     peerStats.getPerAfiSafiAdjRibInRoutes().getAfiSafi().get(0).getCount().toString());
 386             assertEquals(tlvs.getPerAfiSafiLocRibTlv().getCount().toString(),
 387                     peerStats.getPerAfiSafiLocRibRoutes().getAfiSafi().get(0).getCount().toString());
 388             return peerStats;
 389         });
 390
 391         // route mirror message test
 392         final RouteMirroringMessage routeMirrorMsg = TestUtil.createRouteMirrorMsg(PEER1);
 393         waitWriteAndFlushSuccess(channel.writeAndFlush(routeMirrorMsg));
 394
 395         readDataOperational(getDataBroker(), peerIId.child(Mirrors.class), routeMirrors -> {
 396             assertNotNull(routeMirrors.getTimestampSec());
 397             return routeMirrors;
 398         });
 399
 400         waitWriteAndFlushSuccess(channel.writeAndFlush(createRouteMonitMsg(false, PEER1,
 401                 AdjRibInType.PrePolicy)));
 402         waitWriteAndFlushSuccess(channel.writeAndFlush(createRouteMonMsgWithEndOfRibMarker(PEER1,
 403                 AdjRibInType.PrePolicy)));
 404
 405         readDataOperational(getDataBroker(), peerIId.child(PrePolicyRib.class), prePolicyRib -> {
 406             assertTrue(!prePolicyRib.getTables().isEmpty());
 407             final Tables tables = prePolicyRib.getTables().get(0);
 408             assertTrue(tables.getAttributes().isUptodate());
 409             assertEquals(3, ((Ipv4RoutesCase) tables.getRoutes()).getIpv4Routes().getIpv4Route().size());
 410             return tables;
 411         });
 412
 413         waitWriteAndFlushSuccess(channel.writeAndFlush(createRouteMonitMsg(false, PEER1,
 414                 AdjRibInType.PostPolicy)));
 415         waitWriteAndFlushSuccess(channel.writeAndFlush(createRouteMonMsgWithEndOfRibMarker(PEER1,
 416                 AdjRibInType.PostPolicy)));
 417
 418         readDataOperational(getDataBroker(), peerIId.child(PostPolicyRib.class), postPolicyRib -> {
 419             assertTrue(!postPolicyRib.getTables().isEmpty());
 420             final Tables tables = postPolicyRib.getTables().get(0);
 421             assertTrue(tables.getAttributes().isUptodate());
 422             assertEquals(3, ((org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.bgp.inet
 423                     .rev180329.bmp.monitor.monitor.router.peer.post.policy.rib.tables.routes.Ipv4RoutesCase)
 424                     tables.getRoutes()).getIpv4Routes().getIpv4Route().size());
 425             return tables;
 426         });
 427
 428         waitWriteAndFlushSuccess(channel.writeAndFlush(TestUtil.createPeerDownNotification(PEER1)));
 429
 430         readDataOperational(getDataBroker(), routerIId, router -> {
 431             final List<Peer> peersAfterDown = router.getPeer();
 432             assertTrue(peersAfterDown.isEmpty());
 433             return router;
 434         });
 435
 436         return channel;
 437     }
 438
 439     @Test
 440     public void deploySecondInstance() throws Exception {
 441         final BmpDeployerDependencies bmpDependecies = new BmpDeployerDependencies(getDataBroker(), getDomBroker(),
 442             this.ribExtension, this.mappingService.getCodecFactory(), getSchemaContext(), this.clusterSSProv2);
 443
 444         final BmpMonitoringStation monitoringStation2 = new BmpMonitoringStationImpl(bmpDependecies,
 445             this.dispatcher, new MonitorId("monitor2"),
 446                 new InetSocketAddress(InetAddresses.forString(MONITOR_LOCAL_ADDRESS_2), MONITOR_LOCAL_PORT), null);
 447
 448         readDataOperational(getDataBroker(), BMP_II, monitor -> {
 449             assertEquals(2, monitor.getMonitor().size());
 450             return monitor;
 451         });
 452
 453         monitoringStation2.close();
 454     }
 455
 456     private static Channel connectTestClient(final String routerIp, final BmpMessageRegistry msgRegistry)
 457             throws InterruptedException {
 458         final BmpHandlerFactory hf = new BmpHandlerFactory(msgRegistry);
 459         final Bootstrap b = new Bootstrap();
 460         final EventLoopGroup workerGroup;
 461         if (Epoll.isAvailable()) {
 462             b.channel(EpollSocketChannel.class);
 463             workerGroup = new EpollEventLoopGroup();
 464         } else {
 465             b.channel(NioSocketChannel.class);
 466             workerGroup = new NioEventLoopGroup();
 467         }
 468         b.group(workerGroup);
 469         b.option(ChannelOption.SO_KEEPALIVE, true);
 470         b.handler(new ChannelInitializer<SocketChannel>() {
 471             @Override
 472             protected void initChannel(final SocketChannel ch) {
 473                 ch.pipeline().addLast(hf.getDecoders());
 474                 ch.pipeline().addLast(hf.getEncoders());
 475             }
 476         });
 477         b.localAddress(new InetSocketAddress(routerIp, 0));
 478         b.option(ChannelOption.SO_REUSEADDR, true);
 479         final ChannelFuture future = b.connect(new InetSocketAddress(MONITOR_LOCAL_ADDRESS, MONITOR_LOCAL_PORT)).sync();
 480         waitFutureSuccess(future);
 481         return future.channel();
 482     }
 483
 484     private static RouterId getRouterId(final String routerIp) {
 485         return new RouterId(new IpAddress(new Ipv4Address(routerIp)));
 486     }
 487 }