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