import io.netty.channel.Channel;
import io.netty.util.Timer;
import io.netty.util.concurrent.Promise;
+import java.net.SocketAddress;
import java.util.Set;
import org.opendaylight.netconf.mapping.api.NetconfOperationService;
import org.opendaylight.netconf.mapping.api.NetconfOperationServiceFactory;
}
return new NetconfServerSessionNegotiator(proposal, promise, channel, timer,
- getListener(Long.toString(sessionId)), connectionTimeoutMillis);
+ getListener(Long.toString(sessionId), channel.localAddress()), connectionTimeoutMillis);
}
- private NetconfServerSessionListener getListener(final String netconfSessionIdForReporting) {
- final NetconfOperationService service =
- this.aggregatedOpService.createService(netconfSessionIdForReporting);
+ private NetconfServerSessionListener getListener(final String netconfSessionIdForReporting, final SocketAddress socketAddress) {
+ final NetconfOperationService service = getOperationServiceForAddress(netconfSessionIdForReporting, socketAddress);
final NetconfOperationRouter operationRouter =
new NetconfOperationRouterImpl(service, monitoringService, netconfSessionIdForReporting);
return new NetconfServerSessionListener(operationRouter, monitoringService, service);
}
+ protected NetconfOperationService getOperationServiceForAddress(final String netconfSessionIdForReporting, final SocketAddress socketAddress) {
+ return this.aggregatedOpService.createService(netconfSessionIdForReporting);
+ }
+
+ protected final NetconfOperationServiceFactory getOperationServiceFactory() {
+ return aggregatedOpService;
+ }
+
private NetconfHelloMessage createHelloMessage(final long sessionId, final NetconfMonitoringService capabilityProvider) throws NetconfDocumentedException {
return NetconfHelloMessage.createServerHello(Sets.union(transformCapabilities(capabilityProvider.getCapabilities()), baseCapabilities), sessionId);
}
private void scheduleKeepalive() {
Preconditions.checkState(currentDeviceRpc != null);
LOG.trace("{}: Scheduling next keepalive in {} {}", id, keepaliveDelaySeconds, TimeUnit.SECONDS);
- currentKeepalive = executor.schedule(new Keepalive(), keepaliveDelaySeconds, TimeUnit.SECONDS);
+ currentKeepalive = executor.schedule(new Keepalive(currentKeepalive), keepaliveDelaySeconds, TimeUnit.SECONDS);
}
@Override
*/
private class Keepalive implements Runnable, FutureCallback<DOMRpcResult> {
+ private final ScheduledFuture<?> previousKeepalive;
+
+ public Keepalive(final ScheduledFuture<?> previousKeepalive) {
+ this.previousKeepalive = previousKeepalive;
+ }
+
@Override
public void run() {
LOG.trace("{}: Invoking keepalive RPC", id);
try {
- if(!currentKeepalive.isDone()) {
+ if(previousKeepalive != null && !previousKeepalive.isDone()) {
onFailure(new IllegalStateException("Previous keepalive timed out"));
} else {
Futures.addCallback(currentDeviceRpc.invokeRpc(PATH, KEEPALIVE_PAYLOAD), this);
- scheduleKeepalive();
}
} catch (NullPointerException e) {
LOG.debug("{}: Skipping keepalive while reconnecting", id);
verify(underlyingSalFacade).onDeviceConnected(
any(SchemaContext.class), any(NetconfSessionPreferences.class), any(DOMRpcService.class));
- verify(deviceRpc, timeout(15000).atLeast(5)).invokeRpc(any(SchemaPath.class), any(NormalizedNode.class));
+ verify(deviceRpc, timeout(15000).times(5)).invokeRpc(any(SchemaPath.class), any(NormalizedNode.class));
}
@Test
.getLogger(MdsalOperationProvider.class);
private final Set<Capability> caps;
- private final MdsalOperationService mdsalOperationService;
+ private final SchemaContext schemaContext;
public MdsalOperationProvider(final SessionIdProvider idProvider,
final Set<Capability> caps,
final SchemaContext schemaContext) {
this.caps = caps;
- mdsalOperationService = new MdsalOperationService(
- idProvider.getCurrentSessionId(), schemaContext, caps);
+ this.schemaContext = schemaContext;
}
@Override
@Override
public NetconfOperationService createService(String netconfSessionIdForReporting) {
- return mdsalOperationService;
+ return new MdsalOperationService(Long.parseLong(netconfSessionIdForReporting), schemaContext, caps);
}
static class MdsalOperationService implements NetconfOperationService {
private final long currentSessionId;
private final SchemaContext schemaContext;
private final Set<Capability> caps;
+ private final SchemaService schemaService;
+ private final DOMDataBroker dataBroker;
public MdsalOperationService(final long currentSessionId,
final SchemaContext schemaContext,
this.currentSessionId = currentSessionId;
this.schemaContext = schemaContext;
this.caps = caps;
+ this.schemaService = createSchemaService();
+
+ this.dataBroker = createDataStore(schemaService);
+
}
@Override
public Set<NetconfOperation> getNetconfOperations() {
- final SchemaService schemaService = createSchemaService();
-
- final DOMDataBroker db = createDataStore(schemaService);
- TransactionProvider transactionProvider = new TransactionProvider(db, String.valueOf(currentSessionId));
+ TransactionProvider transactionProvider = new TransactionProvider(dataBroker, String.valueOf(currentSessionId));
CurrentSchemaContext currentSchemaContext = new CurrentSchemaContext(schemaService);
ContainerNode netconf = createNetconfState();
YangInstanceIdentifier yangInstanceIdentifier = YangInstanceIdentifier.builder().node(NetconfState.QNAME)
.build();
- final DOMDataWriteTransaction tx = db.newWriteOnlyTransaction();
+ final DOMDataWriteTransaction tx = dataBroker.newWriteOnlyTransaction();
tx.put(LogicalDatastoreType.OPERATIONAL, yangInstanceIdentifier, netconf);
try {
? NetconfServerSessionNegotiatorFactory.DEFAULT_BASE_CAPABILITIES
: Sets.newHashSet(XmlNetconfConstants.URN_IETF_PARAMS_NETCONF_BASE_1_0, XmlNetconfConstants.URN_IETF_PARAMS_NETCONF_BASE_1_1);
- final NetconfServerSessionNegotiatorFactory serverNegotiatorFactory = new NetconfServerSessionNegotiatorFactory(
+ final NetconfServerSessionNegotiatorFactory serverNegotiatorFactory = new TesttoolNegotiationFactory(
hashedWheelTimer, aggregatedNetconfOperationServiceFactory, idProvider, generateConfigsTimeout, monitoringService1, serverCapabilities);
final NetconfServerDispatcherImpl.ServerChannelInitializer serverChannelInitializer = new NetconfServerDispatcherImpl.ServerChannelInitializer(
--- /dev/null
+/*
+ * Copyright (c) 2015 Cisco Systems, Inc. and others. All rights reserved.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License v1.0 which accompanies this distribution,
+ * and is available at http://www.eclipse.org/legal/epl-v10.html
+ */
+
+package org.opendaylight.netconf.test.tool;
+
+import io.netty.util.Timer;
+import java.net.SocketAddress;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+import org.opendaylight.netconf.api.monitoring.NetconfMonitoringService;
+import org.opendaylight.netconf.impl.NetconfServerSessionNegotiatorFactory;
+import org.opendaylight.netconf.impl.SessionIdProvider;
+import org.opendaylight.netconf.mapping.api.NetconfOperationService;
+import org.opendaylight.netconf.mapping.api.NetconfOperationServiceFactory;
+
+public class TesttoolNegotiationFactory extends NetconfServerSessionNegotiatorFactory {
+
+ private final Map<SocketAddress, NetconfOperationService> cachedOperationServices = new HashMap<>();
+
+ public TesttoolNegotiationFactory(final Timer timer, final NetconfOperationServiceFactory netconfOperationProvider,
+ final SessionIdProvider idProvider, final long connectionTimeoutMillis,
+ final NetconfMonitoringService monitoringService) {
+ super(timer, netconfOperationProvider, idProvider, connectionTimeoutMillis, monitoringService);
+ }
+
+ public TesttoolNegotiationFactory(final Timer timer, final NetconfOperationServiceFactory netconfOperationProvider,
+ final SessionIdProvider idProvider, final long connectionTimeoutMillis,
+ final NetconfMonitoringService monitoringService, final Set<String> baseCapabilities) {
+ super(timer, netconfOperationProvider, idProvider, connectionTimeoutMillis, monitoringService, baseCapabilities);
+ }
+
+ @Override
+ protected NetconfOperationService getOperationServiceForAddress(final String netconfSessionIdForReporting, final SocketAddress socketAddress) {
+ if (cachedOperationServices.containsKey(socketAddress)) {
+ return cachedOperationServices.get(socketAddress);
+ } else {
+ final NetconfOperationService service = getOperationServiceFactory().createService(netconfSessionIdForReporting);
+ cachedOperationServices.put(socketAddress, service);
+ return service;
+ }
+ }
+}
* document the behavior).
*/
int tries = 2;
- Status status = Status.NOT_MODIFIED;
while(true) {
try {
if (mountPoint != null) {
- status = broker.readConfigurationData(mountPoint, normalizedII) != null ? Status.OK : Status.CREATED;
broker.commitConfigurationDataPut(mountPoint, normalizedII, payload.getData()).checkedGet();
} else {
- status = broker.readConfigurationData(normalizedII) != null ? Status.OK : Status.CREATED;
broker.commitConfigurationDataPut(controllerContext.getGlobalSchema(), normalizedII, payload.getData()).checkedGet();
}
}
}
- return Response.status(status).build();
+ return Response.status(Status.OK).build();
}
private static void validateTopLevelNodeName(final NormalizedNodeContext node,
Code Review / netconf.git / commitdiff