restconf/restconf-nb/src/main/java/org/opendaylight/restconf/server/mdsal/streams/dtcl/CreateDataChangeEventSubscriptionRpc.java

   1 /*
   2  * Copyright (c) 2023 PANTHEON.tech, s.r.o. 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.server.mdsal.streams.dtcl;
   9
  10 import static java.util.Objects.requireNonNull;
  11
  12 import java.net.URI;
  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;
  40
  41 /**
  42  * RESTCONF implementation of {@link CreateDataChangeEventSubscription}.
  43  */
  44 @Singleton
  45 @Component
  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);
  55
  56     private final DatabindProvider databindProvider;
  57     private final DataTreeChangeExtension changeService;
  58     private final RestconfStream.Registry streamRegistry;
  59
  60     @Inject
  61     @Activate
  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");
  69         }
  70         this.streamRegistry = requireNonNull(streamRegistry);
  71     }
  72
  73     /**
  74      * Create data-change-event stream with POST operation via RPC.
  75      *
  76      * @param input Input of RPC - example in JSON (data-change-event stream):
  77      *              <pre>
  78      *              {@code
  79      *                  {
  80      *                      "input": {
  81      *                          "path": "/toaster:toaster/toaster:toasterStatus",
  82      *                          "sal-remote-augment:datastore": "OPERATIONAL",
  83      *                      }
  84      *                  }
  85      *              }
  86      *              </pre>
  87      * @return Future output of RPC - example in JSON:
  88      *     <pre>
  89      *     {@code
  90      *         {
  91      *             "output": {
  92      *                 "stream-name": "toaster:toaster/toaster:toasterStatus/datastore=OPERATIONAL/scope=ONE"
  93      *             }
  94      *         }
  95      *     }
  96      *     </pre>
  97      */
  98     @Override
  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;
 104
 105         final var path = leaf(body, PATH_NODEID, YangInstanceIdentifier.class);
 106         if (path == null) {
 107             return RestconfFuture.failed(
 108                 new RestconfDocumentedException("missing path", ErrorType.APPLICATION, ErrorTag.MISSING_ELEMENT));
 109         }
 110
 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()))
 118                 .build()));
 119     }
 120 }