opendaylight/md-sal/sal-remoterpc-connector/implementation/src/test/java/org/opendaylight/controller/sal/connector/remoterpc/ClientRequestHandlerTest.java

   1 /*
   2  * Copyright (c) 2014 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 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.dto.Message;
  15 import org.opendaylight.controller.sal.connector.remoterpc.utils.MessagingUtil;
  16 import org.slf4j.Logger;
  17 import org.slf4j.LoggerFactory;
  18 import org.zeromq.ZMQ;
  19
  20 import java.io.IOException;
  21 import java.util.concurrent.*;
  22
  23 /**
  24  *
  25  */
  26 public class ClientRequestHandlerTest {
  27
  28   private final Logger _logger = LoggerFactory.getLogger(ClientRequestHandlerTest.class);
  29
  30   ZMQ.Context context;
  31   ExecutorService serverThread;
  32   final String SERVER_ADDRESS = "localhost:5553";
  33
  34   ClientRequestHandler handler;
  35
  36   @Before
  37   public void setUp() throws Exception {
  38     context = ZMQ.context(1);
  39     serverThread = Executors.newCachedThreadPool();
  40     handler = new ClientRequestHandler(context);
  41   }
  42
  43   @After
  44   public void tearDown() throws Exception {
  45     serverThread.shutdown();
  46     MessagingUtil.closeZmqContext(context);
  47     handler.close();
  48   }
  49
  50  @Test
  51   public void handle_SingleRemote_ShouldReturnResponse() throws Exception {
  52     serverThread.execute(MessagingUtil.startReplyServer(context, SERVER_ADDRESS, 1));
  53     Message request = new Message();
  54     request.setRecipient(SERVER_ADDRESS);
  55     Message response = handleMessageWithTimeout(request);
  56     Assert.assertNotNull(response);
  57     //should be connected to only 1 remote server
  58     Assert.assertEquals(1, handler.getWorkerCount());
  59     Assert.assertEquals(response.getRecipient(), SERVER_ADDRESS);
  60   }
  61
  62  // @Test
  63   public void handle_MultiRemote_ShouldReturnResponses() throws Exception {
  64     ExecutorService threadPool = Executors.newCachedThreadPool();
  65     final int port = 5555;
  66     String serverAddress = null;
  67     for (int i = 0; i < 5; i++) {
  68       serverAddress = "localhost:" + (port + i);
  69       serverThread.execute(MessagingUtil.startReplyServer(context, serverAddress, 1));
  70       threadPool.execute(createEmptyMessageTaskAndHandle(handler, serverAddress));
  71     }
  72     Thread.currentThread().sleep(5000);//wait for all messages to get processed
  73     //should be connected to 5 remote server
  74     Assert.assertEquals(5, handler.getWorkerCount());
  75   }
  76
  77   private Runnable createEmptyMessageTaskAndHandle(final ClientRequestHandler handler, final String serverAddress) {
  78
  79     return new Runnable() {
  80       @Override
  81       public void run() {
  82         Message request = new Message();
  83         request.setRecipient(serverAddress);
  84         try {
  85           Message response = handleMessageWithTimeout(request);
  86           Assert.assertNotNull(response);
  87           Assert.assertEquals(response.getRecipient(), serverAddress);
  88         } catch (Exception e) {
  89           throw new RuntimeException(e);
  90         }
  91       }
  92     };
  93   }
  94
  95   private Message handleMessageWithTimeout(final Message request) {
  96     Message response = null;
  97
  98     FutureTask task = new FutureTask(new Callable<Message>() {
  99
 100       @Override
 101       public Message call() {
 102         try {
 103           return handler.handle(request);
 104         } catch (Exception e) {
 105           _logger.debug("Client handler failed to handle request. Exception is [{}]", e);
 106         }
 107         return null;
 108       }
 109     });
 110
 111     serverThread.execute(task);
 112
 113     try {
 114       response = (Message) task.get(5L, TimeUnit.SECONDS); //wait for max 5 sec for server to respond
 115     } catch (Exception e) {/*ignore and continue*/}
 116
 117     return response;
 118   }
 119
 120 }