Make notification filter a simple lambda
[netconf.git] / netconf / sal-netconf-connector / src / main / java / org / opendaylight / netconf / sal / connect / netconf / NetconfDevice.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.netconf.sal.connect.netconf;
9
10 import static com.google.common.base.Preconditions.checkState;
11 import static java.util.Objects.requireNonNull;
12
13 import com.google.common.base.Predicates;
14 import com.google.common.collect.Collections2;
15 import com.google.common.collect.Lists;
16 import com.google.common.collect.Sets;
17 import com.google.common.util.concurrent.FutureCallback;
18 import com.google.common.util.concurrent.Futures;
19 import com.google.common.util.concurrent.ListenableFuture;
20 import com.google.common.util.concurrent.ListeningExecutorService;
21 import com.google.common.util.concurrent.MoreExecutors;
22 import io.netty.util.concurrent.EventExecutor;
23 import java.util.ArrayList;
24 import java.util.Collection;
25 import java.util.Collections;
26 import java.util.LinkedList;
27 import java.util.List;
28 import java.util.Optional;
29 import java.util.Set;
30 import java.util.concurrent.Callable;
31 import java.util.concurrent.ExecutionException;
32 import java.util.concurrent.TimeUnit;
33 import java.util.stream.Collectors;
34 import org.checkerframework.checker.lock.qual.GuardedBy;
35 import org.opendaylight.mdsal.dom.api.DOMRpcResult;
36 import org.opendaylight.mdsal.dom.api.DOMRpcService;
37 import org.opendaylight.netconf.api.NetconfMessage;
38 import org.opendaylight.netconf.sal.connect.api.DeviceActionFactory;
39 import org.opendaylight.netconf.sal.connect.api.MessageTransformer;
40 import org.opendaylight.netconf.sal.connect.api.NetconfDeviceSchemas;
41 import org.opendaylight.netconf.sal.connect.api.NetconfDeviceSchemasResolver;
42 import org.opendaylight.netconf.sal.connect.api.RemoteDevice;
43 import org.opendaylight.netconf.sal.connect.api.RemoteDeviceCommunicator;
44 import org.opendaylight.netconf.sal.connect.api.RemoteDeviceHandler;
45 import org.opendaylight.netconf.sal.connect.netconf.listener.NetconfDeviceCapabilities;
46 import org.opendaylight.netconf.sal.connect.netconf.listener.NetconfDeviceCommunicator;
47 import org.opendaylight.netconf.sal.connect.netconf.listener.NetconfSessionPreferences;
48 import org.opendaylight.netconf.sal.connect.netconf.sal.NetconfDeviceRpc;
49 import org.opendaylight.netconf.sal.connect.netconf.schema.NetconfRemoteSchemaYangSourceProvider;
50 import org.opendaylight.netconf.sal.connect.netconf.schema.YangLibrarySchemaYangSourceProvider;
51 import org.opendaylight.netconf.sal.connect.netconf.schema.mapping.BaseSchema;
52 import org.opendaylight.netconf.sal.connect.netconf.schema.mapping.NetconfMessageTransformer;
53 import org.opendaylight.netconf.sal.connect.netconf.util.NetconfMessageTransformUtil;
54 import org.opendaylight.netconf.sal.connect.util.RemoteDeviceId;
55 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.netconf.notifications.rev120206.NetconfCapabilityChange;
56 import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.node.optional.rev190614.NetconfNodeAugmentedOptional;
57 import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.node.topology.rev150114.NetconfNode;
58 import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.node.topology.rev150114.netconf.node.connection.status.available.capabilities.AvailableCapabilityBuilder;
59 import org.opendaylight.yang.gen.v1.urn.opendaylight.netconf.node.topology.rev150114.netconf.node.connection.status.unavailable.capabilities.UnavailableCapability;
60 import org.opendaylight.yangtools.yang.common.QName;
61 import org.opendaylight.yangtools.yang.model.api.SchemaContext;
62 import org.opendaylight.yangtools.yang.model.repo.api.MissingSchemaSourceException;
63 import org.opendaylight.yangtools.yang.model.repo.api.RevisionSourceIdentifier;
64 import org.opendaylight.yangtools.yang.model.repo.api.SchemaContextFactory;
65 import org.opendaylight.yangtools.yang.model.repo.api.SchemaRepository;
66 import org.opendaylight.yangtools.yang.model.repo.api.SchemaResolutionException;
67 import org.opendaylight.yangtools.yang.model.repo.api.SourceIdentifier;
68 import org.opendaylight.yangtools.yang.model.repo.api.YangTextSchemaSource;
69 import org.opendaylight.yangtools.yang.model.repo.spi.PotentialSchemaSource;
70 import org.opendaylight.yangtools.yang.model.repo.spi.SchemaSourceProvider;
71 import org.opendaylight.yangtools.yang.model.repo.spi.SchemaSourceRegistration;
72 import org.opendaylight.yangtools.yang.model.repo.spi.SchemaSourceRegistry;
73 import org.slf4j.Logger;
74 import org.slf4j.LoggerFactory;
75
76 /**
77  *  This is a mediator between NetconfDeviceCommunicator and NetconfDeviceSalFacade.
78  */
79 public class NetconfDevice
80         implements RemoteDevice<NetconfSessionPreferences, NetconfMessage, NetconfDeviceCommunicator> {
81
82     private static final Logger LOG = LoggerFactory.getLogger(NetconfDevice.class);
83
84     protected final RemoteDeviceId id;
85     protected final SchemaContextFactory schemaContextFactory;
86     protected final SchemaSourceRegistry schemaRegistry;
87     protected final SchemaRepository schemaRepository;
88
89     protected final List<SchemaSourceRegistration<?>> sourceRegistrations = new ArrayList<>();
90
91     private final RemoteDeviceHandler<NetconfSessionPreferences> salFacade;
92     private final ListeningExecutorService processingExecutor;
93     private final DeviceActionFactory deviceActionFactory;
94     private final NetconfDeviceSchemasResolver stateSchemasResolver;
95     private final NotificationHandler notificationHandler;
96     private final boolean reconnectOnSchemasChange;
97     private final NetconfNode node;
98     private final EventExecutor eventExecutor;
99     private final NetconfNodeAugmentedOptional nodeOptional;
100
101     @GuardedBy("this")
102     private boolean connected = false;
103
104     // Message transformer is constructed once the schemas are available
105     private MessageTransformer<NetconfMessage> messageTransformer;
106
107     /**
108      * Create rpc implementation capable of handling RPC for monitoring and notifications
109      * even before the schemas of remote device are downloaded.
110      */
111     static NetconfDeviceRpc getRpcForInitialization(final NetconfDeviceCommunicator listener,
112                                                     final boolean notificationSupport) {
113         final BaseSchema baseSchema = resolveBaseSchema(notificationSupport);
114
115         return new NetconfDeviceRpc(baseSchema.getSchemaContext(), listener,
116                 new NetconfMessageTransformer(baseSchema.getSchemaContext(), false, baseSchema));
117     }
118
119     private static BaseSchema resolveBaseSchema(final boolean notificationSupport) {
120         return notificationSupport ? BaseSchema.BASE_NETCONF_CTX_WITH_NOTIFICATIONS : BaseSchema.BASE_NETCONF_CTX;
121     }
122
123     public NetconfDevice(final SchemaResourcesDTO schemaResourcesDTO, final RemoteDeviceId id,
124                          final RemoteDeviceHandler<NetconfSessionPreferences> salFacade,
125                          final ListeningExecutorService globalProcessingExecutor,
126                          final boolean reconnectOnSchemasChange) {
127         this(schemaResourcesDTO, id, salFacade, globalProcessingExecutor, reconnectOnSchemasChange, null, null, null,
128                 null);
129     }
130
131     public NetconfDevice(final SchemaResourcesDTO schemaResourcesDTO, final RemoteDeviceId id,
132             final RemoteDeviceHandler<NetconfSessionPreferences> salFacade,
133             final ListeningExecutorService globalProcessingExecutor, final boolean reconnectOnSchemasChange,
134             final DeviceActionFactory deviceActionFactory, final NetconfNode node, final EventExecutor eventExecutor,
135             final NetconfNodeAugmentedOptional nodeOptional) {
136         this.id = id;
137         this.reconnectOnSchemasChange = reconnectOnSchemasChange;
138         this.deviceActionFactory = deviceActionFactory;
139         this.node = node;
140         this.eventExecutor = eventExecutor;
141         this.nodeOptional = nodeOptional;
142         this.schemaRegistry = schemaResourcesDTO.getSchemaRegistry();
143         this.schemaRepository = schemaResourcesDTO.getSchemaRepository();
144         this.schemaContextFactory = schemaResourcesDTO.getSchemaContextFactory();
145         this.salFacade = salFacade;
146         this.stateSchemasResolver = schemaResourcesDTO.getStateSchemasResolver();
147         this.processingExecutor = requireNonNull(globalProcessingExecutor);
148         this.notificationHandler = new NotificationHandler(salFacade, id);
149     }
150
151     @Override
152     public void onRemoteSessionUp(final NetconfSessionPreferences remoteSessionCapabilities,
153                                   final NetconfDeviceCommunicator listener) {
154         // SchemaContext setup has to be performed in a dedicated thread since
155         // we are in a netty thread in this method
156         // Yang models are being downloaded in this method and it would cause a
157         // deadlock if we used the netty thread
158         // http://netty.io/wiki/thread-model.html
159         setConnected(true);
160         LOG.debug("{}: Session to remote device established with {}", id, remoteSessionCapabilities);
161
162         final NetconfDeviceRpc initRpc =
163                 getRpcForInitialization(listener, remoteSessionCapabilities.isNotificationsSupported());
164         final DeviceSourcesResolver task =
165                 new DeviceSourcesResolver(remoteSessionCapabilities, id, stateSchemasResolver, initRpc,
166                         resolveBaseSchema(remoteSessionCapabilities.isNotificationsSupported()).getSchemaContext());
167         final ListenableFuture<DeviceSources> sourceResolverFuture = processingExecutor.submit(task);
168
169         if (shouldListenOnSchemaChange(remoteSessionCapabilities)) {
170             registerToBaseNetconfStream(initRpc, listener);
171         }
172
173         final FutureCallback<DeviceSources> resolvedSourceCallback = new FutureCallback<DeviceSources>() {
174             @Override
175             public void onSuccess(final DeviceSources result) {
176                 addProvidedSourcesToSchemaRegistry(result);
177                 setUpSchema(result);
178             }
179
180             private void setUpSchema(final DeviceSources result) {
181                 processingExecutor.submit(new SchemaSetup(result, remoteSessionCapabilities, listener));
182             }
183
184             @Override
185             public void onFailure(final Throwable throwable) {
186                 LOG.warn("{}: Unexpected error resolving device sources", id, throwable);
187                 handleSalInitializationFailure(throwable, listener);
188             }
189         };
190
191         Futures.addCallback(sourceResolverFuture, resolvedSourceCallback, MoreExecutors.directExecutor());
192     }
193
194     private void registerToBaseNetconfStream(final NetconfDeviceRpc deviceRpc,
195                                              final NetconfDeviceCommunicator listener) {
196         // TODO check whether the model describing create subscription is present in schema
197         // Perhaps add a default schema context to support create-subscription if the model was not provided
198         // (same as what we do for base netconf operations in transformer)
199         final ListenableFuture<DOMRpcResult> rpcResultListenableFuture = deviceRpc.invokeRpc(
200                 NetconfMessageTransformUtil.CREATE_SUBSCRIPTION_RPC_PATH,
201                 NetconfMessageTransformUtil.CREATE_SUBSCRIPTION_RPC_CONTENT);
202
203         Futures.addCallback(rpcResultListenableFuture, new FutureCallback<DOMRpcResult>() {
204             @Override
205             public void onSuccess(final DOMRpcResult domRpcResult) {
206                 notificationHandler.addNotificationFilter(notification -> {
207                     if (NetconfCapabilityChange.QNAME.equals(notification.getBody().getNodeType())) {
208                         LOG.info("{}: Schemas change detected, reconnecting", id);
209                         // Only disconnect is enough,
210                         // the reconnecting nature of the connector will take care of reconnecting
211                         listener.disconnect();
212                         return Optional.empty();
213                     }
214                     return Optional.of(notification);
215                 });
216             }
217
218             @Override
219             public void onFailure(final Throwable throwable) {
220                 LOG.warn("Unable to subscribe to base notification stream. Schemas will not be reloaded on the fly",
221                         throwable);
222             }
223         }, MoreExecutors.directExecutor());
224     }
225
226     private boolean shouldListenOnSchemaChange(final NetconfSessionPreferences remoteSessionCapabilities) {
227         return remoteSessionCapabilities.isNotificationsSupported() && reconnectOnSchemasChange;
228     }
229
230     private synchronized void handleSalInitializationSuccess(final SchemaContext result,
231                                         final NetconfSessionPreferences remoteSessionCapabilities,
232                                         final DOMRpcService deviceRpc,
233                                         final RemoteDeviceCommunicator<NetconfMessage> listener) {
234         //NetconfDevice.SchemaSetup can complete after NetconfDeviceCommunicator was closed. In that case do nothing,
235         //since salFacade.onDeviceDisconnected was already called.
236         if (connected) {
237             final BaseSchema baseSchema =
238                 remoteSessionCapabilities.isNotificationsSupported()
239                         ? BaseSchema.BASE_NETCONF_CTX_WITH_NOTIFICATIONS : BaseSchema.BASE_NETCONF_CTX;
240             this.messageTransformer = new NetconfMessageTransformer(result, true, baseSchema);
241
242             // salFacade.onDeviceConnected has to be called before the notification handler is initialized
243             this.salFacade.onDeviceConnected(result, remoteSessionCapabilities, deviceRpc,
244                     this.deviceActionFactory == null ? null : this.deviceActionFactory.createDeviceAction(
245                             this.messageTransformer, listener, result));
246             this.notificationHandler.onRemoteSchemaUp(this.messageTransformer);
247
248             LOG.info("{}: Netconf connector initialized successfully", id);
249         } else {
250             LOG.warn("{}: Device communicator was closed before schema setup finished.", id);
251         }
252     }
253
254     private void handleSalInitializationFailure(final Throwable throwable,
255                                                 final RemoteDeviceCommunicator<NetconfMessage> listener) {
256         LOG.error("{}: Initialization in sal failed, disconnecting from device", id, throwable);
257         listener.close();
258         onRemoteSessionDown();
259         resetMessageTransformer();
260     }
261
262     /**
263      * Set the transformer to null as is in initial state.
264      */
265     private void resetMessageTransformer() {
266         updateTransformer(null);
267     }
268
269     private synchronized void updateTransformer(final MessageTransformer<NetconfMessage> transformer) {
270         messageTransformer = transformer;
271     }
272
273     private synchronized void setConnected(final boolean connected) {
274         this.connected = connected;
275     }
276
277     private void addProvidedSourcesToSchemaRegistry(final DeviceSources deviceSources) {
278         final SchemaSourceProvider<YangTextSchemaSource> yangProvider = deviceSources.getSourceProvider();
279         for (final SourceIdentifier sourceId : deviceSources.getProvidedSources()) {
280             sourceRegistrations.add(schemaRegistry.registerSchemaSource(yangProvider,
281                     PotentialSchemaSource.create(
282                             sourceId, YangTextSchemaSource.class, PotentialSchemaSource.Costs.REMOTE_IO.getValue())));
283         }
284     }
285
286     @Override
287     public void onRemoteSessionDown() {
288         setConnected(false);
289         notificationHandler.onRemoteSchemaDown();
290
291         salFacade.onDeviceDisconnected();
292         sourceRegistrations.forEach(SchemaSourceRegistration::close);
293         sourceRegistrations.clear();
294         resetMessageTransformer();
295     }
296
297     @Override
298     public void onRemoteSessionFailed(final Throwable throwable) {
299         setConnected(false);
300         salFacade.onDeviceFailed(throwable);
301     }
302
303     @Override
304     public void onNotification(final NetconfMessage notification) {
305         notificationHandler.handleNotification(notification);
306     }
307
308     /**
309      * Just a transfer object containing schema related dependencies. Injected in constructor.
310      */
311     public static class SchemaResourcesDTO {
312         private final SchemaSourceRegistry schemaRegistry;
313         private final SchemaRepository schemaRepository;
314         private final SchemaContextFactory schemaContextFactory;
315         private final NetconfDeviceSchemasResolver stateSchemasResolver;
316
317         public SchemaResourcesDTO(final SchemaSourceRegistry schemaRegistry,
318                                   final SchemaRepository schemaRepository,
319                                   final SchemaContextFactory schemaContextFactory,
320                                   final NetconfDeviceSchemasResolver deviceSchemasResolver) {
321             this.schemaRegistry = requireNonNull(schemaRegistry);
322             this.schemaRepository = requireNonNull(schemaRepository);
323             this.schemaContextFactory = requireNonNull(schemaContextFactory);
324             this.stateSchemasResolver = requireNonNull(deviceSchemasResolver);
325         }
326
327         public SchemaSourceRegistry getSchemaRegistry() {
328             return schemaRegistry;
329         }
330
331         public SchemaRepository getSchemaRepository() {
332             return schemaRepository;
333         }
334
335         public SchemaContextFactory getSchemaContextFactory() {
336             return schemaContextFactory;
337         }
338
339         public NetconfDeviceSchemasResolver getStateSchemasResolver() {
340             return stateSchemasResolver;
341         }
342     }
343
344     /**
345      * Schema building callable.
346      */
347     private static class DeviceSourcesResolver implements Callable<DeviceSources> {
348
349         private final NetconfDeviceRpc deviceRpc;
350         private final NetconfSessionPreferences remoteSessionCapabilities;
351         private final RemoteDeviceId id;
352         private final NetconfDeviceSchemasResolver stateSchemasResolver;
353         private final SchemaContext schemaContext;
354
355         DeviceSourcesResolver(final NetconfDeviceRpc deviceRpc,
356                               final NetconfSessionPreferences remoteSessionCapabilities,
357                               final RemoteDeviceId id, final NetconfDeviceSchemasResolver stateSchemasResolver,
358                               final SchemaContext schemaContext) {
359             this.deviceRpc = deviceRpc;
360             this.remoteSessionCapabilities = remoteSessionCapabilities;
361             this.id = id;
362             this.stateSchemasResolver = stateSchemasResolver;
363             this.schemaContext = schemaContext;
364         }
365
366         DeviceSourcesResolver(final NetconfSessionPreferences remoteSessionCapabilities, final RemoteDeviceId id,
367                                      final NetconfDeviceSchemasResolver stateSchemasResolver,
368                                      final NetconfDeviceRpc rpcForMonitoring, final SchemaContext schemaCtx) {
369             this(rpcForMonitoring, remoteSessionCapabilities, id, stateSchemasResolver, schemaCtx);
370         }
371
372         @Override
373         public DeviceSources call() {
374             final NetconfDeviceSchemas availableSchemas =
375                     stateSchemasResolver.resolve(deviceRpc, remoteSessionCapabilities, id, schemaContext);
376             LOG.debug("{}: Schemas exposed by ietf-netconf-monitoring: {}", id,
377                     availableSchemas.getAvailableYangSchemasQNames());
378
379             final Set<QName> requiredSources = Sets.newHashSet(remoteSessionCapabilities.getModuleBasedCaps());
380             final Set<QName> providedSources = availableSchemas.getAvailableYangSchemasQNames();
381
382             final Set<QName> requiredSourcesNotProvided = Sets.difference(requiredSources, providedSources);
383             if (!requiredSourcesNotProvided.isEmpty()) {
384                 LOG.warn("{}: Netconf device does not provide all yang models reported in hello message capabilities,"
385                         + " required but not provided: {}", id, requiredSourcesNotProvided);
386                 LOG.warn("{}: Attempting to build schema context from required sources", id);
387             }
388
389             // Here all the sources reported in netconf monitoring are merged with those reported in hello.
390             // It is necessary to perform this since submodules are not mentioned in hello but still required.
391             // This clashes with the option of a user to specify supported yang models manually in configuration
392             // for netconf-connector and as a result one is not able to fully override yang models of a device.
393             // It is only possible to add additional models.
394             final Set<QName> providedSourcesNotRequired = Sets.difference(providedSources, requiredSources);
395             if (!providedSourcesNotRequired.isEmpty()) {
396                 LOG.warn("{}: Netconf device provides additional yang models not reported in "
397                         + "hello message capabilities: {}", id, providedSourcesNotRequired);
398                 LOG.warn("{}: Adding provided but not required sources as required to prevent failures", id);
399                 LOG.debug("{}: Netconf device reported in hello: {}", id, requiredSources);
400                 requiredSources.addAll(providedSourcesNotRequired);
401             }
402
403             final SchemaSourceProvider<YangTextSchemaSource> sourceProvider;
404             if (availableSchemas instanceof LibraryModulesSchemas) {
405                 sourceProvider = new YangLibrarySchemaYangSourceProvider(id,
406                         ((LibraryModulesSchemas) availableSchemas).getAvailableModels());
407             } else {
408                 sourceProvider = new NetconfRemoteSchemaYangSourceProvider(id, deviceRpc);
409             }
410
411             return new DeviceSources(requiredSources, providedSources, sourceProvider);
412         }
413     }
414
415     /**
416      * Contains RequiredSources - sources from capabilities.
417      */
418     private static final class DeviceSources {
419         private final Set<QName> requiredSources;
420         private final Set<QName> providedSources;
421         private final SchemaSourceProvider<YangTextSchemaSource> sourceProvider;
422
423         DeviceSources(final Set<QName> requiredSources, final Set<QName> providedSources,
424                              final SchemaSourceProvider<YangTextSchemaSource> sourceProvider) {
425             this.requiredSources = requiredSources;
426             this.providedSources = providedSources;
427             this.sourceProvider = sourceProvider;
428         }
429
430         public Set<QName> getRequiredSourcesQName() {
431             return requiredSources;
432         }
433
434         public Set<QName> getProvidedSourcesQName() {
435             return providedSources;
436         }
437
438         public Collection<SourceIdentifier> getRequiredSources() {
439             return Collections2.transform(requiredSources, DeviceSources::toSourceId);
440         }
441
442         public Collection<SourceIdentifier> getProvidedSources() {
443             return Collections2.transform(providedSources, DeviceSources::toSourceId);
444         }
445
446         public SchemaSourceProvider<YangTextSchemaSource> getSourceProvider() {
447             return sourceProvider;
448         }
449
450         private static SourceIdentifier toSourceId(final QName input) {
451             return RevisionSourceIdentifier.create(input.getLocalName(), input.getRevision());
452         }
453     }
454
455     /**
456      * Schema builder that tries to build schema context from provided sources or biggest subset of it.
457      */
458     private final class SchemaSetup implements Runnable {
459         private final DeviceSources deviceSources;
460         private final NetconfSessionPreferences remoteSessionCapabilities;
461         private final RemoteDeviceCommunicator<NetconfMessage> listener;
462         private final NetconfDeviceCapabilities capabilities;
463
464         SchemaSetup(final DeviceSources deviceSources, final NetconfSessionPreferences remoteSessionCapabilities,
465                            final RemoteDeviceCommunicator<NetconfMessage> listener) {
466             this.deviceSources = deviceSources;
467             this.remoteSessionCapabilities = remoteSessionCapabilities;
468             this.listener = listener;
469             this.capabilities = remoteSessionCapabilities.getNetconfDeviceCapabilities();
470         }
471
472         @Override
473         public void run() {
474
475             final Collection<SourceIdentifier> requiredSources = deviceSources.getRequiredSources();
476             final Collection<SourceIdentifier> missingSources = filterMissingSources(requiredSources);
477
478             capabilities.addUnresolvedCapabilities(getQNameFromSourceIdentifiers(missingSources),
479                     UnavailableCapability.FailureReason.MissingSource);
480
481             requiredSources.removeAll(missingSources);
482             setUpSchema(requiredSources);
483         }
484
485         private Collection<SourceIdentifier> filterMissingSources(final Collection<SourceIdentifier> requiredSources) {
486             return requiredSources.parallelStream().filter(sourceIdentifier -> {
487                 try {
488                     schemaRepository.getSchemaSource(sourceIdentifier, YangTextSchemaSource.class).get();
489                     return false;
490                 } catch (InterruptedException | ExecutionException e) {
491                     return true;
492                 }
493             }).collect(Collectors.toList());
494         }
495
496         /**
497          * Build schema context, in case of success or final failure notify device.
498          *
499          * @param requiredSources required sources
500          */
501         @SuppressWarnings("checkstyle:IllegalCatch")
502         private void setUpSchema(Collection<SourceIdentifier> requiredSources) {
503             while (!requiredSources.isEmpty()) {
504                 LOG.trace("{}: Trying to build schema context from {}", id, requiredSources);
505                 try {
506                     final ListenableFuture<SchemaContext> schemaBuilderFuture = schemaContextFactory
507                             .createSchemaContext(requiredSources);
508                     final SchemaContext result = schemaBuilderFuture.get();
509                     LOG.debug("{}: Schema context built successfully from {}", id, requiredSources);
510                     final Collection<QName> filteredQNames = Sets.difference(deviceSources.getRequiredSourcesQName(),
511                             capabilities.getUnresolvedCapabilites().keySet());
512                     capabilities.addCapabilities(filteredQNames.stream().map(entry -> new AvailableCapabilityBuilder()
513                             .setCapability(entry.toString()).setCapabilityOrigin(
514                                     remoteSessionCapabilities.getModuleBasedCapsOrigin().get(entry)).build())
515                             .collect(Collectors.toList()));
516
517                     capabilities.addNonModuleBasedCapabilities(remoteSessionCapabilities
518                             .getNonModuleCaps().stream().map(entry -> new AvailableCapabilityBuilder()
519                                     .setCapability(entry).setCapabilityOrigin(
520                                             remoteSessionCapabilities.getNonModuleBasedCapsOrigin().get(entry)).build())
521                             .collect(Collectors.toList()));
522
523                     handleSalInitializationSuccess(result, remoteSessionCapabilities, getDeviceSpecificRpc(result),
524                             listener);
525                     return;
526                 } catch (final ExecutionException e) {
527                     // schemaBuilderFuture.checkedGet() throws only SchemaResolutionException
528                     // that might be wrapping a MissingSchemaSourceException so we need to look
529                     // at the cause of the exception to make sure we don't misinterpret it.
530                     final Throwable cause = e.getCause();
531
532                     if (cause instanceof MissingSchemaSourceException) {
533                         requiredSources = handleMissingSchemaSourceException(
534                                 requiredSources, (MissingSchemaSourceException) cause);
535                         continue;
536                     }
537                     if (cause instanceof SchemaResolutionException) {
538                         requiredSources = handleSchemaResolutionException(requiredSources,
539                             (SchemaResolutionException) cause);
540                     } else {
541                         handleSalInitializationFailure(e, listener);
542                         return;
543                     }
544                 } catch (final Exception e) {
545                     // unknown error, fail
546                     handleSalInitializationFailure(e, listener);
547                     return;
548                 }
549             }
550             // No more sources, fail or try to reconnect
551             if (nodeOptional != null && nodeOptional.getIgnoreMissingSchemaSources().isAllowed()) {
552                 eventExecutor.schedule(() -> {
553                     LOG.warn("Reconnection is allowed! This can lead to unexpected errors at runtime.");
554                     LOG.warn("{} : No more sources for schema context.", id);
555                     LOG.info("{} : Try to remount device.", id);
556                     onRemoteSessionDown();
557                     salFacade.onDeviceReconnected(remoteSessionCapabilities, node);
558                 }, nodeOptional.getIgnoreMissingSchemaSources().getReconnectTime(), TimeUnit.MILLISECONDS);
559             } else {
560                 final IllegalStateException cause =
561                         new IllegalStateException(id + ": No more sources for schema context");
562                 handleSalInitializationFailure(cause, listener);
563                 salFacade.onDeviceFailed(cause);
564             }
565         }
566
567         private Collection<SourceIdentifier> handleMissingSchemaSourceException(
568                 final Collection<SourceIdentifier> requiredSources, final MissingSchemaSourceException exception) {
569             // In case source missing, try without it
570             final SourceIdentifier missingSource = exception.getSourceId();
571             LOG.warn("{}: Unable to build schema context, missing source {}, will reattempt without it",
572                 id, missingSource);
573             LOG.debug("{}: Unable to build schema context, missing source {}, will reattempt without it",
574                 id, missingSource, exception);
575             final Collection<QName> qNameOfMissingSource =
576                 getQNameFromSourceIdentifiers(Sets.newHashSet(missingSource));
577             if (!qNameOfMissingSource.isEmpty()) {
578                 capabilities.addUnresolvedCapabilities(
579                         qNameOfMissingSource, UnavailableCapability.FailureReason.MissingSource);
580             }
581             return stripUnavailableSource(requiredSources, missingSource);
582         }
583
584         private Collection<SourceIdentifier> handleSchemaResolutionException(
585             final Collection<SourceIdentifier> requiredSources, final SchemaResolutionException resolutionException) {
586             // In case resolution error, try only with resolved sources
587             // There are two options why schema resolution exception occurred : unsatisfied imports or flawed model
588             // FIXME Do we really have assurance that these two cases cannot happen at once?
589             if (resolutionException.getFailedSource() != null) {
590                 // flawed model - exclude it
591                 final SourceIdentifier failedSourceId = resolutionException.getFailedSource();
592                 LOG.warn("{}: Unable to build schema context, failed to resolve source {}, will reattempt without it",
593                     id, failedSourceId);
594                 LOG.warn("{}: Unable to build schema context, failed to resolve source {}, will reattempt without it",
595                     id, failedSourceId, resolutionException);
596                 capabilities.addUnresolvedCapabilities(
597                         getQNameFromSourceIdentifiers(Collections.singleton(failedSourceId)),
598                         UnavailableCapability.FailureReason.UnableToResolve);
599                 return stripUnavailableSource(requiredSources, resolutionException.getFailedSource());
600             }
601             // unsatisfied imports
602             final Set<SourceIdentifier> unresolvedSources = resolutionException.getUnsatisfiedImports().keySet();
603             capabilities.addUnresolvedCapabilities(getQNameFromSourceIdentifiers(unresolvedSources),
604                 UnavailableCapability.FailureReason.UnableToResolve);
605             LOG.warn("{}: Unable to build schema context, unsatisfied imports {}, will reattempt with resolved only",
606                 id, resolutionException.getUnsatisfiedImports());
607             LOG.debug("{}: Unable to build schema context, unsatisfied imports {}, will reattempt with resolved only",
608                 id, resolutionException.getUnsatisfiedImports(), resolutionException);
609             return resolutionException.getResolvedSources();
610         }
611
612         protected NetconfDeviceRpc getDeviceSpecificRpc(final SchemaContext result) {
613             return new NetconfDeviceRpc(result, listener, new NetconfMessageTransformer(result, true));
614         }
615
616         private Collection<SourceIdentifier> stripUnavailableSource(final Collection<SourceIdentifier> requiredSources,
617                                                                     final SourceIdentifier sourceIdToRemove) {
618             final LinkedList<SourceIdentifier> sourceIdentifiers = Lists.newLinkedList(requiredSources);
619             checkState(sourceIdentifiers.remove(sourceIdToRemove),
620                     "%s: Trying to remove %s from %s failed", id, sourceIdToRemove, requiredSources);
621             return sourceIdentifiers;
622         }
623
624         private Collection<QName> getQNameFromSourceIdentifiers(final Collection<SourceIdentifier> identifiers) {
625             final Collection<QName> qNames = Collections2.transform(identifiers, this::getQNameFromSourceIdentifier);
626
627             if (qNames.isEmpty()) {
628                 LOG.debug("{}: Unable to map any source identifiers to a capability reported by device : {}", id,
629                         identifiers);
630             }
631             return Collections2.filter(qNames, Predicates.notNull());
632         }
633
634         private QName getQNameFromSourceIdentifier(final SourceIdentifier identifier) {
635             // Required sources are all required and provided merged in DeviceSourcesResolver
636             for (final QName qname : deviceSources.getRequiredSourcesQName()) {
637                 if (!qname.getLocalName().equals(identifier.getName())) {
638                     continue;
639                 }
640
641                 if (identifier.getRevision().equals(qname.getRevision())) {
642                     return qname;
643                 }
644             }
645             LOG.warn("Unable to map identifier to a devices reported capability: {} Available: {}",identifier,
646                     deviceSources.getRequiredSourcesQName());
647             // return null since we cannot find the QName,
648             // this capability will be removed from required sources and not reported as unresolved-capability
649             return null;
650         }
651     }
652 }