2 * Copyright (c) 2013 Cisco Systems, Inc. and others. All rights reserved.
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
9 package org.opendaylight.controller.netconf.impl;
11 import static com.google.common.base.Preconditions.checkNotNull;
12 import static org.junit.Assert.assertEquals;
13 import static org.junit.Assert.fail;
14 import static org.mockito.Matchers.any;
15 import static org.mockito.Mockito.doNothing;
16 import static org.mockito.Mockito.mock;
18 import com.google.common.base.Preconditions;
19 import com.google.common.collect.Lists;
20 import io.netty.channel.EventLoopGroup;
21 import io.netty.util.HashedWheelTimer;
22 import java.io.DataOutputStream;
23 import java.io.InputStream;
24 import java.io.InputStreamReader;
25 import java.lang.management.ManagementFactory;
26 import java.net.InetSocketAddress;
27 import java.net.Socket;
28 import java.util.Arrays;
29 import java.util.Collection;
30 import java.util.Collections;
31 import java.util.List;
34 import java.util.concurrent.ExecutionException;
35 import java.util.concurrent.ExecutorService;
36 import java.util.concurrent.Executors;
37 import java.util.concurrent.Future;
38 import java.util.concurrent.ThreadFactory;
39 import java.util.concurrent.atomic.AtomicLong;
40 import org.apache.commons.io.IOUtils;
41 import org.junit.After;
42 import org.junit.AfterClass;
43 import org.junit.Before;
44 import org.junit.BeforeClass;
45 import org.junit.Test;
46 import org.junit.runner.RunWith;
47 import org.junit.runners.Parameterized;
48 import org.opendaylight.controller.netconf.api.NetconfDocumentedException;
49 import org.opendaylight.controller.netconf.api.NetconfMessage;
50 import org.opendaylight.controller.netconf.client.NetconfClientDispatcher;
51 import org.opendaylight.controller.netconf.client.test.TestingNetconfClient;
52 import org.opendaylight.controller.netconf.impl.osgi.NetconfOperationServiceFactoryListenerImpl;
53 import org.opendaylight.controller.netconf.impl.osgi.SessionMonitoringService;
54 import org.opendaylight.controller.netconf.mapping.api.Capability;
55 import org.opendaylight.controller.netconf.mapping.api.HandlingPriority;
56 import org.opendaylight.controller.netconf.mapping.api.NetconfOperation;
57 import org.opendaylight.controller.netconf.mapping.api.NetconfOperationChainedExecution;
58 import org.opendaylight.controller.netconf.mapping.api.NetconfOperationService;
59 import org.opendaylight.controller.netconf.mapping.api.NetconfOperationServiceFactory;
60 import org.opendaylight.controller.netconf.util.messages.NetconfHelloMessageAdditionalHeader;
61 import org.opendaylight.controller.netconf.util.messages.NetconfMessageUtil;
62 import org.opendaylight.controller.netconf.util.messages.NetconfStartExiMessage;
63 import org.opendaylight.controller.netconf.util.test.XmlFileLoader;
64 import org.opendaylight.controller.netconf.util.xml.XmlNetconfConstants;
65 import org.opendaylight.controller.netconf.util.xml.XmlUtil;
66 import org.slf4j.Logger;
67 import org.slf4j.LoggerFactory;
68 import org.w3c.dom.Document;
70 import com.google.common.collect.Sets;
72 import io.netty.channel.ChannelFuture;
73 import io.netty.channel.nio.NioEventLoopGroup;
75 @RunWith(Parameterized.class)
76 public class ConcurrentClientsTest {
77 private static final Logger logger = LoggerFactory.getLogger(ConcurrentClientsTest.class);
79 private static ExecutorService clientExecutor;
81 private static final int CONCURRENCY = 32;
82 private static final InetSocketAddress netconfAddress = new InetSocketAddress("127.0.0.1", 8303);
84 private int nettyThreads;
85 private Class<? extends Runnable> clientRunnable;
86 private Set<String> serverCaps;
88 public ConcurrentClientsTest(int nettyThreads, Class<? extends Runnable> clientRunnable, Set<String> serverCaps) {
89 this.nettyThreads = nettyThreads;
90 this.clientRunnable = clientRunnable;
91 this.serverCaps = serverCaps;
94 @Parameterized.Parameters()
95 public static Collection<Object[]> data() {
96 return Arrays.asList(new Object[][]{
97 {4, TestingNetconfClientRunnable.class, NetconfServerSessionNegotiatorFactory.DEFAULT_BASE_CAPABILITIES},
98 {1, TestingNetconfClientRunnable.class, NetconfServerSessionNegotiatorFactory.DEFAULT_BASE_CAPABILITIES},
99 // empty set of capabilities = only base 1.0 netconf capability
100 {4, TestingNetconfClientRunnable.class, Collections.emptySet()},
101 {4, TestingNetconfClientRunnable.class, getOnlyExiServerCaps()},
102 {4, TestingNetconfClientRunnable.class, getOnlyChunkServerCaps()},
104 {4, BlockingClientRunnable.class, getOnlyExiServerCaps()},
105 {1, BlockingClientRunnable.class, getOnlyExiServerCaps()},
109 private EventLoopGroup nettyGroup;
110 private NetconfClientDispatcher netconfClientDispatcher;
112 private DefaultCommitNotificationProducer commitNot;
114 HashedWheelTimer hashedWheelTimer;
115 private TestingNetconfOperation testingNetconfOperation;
117 public static SessionMonitoringService createMockedMonitoringService() {
118 SessionMonitoringService monitoring = mock(SessionMonitoringService.class);
119 doNothing().when(monitoring).onSessionUp(any(NetconfServerSession.class));
120 doNothing().when(monitoring).onSessionDown(any(NetconfServerSession.class));
125 public static void setUpClientExecutor() {
126 clientExecutor = Executors.newFixedThreadPool(CONCURRENCY, new ThreadFactory() {
130 public Thread newThread(final Runnable r) {
131 Thread thread = new Thread(r);
132 thread.setName("client-" + i++);
133 thread.setDaemon(true);
140 public void setUp() throws Exception {
142 nettyGroup = new NioEventLoopGroup(nettyThreads);
143 NetconfHelloMessageAdditionalHeader additionalHeader = new NetconfHelloMessageAdditionalHeader("uname", "10.10.10.1", "830", "tcp", "client");
144 netconfClientDispatcher = new NetconfClientDispatcher( nettyGroup, nettyGroup, additionalHeader, 5000);
146 NetconfOperationServiceFactoryListenerImpl factoriesListener = new NetconfOperationServiceFactoryListenerImpl();
148 testingNetconfOperation = new TestingNetconfOperation();
149 factoriesListener.onAddNetconfOperationServiceFactory(new TestingOperationServiceFactory(testingNetconfOperation));
151 SessionIdProvider idProvider = new SessionIdProvider();
152 hashedWheelTimer = new HashedWheelTimer();
154 NetconfServerSessionNegotiatorFactory serverNegotiatorFactory = new NetconfServerSessionNegotiatorFactory(
155 hashedWheelTimer, factoriesListener, idProvider, 5000, commitNot, createMockedMonitoringService(), serverCaps);
157 commitNot = new DefaultCommitNotificationProducer(ManagementFactory.getPlatformMBeanServer());
159 NetconfServerDispatcher.ServerChannelInitializer serverChannelInitializer = new NetconfServerDispatcher.ServerChannelInitializer(serverNegotiatorFactory);
160 final NetconfServerDispatcher dispatch = new NetconfServerDispatcher(serverChannelInitializer, nettyGroup, nettyGroup);
162 ChannelFuture s = dispatch.createServer(netconfAddress);
167 public void tearDown(){
169 hashedWheelTimer.stop();
171 nettyGroup.shutdownGracefully().get();
172 } catch (InterruptedException | ExecutionException e) {
173 logger.warn("Ignoring exception while cleaning up after test", e);
178 public static void tearDownClientExecutor() {
179 clientExecutor.shutdownNow();
182 @Test(timeout = CONCURRENCY * 1000)
183 public void testConcurrentClients() throws Exception {
185 List<Future<?>> futures = Lists.newArrayListWithCapacity(CONCURRENCY);
187 for (int i = 0; i < CONCURRENCY; i++) {
188 futures.add(clientExecutor.submit(getInstanceOfClientRunnable()));
191 for (Future<?> future : futures) {
194 } catch (InterruptedException e) {
195 throw new IllegalStateException(e);
196 } catch (ExecutionException e) {
197 logger.error("Thread for testing client failed", e);
198 fail("Client failed: " + e.getMessage());
202 assertEquals(CONCURRENCY, testingNetconfOperation.getMessageCount());
205 public static Set<String> getOnlyExiServerCaps() {
206 return Sets.newHashSet(
207 XmlNetconfConstants.URN_IETF_PARAMS_NETCONF_BASE_1_0,
208 XmlNetconfConstants.URN_IETF_PARAMS_NETCONF_CAPABILITY_EXI_1_0
212 public static Set<String> getOnlyChunkServerCaps() {
213 return Sets.newHashSet(
214 XmlNetconfConstants.URN_IETF_PARAMS_NETCONF_BASE_1_0,
215 XmlNetconfConstants.URN_IETF_PARAMS_NETCONF_BASE_1_1
219 public Runnable getInstanceOfClientRunnable() throws Exception {
220 return clientRunnable.getConstructor(ConcurrentClientsTest.class).newInstance(this);
224 * Responds to all operations except start-exi and counts all requests
226 private static class TestingNetconfOperation implements NetconfOperation {
228 private final AtomicLong counter = new AtomicLong();
231 public HandlingPriority canHandle(Document message) {
232 return XmlUtil.toString(message).contains(NetconfStartExiMessage.START_EXI) ?
233 HandlingPriority.CANNOT_HANDLE :
234 HandlingPriority.HANDLE_WITH_MAX_PRIORITY;
238 public Document handle(Document requestMessage, NetconfOperationChainedExecution subsequentOperation) throws NetconfDocumentedException {
240 logger.info("Handling netconf message from test {}", XmlUtil.toString(requestMessage));
241 counter.getAndIncrement();
242 return XmlUtil.readXmlToDocument("<test/>");
243 } catch (Exception e) {
244 throw new RuntimeException(e);
248 public long getMessageCount() {
249 return counter.get();
254 * Hardcoded operation service factory
256 private static class TestingOperationServiceFactory implements NetconfOperationServiceFactory {
257 private final NetconfOperation[] operations;
259 public TestingOperationServiceFactory(final NetconfOperation... operations) {
260 this.operations = operations;
264 public NetconfOperationService createService(String netconfSessionIdForReporting) {
265 return new NetconfOperationService() {
267 public Set<Capability> getCapabilities() {
268 return Collections.emptySet();
272 public Set<NetconfOperation> getNetconfOperations() {
273 return Sets.newHashSet(operations);
277 public void close() {}
283 * Pure socket based blocking client
285 public final class BlockingClientRunnable implements Runnable {
291 } catch (Exception e) {
292 throw new IllegalStateException(Thread.currentThread().getName(), e);
296 private void run2() throws Exception {
297 InputStream clientHello = checkNotNull(XmlFileLoader
298 .getResourceAsStream("netconfMessages/client_hello.xml"));
299 InputStream getConfig = checkNotNull(XmlFileLoader.getResourceAsStream("netconfMessages/getConfig.xml"));
301 Socket clientSocket = new Socket(netconfAddress.getHostString(), netconfAddress.getPort());
302 DataOutputStream outToServer = new DataOutputStream(clientSocket.getOutputStream());
303 InputStreamReader inFromServer = new InputStreamReader(clientSocket.getInputStream());
305 StringBuffer sb = new StringBuffer();
306 while (sb.toString().endsWith("]]>]]>") == false) {
307 sb.append((char) inFromServer.read());
309 logger.info(sb.toString());
311 outToServer.write(IOUtils.toByteArray(clientHello));
312 outToServer.write("]]>]]>".getBytes());
314 // Thread.sleep(100);
315 outToServer.write(IOUtils.toByteArray(getConfig));
316 outToServer.write("]]>]]>".getBytes());
319 sb = new StringBuffer();
320 while (sb.toString().endsWith("]]>]]>") == false) {
321 sb.append((char) inFromServer.read());
323 logger.info(sb.toString());
324 clientSocket.close();
329 * TestingNetconfClient based runnable
331 public final class TestingNetconfClientRunnable implements Runnable {
336 final TestingNetconfClient netconfClient = new TestingNetconfClient(Thread.currentThread().getName(),
337 netconfAddress, netconfClientDispatcher);
338 long sessionId = netconfClient.getSessionId();
339 logger.info("Client with session id {}: hello exchanged", sessionId);
341 final NetconfMessage getMessage = XmlFileLoader
342 .xmlFileToNetconfMessage("netconfMessages/getConfig.xml");
343 NetconfMessage result = netconfClient.sendRequest(getMessage).get();
344 logger.info("Client with session id {}: got result {}", sessionId, result);
346 Preconditions.checkState(NetconfMessageUtil.isErrorMessage(result) == false,
347 "Received error response: " + XmlUtil.toString(result.getDocument()) + " to request: "
348 + XmlUtil.toString(getMessage.getDocument()));
350 netconfClient.close();
351 logger.info("Client with session id {}: ended", sessionId);
352 } catch (final Exception e) {
353 throw new IllegalStateException(Thread.currentThread().getName(), e);