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.netconf.impl;
11 import static com.google.common.base.Preconditions.checkNotNull;
12 import static org.junit.Assert.assertEquals;
13 import static org.mockito.ArgumentMatchers.any;
14 import static org.mockito.Mockito.doNothing;
15 import static org.mockito.Mockito.doReturn;
16 import static org.mockito.Mockito.mock;
18 import com.google.common.base.Preconditions;
19 import com.google.common.collect.Lists;
20 import com.google.common.collect.Sets;
21 import com.google.common.io.ByteStreams;
22 import io.netty.channel.ChannelFuture;
23 import io.netty.channel.EventLoopGroup;
24 import io.netty.channel.nio.NioEventLoopGroup;
25 import io.netty.util.HashedWheelTimer;
26 import io.netty.util.concurrent.GlobalEventExecutor;
27 import java.io.DataOutputStream;
28 import java.io.InputStream;
29 import java.io.InputStreamReader;
30 import java.net.InetSocketAddress;
31 import java.net.Socket;
32 import java.util.Arrays;
33 import java.util.Collection;
34 import java.util.Collections;
35 import java.util.List;
37 import java.util.concurrent.ExecutionException;
38 import java.util.concurrent.ExecutorService;
39 import java.util.concurrent.Executors;
40 import java.util.concurrent.Future;
41 import java.util.concurrent.ThreadFactory;
42 import java.util.concurrent.atomic.AtomicLong;
43 import org.junit.After;
44 import org.junit.AfterClass;
45 import org.junit.Before;
46 import org.junit.BeforeClass;
47 import org.junit.Test;
48 import org.junit.runner.RunWith;
49 import org.junit.runners.Parameterized;
50 import org.opendaylight.netconf.api.DocumentedException;
51 import org.opendaylight.netconf.api.NetconfMessage;
52 import org.opendaylight.netconf.api.capability.Capability;
53 import org.opendaylight.netconf.api.messages.NetconfHelloMessageAdditionalHeader;
54 import org.opendaylight.netconf.api.monitoring.CapabilityListener;
55 import org.opendaylight.netconf.api.monitoring.NetconfMonitoringService;
56 import org.opendaylight.netconf.api.monitoring.SessionEvent;
57 import org.opendaylight.netconf.api.monitoring.SessionListener;
58 import org.opendaylight.netconf.api.xml.XmlNetconfConstants;
59 import org.opendaylight.netconf.api.xml.XmlUtil;
60 import org.opendaylight.netconf.client.NetconfClientDispatcher;
61 import org.opendaylight.netconf.client.NetconfClientDispatcherImpl;
62 import org.opendaylight.netconf.client.SimpleNetconfClientSessionListener;
63 import org.opendaylight.netconf.client.TestingNetconfClient;
64 import org.opendaylight.netconf.client.conf.NetconfClientConfiguration;
65 import org.opendaylight.netconf.client.conf.NetconfClientConfigurationBuilder;
66 import org.opendaylight.netconf.impl.osgi.AggregatedNetconfOperationServiceFactory;
67 import org.opendaylight.netconf.mapping.api.HandlingPriority;
68 import org.opendaylight.netconf.mapping.api.NetconfOperation;
69 import org.opendaylight.netconf.mapping.api.NetconfOperationChainedExecution;
70 import org.opendaylight.netconf.mapping.api.NetconfOperationService;
71 import org.opendaylight.netconf.mapping.api.NetconfOperationServiceFactory;
72 import org.opendaylight.netconf.nettyutil.NeverReconnectStrategy;
73 import org.opendaylight.netconf.nettyutil.handler.exi.NetconfStartExiMessage;
74 import org.opendaylight.netconf.util.messages.NetconfMessageUtil;
75 import org.opendaylight.netconf.util.test.XmlFileLoader;
76 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.netconf.monitoring.rev101004.netconf.state.CapabilitiesBuilder;
77 import org.slf4j.Logger;
78 import org.slf4j.LoggerFactory;
79 import org.w3c.dom.Document;
81 @RunWith(Parameterized.class)
82 public class ConcurrentClientsTest {
83 private static final Logger LOG = LoggerFactory.getLogger(ConcurrentClientsTest.class);
85 private static ExecutorService clientExecutor;
87 private static final int CONCURRENCY = 32;
88 private static final InetSocketAddress NETCONF_ADDRESS = new InetSocketAddress("127.0.0.1", 8303);
90 private final int nettyThreads;
91 private final Class<? extends Runnable> clientRunnable;
92 private final Set<String> serverCaps;
94 public ConcurrentClientsTest(int nettyThreads, Class<? extends Runnable> clientRunnable, Set<String> serverCaps) {
95 this.nettyThreads = nettyThreads;
96 this.clientRunnable = clientRunnable;
97 this.serverCaps = serverCaps;
100 @Parameterized.Parameters()
101 public static Collection<Object[]> data() {
102 return Arrays.asList(new Object[][]{{4, TestingNetconfClientRunnable.class,
103 NetconfServerSessionNegotiatorFactory.DEFAULT_BASE_CAPABILITIES},
104 {1, TestingNetconfClientRunnable.class, NetconfServerSessionNegotiatorFactory.DEFAULT_BASE_CAPABILITIES},
105 // empty set of capabilities = only base 1.0 netconf capability
106 {4, TestingNetconfClientRunnable.class, Collections.emptySet()},
107 {4, TestingNetconfClientRunnable.class, getOnlyExiServerCaps()},
108 {4, TestingNetconfClientRunnable.class, getOnlyChunkServerCaps()},
109 {4, BlockingClientRunnable.class, getOnlyExiServerCaps()},
110 {1, BlockingClientRunnable.class, getOnlyExiServerCaps()},
114 private EventLoopGroup nettyGroup;
115 private NetconfClientDispatcher netconfClientDispatcher;
117 HashedWheelTimer hashedWheelTimer;
118 private TestingNetconfOperation testingNetconfOperation;
120 public static NetconfMonitoringService createMockedMonitoringService() {
121 NetconfMonitoringService monitoring = mock(NetconfMonitoringService.class);
122 final SessionListener sessionListener = mock(SessionListener.class);
123 doNothing().when(sessionListener).onSessionUp(any(NetconfServerSession.class));
124 doNothing().when(sessionListener).onSessionDown(any(NetconfServerSession.class));
125 doNothing().when(sessionListener).onSessionEvent(any(SessionEvent.class));
126 doReturn((AutoCloseable) () -> {
128 }).when(monitoring).registerCapabilitiesListener(any(NetconfMonitoringService.CapabilitiesListener.class));
129 doReturn(sessionListener).when(monitoring).getSessionListener();
130 doReturn(new CapabilitiesBuilder().setCapability(Collections.emptyList()).build()).when(monitoring)
136 public static void setUpClientExecutor() {
137 clientExecutor = Executors.newFixedThreadPool(CONCURRENCY, new ThreadFactory() {
141 public Thread newThread(final Runnable runnable) {
142 Thread thread = new Thread(runnable);
143 thread.setName("client-" + index++);
144 thread.setDaemon(true);
151 public void setUp() throws Exception {
152 hashedWheelTimer = new HashedWheelTimer();
153 nettyGroup = new NioEventLoopGroup(nettyThreads);
154 netconfClientDispatcher = new NetconfClientDispatcherImpl(nettyGroup, nettyGroup, hashedWheelTimer);
156 AggregatedNetconfOperationServiceFactory factoriesListener = new AggregatedNetconfOperationServiceFactory();
158 testingNetconfOperation = new TestingNetconfOperation();
159 factoriesListener.onAddNetconfOperationServiceFactory(
160 new TestingOperationServiceFactory(testingNetconfOperation));
162 SessionIdProvider idProvider = new SessionIdProvider();
164 NetconfServerSessionNegotiatorFactory serverNegotiatorFactory = new
165 NetconfServerSessionNegotiatorFactoryBuilder()
166 .setTimer(hashedWheelTimer)
167 .setAggregatedOpService(factoriesListener)
168 .setIdProvider(idProvider)
169 .setConnectionTimeoutMillis(5000)
170 .setMonitoringService(createMockedMonitoringService())
171 .setBaseCapabilities(serverCaps)
174 NetconfServerDispatcherImpl.ServerChannelInitializer serverChannelInitializer =
175 new NetconfServerDispatcherImpl.ServerChannelInitializer(serverNegotiatorFactory);
176 final NetconfServerDispatcherImpl dispatch =
177 new NetconfServerDispatcherImpl(serverChannelInitializer, nettyGroup, nettyGroup);
179 ChannelFuture server = dispatch.createServer(NETCONF_ADDRESS);
184 public void tearDown() {
185 hashedWheelTimer.stop();
187 nettyGroup.shutdownGracefully().get();
188 } catch (InterruptedException | ExecutionException e) {
189 LOG.warn("Ignoring exception while cleaning up after test", e);
194 public static void tearDownClientExecutor() {
195 clientExecutor.shutdownNow();
198 @Test(timeout = CONCURRENCY * 1000)
199 public void testConcurrentClients() throws Exception {
201 List<Future<?>> futures = Lists.newArrayListWithCapacity(CONCURRENCY);
203 for (int i = 0; i < CONCURRENCY; i++) {
204 futures.add(clientExecutor.submit(getInstanceOfClientRunnable()));
207 for (Future<?> future : futures) {
210 } catch (InterruptedException e) {
211 throw new IllegalStateException(e);
212 } catch (ExecutionException e) {
213 LOG.error("Thread for testing client failed", e);
218 assertEquals(CONCURRENCY, testingNetconfOperation.getMessageCount());
221 public static Set<String> getOnlyExiServerCaps() {
222 return Sets.newHashSet(
223 XmlNetconfConstants.URN_IETF_PARAMS_NETCONF_BASE_1_0,
224 XmlNetconfConstants.URN_IETF_PARAMS_NETCONF_CAPABILITY_EXI_1_0
228 public static Set<String> getOnlyChunkServerCaps() {
229 return Sets.newHashSet(
230 XmlNetconfConstants.URN_IETF_PARAMS_NETCONF_BASE_1_0,
231 XmlNetconfConstants.URN_IETF_PARAMS_NETCONF_BASE_1_1
235 public Runnable getInstanceOfClientRunnable() throws Exception {
236 return clientRunnable.getConstructor(ConcurrentClientsTest.class).newInstance(this);
240 * Responds to all operations except start-exi and counts all requests.
242 private static class TestingNetconfOperation implements NetconfOperation {
244 private final AtomicLong counter = new AtomicLong();
247 public HandlingPriority canHandle(Document message) {
248 return XmlUtil.toString(message).contains(NetconfStartExiMessage.START_EXI)
249 ? HandlingPriority.CANNOT_HANDLE :
250 HandlingPriority.HANDLE_WITH_MAX_PRIORITY;
253 @SuppressWarnings("checkstyle:IllegalCatch")
255 public Document handle(Document requestMessage, NetconfOperationChainedExecution subsequentOperation)
256 throws DocumentedException {
258 LOG.info("Handling netconf message from test {}", XmlUtil.toString(requestMessage));
259 counter.getAndIncrement();
260 return XmlUtil.readXmlToDocument("<test/>");
261 } catch (Exception e) {
262 throw new RuntimeException(e);
266 public long getMessageCount() {
267 return counter.get();
272 * Hardcoded operation service factory.
274 private static class TestingOperationServiceFactory implements NetconfOperationServiceFactory {
275 private final NetconfOperation[] operations;
277 TestingOperationServiceFactory(final NetconfOperation... operations) {
278 this.operations = operations;
282 public Set<Capability> getCapabilities() {
283 return Collections.emptySet();
287 public AutoCloseable registerCapabilityListener(final CapabilityListener listener) {
293 public NetconfOperationService createService(String netconfSessionIdForReporting) {
294 return new NetconfOperationService() {
297 public Set<NetconfOperation> getNetconfOperations() {
298 return Sets.newHashSet(operations);
302 public void close() {
309 * Pure socket based blocking client.
311 @SuppressWarnings("checkstyle:IllegalCatch")
312 public final class BlockingClientRunnable implements Runnable {
318 } catch (Exception e) {
319 throw new IllegalStateException(Thread.currentThread().getName(), e);
323 private void run2() throws Exception {
324 InputStream clientHello = checkNotNull(XmlFileLoader
325 .getResourceAsStream("netconfMessages/client_hello.xml"));
326 final InputStream getConfig =
327 checkNotNull(XmlFileLoader.getResourceAsStream("netconfMessages/getConfig.xml"));
329 Socket clientSocket = new Socket(NETCONF_ADDRESS.getHostString(), NETCONF_ADDRESS.getPort());
330 DataOutputStream outToServer = new DataOutputStream(clientSocket.getOutputStream());
331 InputStreamReader inFromServer = new InputStreamReader(clientSocket.getInputStream());
333 StringBuffer sb = new StringBuffer();
334 while (sb.toString().endsWith("]]>]]>") == false) {
335 sb.append((char) inFromServer.read());
337 LOG.info(sb.toString());
339 outToServer.write(ByteStreams.toByteArray(clientHello));
340 outToServer.write("]]>]]>".getBytes());
342 // Thread.sleep(100);
343 outToServer.write(ByteStreams.toByteArray(getConfig));
344 outToServer.write("]]>]]>".getBytes());
347 sb = new StringBuffer();
348 while (sb.toString().endsWith("]]>]]>") == false) {
349 sb.append((char) inFromServer.read());
351 LOG.info(sb.toString());
352 clientSocket.close();
357 * TestingNetconfClient based runnable.
359 public final class TestingNetconfClientRunnable implements Runnable {
361 @SuppressWarnings("checkstyle:IllegalCatch")
365 final TestingNetconfClient netconfClient =
366 new TestingNetconfClient(Thread.currentThread().getName(), netconfClientDispatcher,
368 long sessionId = netconfClient.getSessionId();
369 LOG.info("Client with session id {}: hello exchanged", sessionId);
371 final NetconfMessage getMessage = XmlFileLoader
372 .xmlFileToNetconfMessage("netconfMessages/getConfig.xml");
373 NetconfMessage result = netconfClient.sendRequest(getMessage).get();
374 LOG.info("Client with session id {}: got result {}", sessionId, result);
376 Preconditions.checkState(NetconfMessageUtil.isErrorMessage(result) == false,
377 "Received error response: " + XmlUtil.toString(result.getDocument()) + " to request: "
378 + XmlUtil.toString(getMessage.getDocument()));
380 netconfClient.close();
381 LOG.info("Client with session id {}: ended", sessionId);
382 } catch (final Exception e) {
383 throw new IllegalStateException(Thread.currentThread().getName(), e);
387 private NetconfClientConfiguration getClientConfig() {
388 final NetconfClientConfigurationBuilder b = NetconfClientConfigurationBuilder.create();
389 b.withAddress(NETCONF_ADDRESS);
390 b.withAdditionalHeader(new NetconfHelloMessageAdditionalHeader("uname", "10.10.10.1", "830", "tcp",
392 b.withSessionListener(new SimpleNetconfClientSessionListener());
393 b.withReconnectStrategy(new NeverReconnectStrategy(GlobalEventExecutor.INSTANCE,
394 NetconfClientConfigurationBuilder.DEFAULT_CONNECTION_TIMEOUT_MILLIS));