2 * Copyright (c) 2023 PANTHEON.tech, s.r.o. 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
8 package org.opendaylight.restconf.server.mdsal.streams.dtcl;
10 import static java.util.Objects.requireNonNull;
13 import javax.inject.Inject;
14 import javax.inject.Singleton;
15 import org.eclipse.jdt.annotation.NonNull;
16 import org.opendaylight.mdsal.common.api.LogicalDatastoreType;
17 import org.opendaylight.mdsal.dom.api.DOMDataBroker;
18 import org.opendaylight.mdsal.dom.api.DOMDataBroker.DataTreeChangeExtension;
19 import org.opendaylight.restconf.common.errors.RestconfDocumentedException;
20 import org.opendaylight.restconf.common.errors.RestconfFuture;
21 import org.opendaylight.restconf.nb.rfc8040.utils.parser.YangInstanceIdentifierSerializer;
22 import org.opendaylight.restconf.server.api.OperationsPostResult;
23 import org.opendaylight.restconf.server.spi.DatabindProvider;
24 import org.opendaylight.restconf.server.spi.OperationInput;
25 import org.opendaylight.restconf.server.spi.RestconfStream;
26 import org.opendaylight.restconf.server.spi.RpcImplementation;
27 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.remote.rev140114.CreateDataChangeEventSubscription;
28 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.remote.rev140114.CreateDataChangeEventSubscriptionInput;
29 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.controller.md.sal.remote.rev140114.CreateDataChangeEventSubscriptionOutput;
30 import org.opendaylight.yang.gen.v1.urn.sal.restconf.event.subscription.rev231103.CreateDataChangeEventSubscriptionInput1;
31 import org.opendaylight.yangtools.yang.common.ErrorTag;
32 import org.opendaylight.yangtools.yang.common.ErrorType;
33 import org.opendaylight.yangtools.yang.common.QName;
34 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
35 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier.NodeIdentifier;
36 import org.opendaylight.yangtools.yang.data.spi.node.ImmutableNodes;
37 import org.osgi.service.component.annotations.Activate;
38 import org.osgi.service.component.annotations.Component;
39 import org.osgi.service.component.annotations.Reference;
42 * RESTCONF implementation of {@link CreateDataChangeEventSubscription}.
45 @Component(service = RpcImplementation.class)
46 public final class CreateDataChangeEventSubscriptionRpc extends RpcImplementation {
47 private static final @NonNull NodeIdentifier DATASTORE_NODEID = NodeIdentifier.create(
48 QName.create(CreateDataChangeEventSubscriptionInput1.QNAME, "datastore").intern());
49 private static final @NonNull NodeIdentifier STREAM_NAME_NODEID =
50 NodeIdentifier.create(QName.create(CreateDataChangeEventSubscriptionOutput.QNAME, "stream-name").intern());
51 private static final @NonNull NodeIdentifier PATH_NODEID =
52 NodeIdentifier.create(QName.create(CreateDataChangeEventSubscriptionInput.QNAME, "path").intern());
53 private static final @NonNull NodeIdentifier OUTPUT_NODEID =
54 NodeIdentifier.create(CreateDataChangeEventSubscriptionOutput.QNAME);
56 private final DatabindProvider databindProvider;
57 private final DataTreeChangeExtension changeService;
58 private final RestconfStream.Registry streamRegistry;
62 public CreateDataChangeEventSubscriptionRpc(@Reference final RestconfStream.Registry streamRegistry,
63 @Reference final DatabindProvider databindProvider, @Reference final DOMDataBroker dataBroker) {
64 super(CreateDataChangeEventSubscription.QNAME);
65 this.databindProvider = requireNonNull(databindProvider);
66 changeService = dataBroker.extension(DataTreeChangeExtension.class);
67 if (changeService == null) {
68 throw new UnsupportedOperationException("DOMDataBroker does not support the DOMDataTreeChangeService");
70 this.streamRegistry = requireNonNull(streamRegistry);
74 * Create data-change-event stream with POST operation via RPC.
76 * @param input Input of RPC - example in JSON (data-change-event stream):
81 * "path": "/toaster:toaster/toaster:toasterStatus",
82 * "sal-remote-augment:datastore": "OPERATIONAL",
87 * @return Future output of RPC - example in JSON:
92 * "stream-name": "toaster:toaster/toaster:toasterStatus/datastore=OPERATIONAL/scope=ONE"
99 public RestconfFuture<OperationsPostResult> invoke(final URI restconfURI, final OperationInput input) {
100 final var body = input.input();
101 final var datastoreName = leaf(body, DATASTORE_NODEID, String.class);
102 final var datastore = datastoreName != null ? LogicalDatastoreType.valueOf(datastoreName)
103 : LogicalDatastoreType.CONFIGURATION;
105 final var path = leaf(body, PATH_NODEID, YangInstanceIdentifier.class);
107 return RestconfFuture.failed(
108 new RestconfDocumentedException("missing path", ErrorType.APPLICATION, ErrorTag.MISSING_ELEMENT));
111 return streamRegistry.createStream(restconfURI,
112 new DataTreeChangeSource(databindProvider, changeService, datastore, path),
113 "Events occuring in " + datastore + " datastore under /"
114 + new YangInstanceIdentifierSerializer(input.databind()).serializePath(path))
115 .transform(stream -> input.newOperationOutput(ImmutableNodes.newContainerBuilder()
116 .withNodeIdentifier(OUTPUT_NODEID)
117 .withChild(ImmutableNodes.leafNode(STREAM_NAME_NODEID, stream.name()))