<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-jar-plugin</artifactId>
<version>2.4</version>
+ <executions>
+ <execution>
+ <phase>package</phase>
+ <goals>
+ <goal>test-jar</goal>
+ </goals>
+ </execution>
+ </executions>
</plugin>
</plugins>
</build>
import io.netty.channel.EventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.util.concurrent.Promise;
-
-import java.net.InetSocketAddress;
-
import org.opendaylight.controller.netconf.impl.util.DeserializerExceptionHandler;
import org.opendaylight.controller.netconf.util.AbstractChannelInitializer;
import org.opendaylight.protocol.framework.AbstractDispatcher;
+import java.net.InetSocketAddress;
+
public class NetconfServerDispatcher extends AbstractDispatcher<NetconfServerSession, NetconfServerSessionListener> {
private final ServerChannelInitializer initializer;
public static final String DESERIALIZER_EX_HANDLER_KEY = "deserializerExHandler";
private final NetconfServerSessionNegotiatorFactory negotiatorFactory;
- private final NetconfServerSessionListenerFactory listenerFactory;
- public ServerChannelInitializer(NetconfServerSessionNegotiatorFactory negotiatorFactory,
- NetconfServerSessionListenerFactory listenerFactory) {
+
+ public ServerChannelInitializer(NetconfServerSessionNegotiatorFactory negotiatorFactory) {
this.negotiatorFactory = negotiatorFactory;
- this.listenerFactory = listenerFactory;
+
}
@Override
@Override
protected void initializeSessionNegotiator(SocketChannel ch, Promise<NetconfServerSession> promise) {
- ch.pipeline().addAfter(DESERIALIZER_EX_HANDLER_KEY, AbstractChannelInitializer.NETCONF_SESSION_NEGOTIATOR, negotiatorFactory.getSessionNegotiator(listenerFactory, ch, promise));
+ ch.pipeline().addAfter(DESERIALIZER_EX_HANDLER_KEY, AbstractChannelInitializer.NETCONF_SESSION_NEGOTIATOR,
+ negotiatorFactory.getSessionNegotiator(null, ch, promise));
}
}
static final Logger logger = LoggerFactory.getLogger(NetconfServerSessionListener.class);
private final SessionMonitoringService monitoringService;
private final NetconfOperationRouter operationRouter;
+ private final AutoCloseable onSessionDownCloseable;
- public NetconfServerSessionListener(NetconfOperationRouter operationRouter, SessionMonitoringService monitoringService) {
+ public NetconfServerSessionListener(NetconfOperationRouter operationRouter, SessionMonitoringService monitoringService,
+ AutoCloseable onSessionDownCloseable) {
this.operationRouter = operationRouter;
this.monitoringService = monitoringService;
+ this.onSessionDownCloseable = onSessionDownCloseable;
}
@Override
@Override
public void onSessionDown(NetconfServerSession netconfNetconfServerSession, Exception cause) {
logger.debug("Session {} down, reason: {}", netconfNetconfServerSession, cause.getMessage());
+ onDown(netconfNetconfServerSession);
+ }
+
+ public void onDown(NetconfServerSession netconfNetconfServerSession) {
monitoringService.onSessionDown(netconfNetconfServerSession);
try {
} catch (Exception closingEx) {
logger.debug("Ignoring exception while closing operationRouter", closingEx);
}
+ try {
+ onSessionDownCloseable.close();
+ } catch(Exception ex){
+ logger.debug("Ignoring exception while closing onSessionDownCloseable", ex);
+ }
}
@Override
NetconfTerminationReason netconfTerminationReason) {
logger.debug("Session {} terminated, reason: {}", netconfNetconfServerSession,
netconfTerminationReason.getErrorMessage());
- monitoringService.onSessionDown(netconfNetconfServerSession);
-
- try {
- operationRouter.close();
- } catch (Exception closingEx) {
- logger.debug("Ignoring exception while closing operationRouter", closingEx);
- }
+ onDown(netconfNetconfServerSession);
}
@Override
import org.opendaylight.controller.netconf.impl.mapping.CapabilityProvider;
import org.opendaylight.controller.netconf.impl.osgi.NetconfOperationRouterImpl;
import org.opendaylight.controller.netconf.impl.osgi.SessionMonitoringService;
-import org.opendaylight.controller.netconf.mapping.api.NetconfOperationProvider;
import org.opendaylight.controller.netconf.mapping.api.NetconfOperationServiceSnapshot;
import org.opendaylight.protocol.framework.SessionListenerFactory;
-import static org.opendaylight.controller.netconf.mapping.api.NetconfOperationProvider.NetconfOperationProviderUtil.getNetconfSessionIdForReporting;
-
public class NetconfServerSessionListenerFactory implements SessionListenerFactory<NetconfServerSessionListener> {
- private final NetconfOperationProvider netconfOperationProvider;
-
private final DefaultCommitNotificationProducer commitNotifier;
-
- private final SessionIdProvider idProvider;
-
private final SessionMonitoringService monitor;
+ private final NetconfOperationServiceSnapshot netconfOperationServiceSnapshot;
+ private final CapabilityProvider capabilityProvider;
+
+ public NetconfServerSessionListenerFactory(DefaultCommitNotificationProducer commitNotifier,
+ SessionMonitoringService monitor,
+ NetconfOperationServiceSnapshot netconfOperationServiceSnapshot,
+ CapabilityProvider capabilityProvider) {
- public NetconfServerSessionListenerFactory(NetconfOperationProvider netconfOperationProvider,
- DefaultCommitNotificationProducer commitNotifier,
- SessionIdProvider idProvider, SessionMonitoringService monitor) {
- this.netconfOperationProvider = netconfOperationProvider;
this.commitNotifier = commitNotifier;
- this.idProvider = idProvider;
this.monitor = monitor;
+ this.netconfOperationServiceSnapshot = netconfOperationServiceSnapshot;
+ this.capabilityProvider = capabilityProvider;
}
@Override
public NetconfServerSessionListener getSessionListener() {
- NetconfOperationServiceSnapshot netconfOperationServiceSnapshot = netconfOperationProvider.getSnapshot(
- getNetconfSessionIdForReporting(idProvider.getCurrentSessionId()));
-
- CapabilityProvider capabilityProvider = new CapabilityProviderImpl(netconfOperationServiceSnapshot);
-
NetconfOperationRouter operationRouter = NetconfOperationRouterImpl.createOperationRouter(
- netconfOperationServiceSnapshot, capabilityProvider,
- commitNotifier);
-
- return new NetconfServerSessionListener(operationRouter, monitor);
+ netconfOperationServiceSnapshot, capabilityProvider, commitNotifier);
+ return new NetconfServerSessionListener(operationRouter, monitor, netconfOperationServiceSnapshot);
}
}
package org.opendaylight.controller.netconf.impl;
-import java.net.InetSocketAddress;
-
+import com.google.common.base.Optional;
+import io.netty.channel.Channel;
+import io.netty.util.Timer;
+import io.netty.util.concurrent.Promise;
import org.opendaylight.controller.netconf.api.NetconfServerSessionPreferences;
import org.opendaylight.controller.netconf.util.AbstractNetconfSessionNegotiator;
import org.opendaylight.controller.netconf.util.messages.NetconfHelloMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import com.google.common.base.Optional;
-
-import io.netty.channel.Channel;
-import io.netty.util.Timer;
-import io.netty.util.concurrent.Promise;
+import java.net.InetSocketAddress;
public class NetconfServerSessionNegotiator extends
AbstractNetconfSessionNegotiator<NetconfServerSessionPreferences, NetconfServerSession, NetconfServerSessionListener> {
return new NetconfServerSession(sessionListener, channel, getSessionPreferences().getSessionId(), parsedHeader);
}
- }
+}
import io.netty.util.concurrent.Promise;
import org.opendaylight.controller.netconf.api.NetconfServerSessionPreferences;
import org.opendaylight.controller.netconf.impl.mapping.CapabilityProvider;
+import org.opendaylight.controller.netconf.impl.osgi.SessionMonitoringService;
import org.opendaylight.controller.netconf.mapping.api.NetconfOperationProvider;
+import org.opendaylight.controller.netconf.mapping.api.NetconfOperationServiceSnapshot;
import org.opendaylight.controller.netconf.util.NetconfUtil;
import org.opendaylight.controller.netconf.util.messages.NetconfHelloMessage;
import org.opendaylight.controller.netconf.util.xml.XMLNetconfUtil;
private final SessionIdProvider idProvider;
private final NetconfOperationProvider netconfOperationProvider;
private final long connectionTimeoutMillis;
+ private final DefaultCommitNotificationProducer commitNotificationProducer;
+ private final SessionMonitoringService monitoringService;
public NetconfServerSessionNegotiatorFactory(Timer timer, NetconfOperationProvider netconfOperationProvider,
- SessionIdProvider idProvider, long connectionTimeoutMillis) {
+ SessionIdProvider idProvider, long connectionTimeoutMillis,
+ DefaultCommitNotificationProducer commitNot, SessionMonitoringService monitoringService) {
this.timer = timer;
this.netconfOperationProvider = netconfOperationProvider;
this.idProvider = idProvider;
this.connectionTimeoutMillis = connectionTimeoutMillis;
+ this.commitNotificationProducer = commitNot;
+ this.monitoringService = monitoringService;
}
private static Document loadHelloMessageTemplate() {
return NetconfUtil.createMessage(resourceAsStream).getDocument();
}
+ /**
+ *
+ * @param defunctSessionListenerFactory will not be taken into account as session listener factory can
+ * only be created after snapshot is opened, thus this method constructs
+ * proper session listener factory.
+ * @param channel Underlying channel
+ * @param promise Promise to be notified
+ * @return session negotiator
+ */
@Override
- public SessionNegotiator<NetconfServerSession> getSessionNegotiator(SessionListenerFactory<NetconfServerSessionListener> sessionListenerFactory, Channel channel,
- Promise<NetconfServerSession> promise) {
+ public SessionNegotiator<NetconfServerSession> getSessionNegotiator(SessionListenerFactory<NetconfServerSessionListener> defunctSessionListenerFactory,
+ Channel channel, Promise<NetconfServerSession> promise) {
long sessionId = idProvider.getNextSessionId();
+ NetconfOperationServiceSnapshot netconfOperationServiceSnapshot = netconfOperationProvider.openSnapshot(
+ getNetconfSessionIdForReporting(sessionId));
+ CapabilityProvider capabilityProvider = new CapabilityProviderImpl(netconfOperationServiceSnapshot);
+
+ NetconfServerSessionPreferences proposal = new NetconfServerSessionPreferences(
+ createHelloMessage(sessionId, capabilityProvider), sessionId);
+
+ NetconfServerSessionListenerFactory sessionListenerFactory = new NetconfServerSessionListenerFactory(
+ commitNotificationProducer, monitoringService,
+ netconfOperationServiceSnapshot, capabilityProvider);
- NetconfServerSessionPreferences proposal = new NetconfServerSessionPreferences(createHelloMessage(sessionId),
- sessionId);
return new NetconfServerSessionNegotiator(proposal, promise, channel, timer,
sessionListenerFactory.getSessionListener(), connectionTimeoutMillis);
}
private static final XPathExpression capabilitiesXPath = XMLNetconfUtil
.compileXPath("/netconf:hello/netconf:capabilities");
- private NetconfHelloMessage createHelloMessage(long sessionId) {
+ private NetconfHelloMessage createHelloMessage(long sessionId, CapabilityProvider capabilityProvider) {
Document helloMessageTemplate = getHelloTemplateClone();
// change session ID
final Element capabilitiesElement = (Element) XmlUtil.evaluateXPath(capabilitiesXPath, helloMessageTemplate,
XPathConstants.NODE);
- CapabilityProvider capabilityProvider = new CapabilityProviderImpl(netconfOperationProvider.getSnapshot(
- getNetconfSessionIdForReporting(sessionId)));
-
for (String capability : capabilityProvider.getCapabilities()) {
final Element capabilityElement = helloMessageTemplate.createElement(XmlNetconfConstants.CAPABILITY);
capabilityElement.setTextContent(capability);
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.util.HashedWheelTimer;
-import org.opendaylight.controller.netconf.mapping.api.NetconfOperationProvider;
import org.opendaylight.controller.netconf.api.monitoring.NetconfMonitoringService;
import org.opendaylight.controller.netconf.impl.DefaultCommitNotificationProducer;
import org.opendaylight.controller.netconf.impl.NetconfServerDispatcher;
-import org.opendaylight.controller.netconf.impl.NetconfServerSessionListenerFactory;
import org.opendaylight.controller.netconf.impl.NetconfServerSessionNegotiatorFactory;
import org.opendaylight.controller.netconf.impl.SessionIdProvider;
+import org.opendaylight.controller.netconf.mapping.api.NetconfOperationProvider;
import org.opendaylight.controller.netconf.util.osgi.NetconfConfigUtil;
import org.osgi.framework.BundleActivator;
import org.osgi.framework.BundleContext;
SessionIdProvider idProvider = new SessionIdProvider();
timer = new HashedWheelTimer();
long connectionTimeoutMillis = NetconfConfigUtil.extractTimeoutMillis(context);
- NetconfServerSessionNegotiatorFactory serverNegotiatorFactory = new NetconfServerSessionNegotiatorFactory(
- timer, factoriesListener, idProvider, connectionTimeoutMillis);
+
commitNot = new DefaultCommitNotificationProducer(ManagementFactory.getPlatformMBeanServer());
- NetconfMonitoringServiceImpl monitoringService = startMonitoringService(context, factoriesListener);
+ SessionMonitoringService monitoringService = startMonitoringService(context, factoriesListener);
- NetconfServerSessionListenerFactory listenerFactory = new NetconfServerSessionListenerFactory(
- factoriesListener, commitNot, idProvider, monitoringService);
+ NetconfServerSessionNegotiatorFactory serverNegotiatorFactory = new NetconfServerSessionNegotiatorFactory(
+ timer, factoriesListener, idProvider, connectionTimeoutMillis, commitNot, monitoringService);
eventLoopGroup = new NioEventLoopGroup();
NetconfServerDispatcher.ServerChannelInitializer serverChannelInitializer = new NetconfServerDispatcher.ServerChannelInitializer(
- serverNegotiatorFactory, listenerFactory);
+ serverNegotiatorFactory);
NetconfServerDispatcher dispatch = new NetconfServerDispatcher(serverChannelInitializer, eventLoopGroup, eventLoopGroup);
logger.info("Starting TCP netconf server at {}", address);
@Override
public Schemas getSchemas() {
// capabilities should be split from operations (it will allow to move getSchema operation to monitoring module)
- try (NetconfOperationServiceSnapshot snapshot = netconfOperationProvider.getSnapshot("netconf-monitoring")) {
+ try (NetconfOperationServiceSnapshot snapshot = netconfOperationProvider.openSnapshot("netconf-monitoring")) {
return transformSchemas(snapshot.getServices());
} catch (RuntimeException e) {
throw e;
}
@Override
- public synchronized NetconfOperationServiceSnapshotImpl getSnapshot(String sessionIdForReporting) {
+ public synchronized NetconfOperationServiceSnapshotImpl openSnapshot(String sessionIdForReporting) {
return new NetconfOperationServiceSnapshotImpl(allFactories, sessionIdForReporting);
}
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
-import org.mockito.Mock;
import org.opendaylight.controller.netconf.api.NetconfDocumentedException;
import org.opendaylight.controller.netconf.api.NetconfMessage;
import org.opendaylight.controller.netconf.client.test.TestingNetconfClient;
import static org.junit.Assert.fail;
import static org.mockito.Matchers.any;
import static org.mockito.Mockito.doNothing;
-import static org.mockito.MockitoAnnotations.initMocks;
+import static org.mockito.Mockito.mock;
public class ConcurrentClientsTest {
private DefaultCommitNotificationProducer commitNot;
private NetconfServerDispatcher dispatch;
- @Mock
- private SessionMonitoringService monitoring;
+
HashedWheelTimer hashedWheelTimer;
+ public static SessionMonitoringService createMockedMonitoringService() {
+ SessionMonitoringService monitoring = mock(SessionMonitoringService.class);
+ doNothing().when(monitoring).onSessionUp(any(NetconfServerSession.class));
+ doNothing().when(monitoring).onSessionDown(any(NetconfServerSession.class));
+ return monitoring;
+ }
+
@Before
public void setUp() throws Exception {
- initMocks(this);
+
nettyGroup = new NioEventLoopGroup();
NetconfHelloMessageAdditionalHeader additionalHeader = new NetconfHelloMessageAdditionalHeader("uname", "10.10.10.1", "830", "tcp", "client");
netconfClientDispatcher = new NetconfClientDispatcher( nettyGroup, nettyGroup, additionalHeader, 5000);
SessionIdProvider idProvider = new SessionIdProvider();
hashedWheelTimer = new HashedWheelTimer();
NetconfServerSessionNegotiatorFactory serverNegotiatorFactory = new NetconfServerSessionNegotiatorFactory(
- hashedWheelTimer, factoriesListener, idProvider, 5000);
+ hashedWheelTimer, factoriesListener, idProvider, 5000, commitNot, createMockedMonitoringService());
commitNot = new DefaultCommitNotificationProducer(ManagementFactory.getPlatformMBeanServer());
- doNothing().when(monitoring).onSessionUp(any(NetconfServerSession.class));
- doNothing().when(monitoring).onSessionDown(any(NetconfServerSession.class));
- NetconfServerSessionListenerFactory listenerFactory = new NetconfServerSessionListenerFactory(
- factoriesListener, commitNot, idProvider, monitoring);
- NetconfServerDispatcher.ServerChannelInitializer serverChannelInitializer = new NetconfServerDispatcher.ServerChannelInitializer(serverNegotiatorFactory, listenerFactory);
+
+ NetconfServerDispatcher.ServerChannelInitializer serverChannelInitializer = new NetconfServerDispatcher.ServerChannelInitializer(serverNegotiatorFactory);
dispatch = new NetconfServerDispatcher(serverChannelInitializer, nettyGroup, nettyGroup);
ChannelFuture s = dispatch.createServer(netconfAddress);
private DefaultCommitNotificationProducer commitNot;
private HashedWheelTimer hashedWheelTimer;
+
@Before
public void setUp() throws Exception {
nettyGroup = new NioEventLoopGroup();
SessionIdProvider idProvider = new SessionIdProvider();
hashedWheelTimer = new HashedWheelTimer();
NetconfServerSessionNegotiatorFactory serverNegotiatorFactory = new NetconfServerSessionNegotiatorFactory(
- hashedWheelTimer, factoriesListener, idProvider, 5000);
+ hashedWheelTimer, factoriesListener, idProvider, 5000, commitNot, ConcurrentClientsTest.createMockedMonitoringService());
- NetconfServerSessionListenerFactory listenerFactory = new NetconfServerSessionListenerFactory(
- factoriesListener, commitNot, idProvider, null);
- NetconfServerDispatcher.ServerChannelInitializer serverChannelInitializer = new NetconfServerDispatcher.ServerChannelInitializer(serverNegotiatorFactory, listenerFactory);
+ NetconfServerDispatcher.ServerChannelInitializer serverChannelInitializer = new NetconfServerDispatcher.ServerChannelInitializer(serverNegotiatorFactory);
dispatch = new NetconfServerDispatcher(
serverChannelInitializer, nettyGroup, nettyGroup);
<artifactId>netconf-impl</artifactId>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>${project.groupId}</groupId>
+ <artifactId>netconf-impl</artifactId>
+ <scope>test</scope>
+ <type>test-jar</type>
+ </dependency>
<dependency>
<groupId>${project.groupId}</groupId>
<artifactId>netconf-monitoring</artifactId>
import org.opendaylight.controller.config.manager.impl.AbstractConfigTest;
import org.opendaylight.controller.netconf.impl.DefaultCommitNotificationProducer;
import org.opendaylight.controller.netconf.impl.NetconfServerDispatcher;
-import org.opendaylight.controller.netconf.impl.NetconfServerSessionListenerFactory;
import org.opendaylight.controller.netconf.impl.NetconfServerSessionNegotiatorFactory;
import org.opendaylight.controller.netconf.impl.SessionIdProvider;
import org.opendaylight.controller.netconf.impl.osgi.NetconfOperationServiceFactoryListenerImpl;
SessionIdProvider idProvider = new SessionIdProvider();
NetconfServerSessionNegotiatorFactory serverNegotiatorFactory = new NetconfServerSessionNegotiatorFactory(
- hashedWheelTimer, factoriesListener, idProvider, 5000);
-
- NetconfServerSessionListenerFactory listenerFactory = new NetconfServerSessionListenerFactory(
- factoriesListener, commitNotifier, idProvider,
- sessionMonitoringService);
+ hashedWheelTimer, factoriesListener, idProvider, 5000, commitNotifier, sessionMonitoringService);
NetconfServerDispatcher.ServerChannelInitializer serverChannelInitializer = new NetconfServerDispatcher.ServerChannelInitializer(
- serverNegotiatorFactory, listenerFactory);
+ serverNegotiatorFactory);
return new NetconfServerDispatcher(serverChannelInitializer, nettyThreadgroup, nettyThreadgroup);
}
doReturn(caps).when(service).getCapabilities();
Set<NetconfOperationService> services = Sets.newHashSet(service);
doReturn(services).when(snap).getServices();
- doReturn(snap).when(factoriesListener).getSnapshot(anyString());
+ doReturn(snap).when(factoriesListener).openSnapshot(anyString());
return factoriesListener;
}
NetconfOperationProvider netconfOperationProvider = mock(NetconfOperationProvider.class);
NetconfOperationServiceSnapshotImpl snap = mock(NetconfOperationServiceSnapshotImpl.class);
doReturn(Collections.<NetconfOperationService>emptySet()).when(snap).getServices();
- doReturn(snap).when(netconfOperationProvider).getSnapshot(anyString());
+ doReturn(snap).when(netconfOperationProvider).openSnapshot(anyString());
return new NetconfMonitoringServiceImpl(netconfOperationProvider);
}
doReturn(caps).when(service).getCapabilities();
Set<NetconfOperationService> services = Sets.newHashSet(service);
doReturn(services).when(snap).getServices();
- doReturn(snap).when(factoriesListener).getSnapshot(anyString());
+ doReturn(snap).when(factoriesListener).openSnapshot(anyString());
return factoriesListener;
}
public interface NetconfOperationProvider {
- NetconfOperationServiceSnapshot getSnapshot(String sessionIdForReporting);
+ NetconfOperationServiceSnapshot openSnapshot(String sessionIdForReporting);
public static class NetconfOperationProviderUtil {
<artifactId>netconf-impl</artifactId>
<version>${netconf.version}</version>
</dependency>
+ <dependency>
+ <groupId>${project.groupId}</groupId>
+ <artifactId>netconf-impl</artifactId>
+ <version>${netconf.version}</version>
+ <type>test-jar</type>
+ </dependency>
<dependency>
<groupId>${project.groupId}</groupId>
<artifactId>netconf-monitoring</artifactId>