git.opendaylight Code Review - controller.git/blob

   1 /*
   2  * Copyright (c) 2013 Cisco Systems, Inc. and others.  All rights reserved.
   3  * This program and the accompanying materials are made available under the
   4  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
   5  * and is available at http://www.eclipse.org/legal/epl-v10.html
   6  */
   7
   8 package org.opendaylight.controller.sal.connector.remoterpc;
   9
  10 import junit.framework.Assert;
  11 import org.junit.After;
  12 import org.junit.Before;
  13 import org.junit.Test;
  14 import org.opendaylight.controller.sal.connector.remoterpc.utils.MessagingUtil;
  15 import org.opendaylight.controller.sal.core.api.Broker;
  16 import org.zeromq.ZMQ;
  17
  18 import java.util.ArrayList;
  19 import java.util.List;
  20 import java.util.concurrent.ExecutorService;
  21 import java.util.concurrent.Executors;
  22
  23 import static org.mockito.Mockito.mock;
  24
  25 public class ServerRequestHandlerTest {
  26
  27   ServerRequestHandler handler;
  28   ZMQ.Context context;
  29   ExecutorService executorService = Executors.newCachedThreadPool();
  30   private int workerCount = 2;
  31   private String mockDealerAddress = "inproc://rpc-request-handler";
  32   private String mockServerIp = "localhost";
  33   private int mockServerPort = 5554;
  34
  35   @Before
  36   public void setUp() throws Exception {
  37     context = ZMQ.context(1);
  38     String mockServerAddress = mockServerIp + ":" + mockServerPort;
  39     Broker.ProviderSession mockSession = mock(Broker.ProviderSession.class);
  40     handler = new ServerRequestHandler(context, mockSession, workerCount, mockDealerAddress, mockServerAddress);
  41     handler.start();
  42   }
  43
  44   @After
  45   public void tearDown() throws Exception {
  46     executorService.shutdown();
  47     MessagingUtil.closeZmqContext(context);
  48     handler.close();
  49   }
  50
  51   @Test
  52   public void testStart() throws Exception {
  53     //should start workers == workerCount
  54     Assert.assertEquals(workerCount, handler.getWorkerPool().getPoolSize());
  55
  56     //killing a thread should recreate another one
  57
  58     //start router-dealer bridge
  59     executorService.execute(MessagingUtil.createRouterDealerBridge(context, mockDealerAddress, mockServerPort));
  60     Thread.sleep(1000); //give sometime for socket initialization
  61
  62     //this will kill the thread
  63     final String WORKER_THREAD_NAME = "remote-rpc-worker";
  64     interruptAThreadWithName(WORKER_THREAD_NAME);
  65
  66     //send 4 message to router
  67     for (int i = 0; i < 4; i++)
  68       executorService.execute(MessagingUtil.sendAnEmptyMessage(context, "tcp://" + mockServerIp + ":" + mockServerPort));
  69
  70     //worker pool size should not change.
  71     Assert.assertEquals(workerCount, handler.getWorkerPool().getPoolSize());
  72
  73     Thread.sleep(10000); //wait for processing to complete
  74   }
  75
  76   @Test
  77   public void testClose() throws Exception {
  78
  79   }
  80
  81   /**
  82    * Interrupts the first thread found whose name starts with the provided name
  83    *
  84    * @param name
  85    */
  86   private void interruptAThreadWithName(String name) {
  87     List<Thread> workerThreads = findThreadsWithName(name);
  88     if (workerThreads.size() > 0) workerThreads.get(0).interrupt();
  89   }
  90
  91   /**
  92    * Find all threads that start with the given name
  93    *
  94    * @param name
  95    * @return
  96    */
  97   private List<Thread> findThreadsWithName(String name) {
  98     Thread[] threads = new Thread[Thread.activeCount()];
  99     Thread.enumerate(threads);
 100
 101     List<Thread> foundThreads = new ArrayList();
 102     for (Thread t : threads) {
 103       if (t.getName().startsWith(name))
 104         foundThreads.add(t);
 105     }
 106
 107     return foundThreads;
 108   }
 109 }