Integrate AbstractCommonSubscriber more tightly
[netconf.git] / restconf / restconf-nb-rfc8040 / src / main / java / org / opendaylight / restconf / nb / rfc8040 / streams / listeners / ListenerAdapter.java
1 /*
2  * Copyright (c) 2014, 2016 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.restconf.nb.rfc8040.streams.listeners;
9
10 import com.google.common.annotations.VisibleForTesting;
11 import java.time.Instant;
12 import java.util.Collection;
13 import java.util.Optional;
14 import java.util.stream.Collectors;
15 import org.opendaylight.mdsal.common.api.LogicalDatastoreType;
16 import org.opendaylight.mdsal.dom.api.ClusteredDOMDataTreeChangeListener;
17 import org.opendaylight.mdsal.dom.api.DOMDataBroker;
18 import org.opendaylight.mdsal.dom.api.DOMDataTreeChangeService;
19 import org.opendaylight.mdsal.dom.api.DOMDataTreeIdentifier;
20 import org.opendaylight.restconf.common.formatters.DataTreeCandidateFormatterFactory;
21 import org.opendaylight.restconf.common.formatters.JSONDataTreeCandidateFormatter;
22 import org.opendaylight.restconf.common.formatters.XMLDataTreeCandidateFormatter;
23 import org.opendaylight.yang.gen.v1.urn.sal.restconf.event.subscription.rev140708.NotificationOutputTypeGrouping.NotificationOutputType;
24 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
25 import org.opendaylight.yangtools.yang.data.api.schema.tree.DataTreeCandidate;
26 import org.opendaylight.yangtools.yang.data.codec.gson.JSONCodecFactorySupplier;
27 import org.slf4j.Logger;
28 import org.slf4j.LoggerFactory;
29
30 /**
31  * {@link ListenerAdapter} is responsible to track events, which occurred by changing data in data source.
32  */
33 public class ListenerAdapter extends AbstractCommonSubscriber<YangInstanceIdentifier, Collection<DataTreeCandidate>>
34         implements ClusteredDOMDataTreeChangeListener {
35     private static final Logger LOG = LoggerFactory.getLogger(ListenerAdapter.class);
36     private static final DataTreeCandidateFormatterFactory JSON_FORMATTER_FACTORY =
37             JSONDataTreeCandidateFormatter.createFactory(JSONCodecFactorySupplier.RFC7951);
38
39     /**
40      * Creates new {@link ListenerAdapter} listener specified by path and stream name and register for subscribing.
41      *
42      * @param path       Path to data in data store.
43      * @param streamName The name of the stream.
44      * @param outputType Type of output on notification (JSON, XML).
45      */
46     @VisibleForTesting
47     public ListenerAdapter(final YangInstanceIdentifier path, final String streamName,
48             final NotificationOutputType outputType) {
49         super(path.getLastPathArgument().getNodeType(), streamName, path, outputType, getFormatterFactory(outputType));
50     }
51
52     private static DataTreeCandidateFormatterFactory getFormatterFactory(final NotificationOutputType outputType) {
53         switch (outputType) {
54             case JSON:
55                 return JSON_FORMATTER_FACTORY;
56             case XML:
57                 return XMLDataTreeCandidateFormatter.FACTORY;
58             default:
59                 throw new IllegalArgumentException("Unsupported outputType" + outputType);
60         }
61     }
62
63     @Override
64     public void onInitialData() {
65         // No-op
66     }
67
68     @Override
69     @SuppressWarnings("checkstyle:IllegalCatch")
70     public void onDataTreeChanged(final Collection<DataTreeCandidate> dataTreeCandidates) {
71         final Instant now = Instant.now();
72         if (!checkStartStop(now)) {
73             return;
74         }
75
76         final Optional<String> maybeData;
77         try {
78             maybeData = formatter().eventData(schemaHandler.get(), dataTreeCandidates, now, getLeafNodesOnly(),
79                     isSkipNotificationData());
80         } catch (final Exception e) {
81             LOG.error("Failed to process notification {}",
82                     dataTreeCandidates.stream().map(Object::toString).collect(Collectors.joining(",")), e);
83             return;
84         }
85
86         if (maybeData.isPresent()) {
87             post(maybeData.get());
88         }
89     }
90
91     /**
92      * Get path pointed to data in data store.
93      *
94      * @return Path pointed to data in data store.
95      */
96     public YangInstanceIdentifier getPath() {
97         return path();
98     }
99
100     /**
101      * Register data change listener in DOM data broker and set it to listener on stream.
102      *
103      * @param domDataBroker data broker for register data change listener
104      * @param datastore     {@link LogicalDatastoreType}
105      */
106     public final synchronized void listen(final DOMDataBroker domDataBroker, final LogicalDatastoreType datastore) {
107         if (!isListening()) {
108             final DOMDataTreeChangeService changeService = domDataBroker.getExtensions()
109                 .getInstance(DOMDataTreeChangeService.class);
110             if (changeService == null) {
111                 throw new UnsupportedOperationException("DOMDataBroker does not support the DOMDataTreeChangeService");
112             }
113
114             setRegistration(changeService.registerDataTreeChangeListener(
115                 new DOMDataTreeIdentifier(datastore, getPath()), this));
116         }
117     }
118 }