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.impl.schema.Builders;
  37 import org.opendaylight.yangtools.yang.data.impl.schema.ImmutableNodes;
  38 import org.osgi.service.component.annotations.Activate;
  39 import org.osgi.service.component.annotations.Component;
  40 import org.osgi.service.component.annotations.Reference;
  41
  42 /**
  43  * RESTCONF implementation of {@link CreateDataChangeEventSubscription}.
  44  */
  45 @Singleton
  46 @Component
  47 public final class CreateDataChangeEventSubscriptionRpc extends RpcImplementation {
  48     private static final @NonNull NodeIdentifier DATASTORE_NODEID = NodeIdentifier.create(
  49         QName.create(CreateDataChangeEventSubscriptionInput1.QNAME, "datastore").intern());
  50     private static final @NonNull NodeIdentifier STREAM_NAME_NODEID =
  51         NodeIdentifier.create(QName.create(CreateDataChangeEventSubscriptionOutput.QNAME, "stream-name").intern());
  52     private static final @NonNull NodeIdentifier PATH_NODEID =
  53         NodeIdentifier.create(QName.create(CreateDataChangeEventSubscriptionInput.QNAME, "path").intern());
  54     private static final @NonNull NodeIdentifier OUTPUT_NODEID =
  55         NodeIdentifier.create(CreateDataChangeEventSubscriptionOutput.QNAME);
  56
  57     private final DatabindProvider databindProvider;
  58     private final DataTreeChangeExtension changeService;
  59     private final RestconfStream.Registry streamRegistry;
  60
  61     @Inject
  62     @Activate
  63     public CreateDataChangeEventSubscriptionRpc(@Reference final RestconfStream.Registry streamRegistry,
  64             @Reference final DatabindProvider databindProvider, @Reference final DOMDataBroker dataBroker) {
  65         super(CreateDataChangeEventSubscription.QNAME);
  66         this.databindProvider = requireNonNull(databindProvider);
  67         changeService = dataBroker.extension(DataTreeChangeExtension.class);
  68         if (changeService == null) {
  69             throw new UnsupportedOperationException("DOMDataBroker does not support the DOMDataTreeChangeService");
  70         }
  71         this.streamRegistry = requireNonNull(streamRegistry);
  72     }
  73
  74     /**
  75      * Create data-change-event stream with POST operation via RPC.
  76      *
  77      * @param input Input of RPC - example in JSON (data-change-event stream):
  78      *              <pre>
  79      *              {@code
  80      *                  {
  81      *                      "input": {
  82      *                          "path": "/toaster:toaster/toaster:toasterStatus",
  83      *                          "sal-remote-augment:datastore": "OPERATIONAL",
  84      *                      }
  85      *                  }
  86      *              }
  87      *              </pre>
  88      * @return Future output of RPC - example in JSON:
  89      *     <pre>
  90      *     {@code
  91      *         {
  92      *             "output": {
  93      *                 "stream-name": "toaster:toaster/toaster:toasterStatus/datastore=OPERATIONAL/scope=ONE"
  94      *             }
  95      *         }
  96      *     }
  97      *     </pre>
  98      */
  99     @Override
 100     public RestconfFuture<OperationsPostResult> invoke(final URI restconfURI, final OperationInput input) {
 101         final var body = input.input();
 102         final var datastoreName = leaf(body, DATASTORE_NODEID, String.class);
 103         final var datastore = datastoreName != null ? LogicalDatastoreType.valueOf(datastoreName)
 104             : LogicalDatastoreType.CONFIGURATION;
 105
 106         final var path = leaf(body, PATH_NODEID, YangInstanceIdentifier.class);
 107         if (path == null) {
 108             return RestconfFuture.failed(
 109                 new RestconfDocumentedException("missing path", ErrorType.APPLICATION, ErrorTag.MISSING_ELEMENT));
 110         }
 111
 112         return streamRegistry.createStream(restconfURI,
 113             new DataTreeChangeSource(databindProvider, changeService, datastore, path),
 114             "Events occuring in " + datastore + " datastore under /"
 115                 + new YangInstanceIdentifierSerializer(input.databind()).serializePath(path))
 116             .transform(stream -> input.newOperationOutput(Builders.containerBuilder()
 117                 .withNodeIdentifier(OUTPUT_NODEID)
 118                 .withChild(ImmutableNodes.leafNode(STREAM_NAME_NODEID, stream.name()))
 119                 .build()));
 120     }
 121 }