restconf/restconf-nb/src/main/java/org/opendaylight/restconf/nb/rfc8040/streams/ListenersBroker.java

   1 /*
   2  * Copyright © 2019 FRINX s.r.o. 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;
   9
  10 import static com.google.common.base.Strings.isNullOrEmpty;
  11 import static java.util.Objects.requireNonNull;
  12
  13 import com.google.common.collect.BiMap;
  14 import com.google.common.collect.HashBiMap;
  15 import com.google.common.collect.ImmutableSet;
  16 import java.net.URI;
  17 import java.util.Optional;
  18 import java.util.concurrent.ExecutionException;
  19 import java.util.concurrent.locks.StampedLock;
  20 import javax.ws.rs.core.UriInfo;
  21 import org.eclipse.jdt.annotation.NonNull;
  22 import org.eclipse.jdt.annotation.Nullable;
  23 import org.opendaylight.mdsal.common.api.LogicalDatastoreType;
  24 import org.opendaylight.mdsal.dom.api.DOMDataBroker;
  25 import org.opendaylight.mdsal.dom.api.DOMDataTreeWriteOperations;
  26 import org.opendaylight.mdsal.dom.api.DOMDataTreeWriteTransaction;
  27 import org.opendaylight.mdsal.dom.api.DOMMountPointService;
  28 import org.opendaylight.restconf.common.errors.RestconfDocumentedException;
  29 import org.opendaylight.restconf.nb.rfc8040.NotificationQueryParams;
  30 import org.opendaylight.restconf.nb.rfc8040.URLConstants;
  31 import org.opendaylight.restconf.nb.rfc8040.monitoring.RestconfStateStreams;
  32 import org.opendaylight.restconf.nb.rfc8040.rests.services.impl.RestconfStreamsSubscriptionServiceImpl.HandlersHolder;
  33 import org.opendaylight.restconf.nb.rfc8040.rests.utils.RestconfStreamsConstants;
  34 import org.opendaylight.restconf.nb.rfc8040.utils.parser.IdentifierCodec;
  35 import org.opendaylight.yang.gen.v1.urn.sal.restconf.event.subscription.rev140708.CreateDataChangeEventSubscriptionInput1.Scope;
  36 import org.opendaylight.yang.gen.v1.urn.sal.restconf.event.subscription.rev140708.NotificationOutputTypeGrouping.NotificationOutputType;
  37 import org.opendaylight.yangtools.yang.common.ErrorTag;
  38 import org.opendaylight.yangtools.yang.common.ErrorType;
  39 import org.opendaylight.yangtools.yang.common.QName;
  40 import org.opendaylight.yangtools.yang.data.api.YangInstanceIdentifier;
  41 import org.opendaylight.yangtools.yang.data.api.schema.MapEntryNode;
  42 import org.opendaylight.yangtools.yang.model.api.EffectiveModelContext;
  43 import org.opendaylight.yangtools.yang.model.api.NotificationDefinition;
  44 import org.opendaylight.yangtools.yang.model.api.stmt.NotificationEffectiveStatement;
  45 import org.slf4j.Logger;
  46 import org.slf4j.LoggerFactory;
  47
  48 /**
  49  * This singleton class is responsible for creation, removal and searching for {@link ListenerAdapter} or
  50  * {@link NotificationListenerAdapter} listeners.
  51  */
  52 // FIXME: furthermore, this should be tied to ietf-restconf-monitoring, as the Strings used in its maps are stream
  53 //        names. We essentially need a component which deals with allocation of stream names and their lifecycle and
  54 //        the contents of /restconf-state/streams.
  55 public abstract sealed class ListenersBroker {
  56     /**
  57      * A ListenersBroker working with Server-Sent Events.
  58      */
  59     public static final class ServerSentEvents extends ListenersBroker {
  60         @Override
  61         public URI prepareUriByStreamName(final UriInfo uriInfo, final String streamName) {
  62             return uriInfo.getBaseUriBuilder()
  63                 .replacePath(URLConstants.BASE_PATH + '/' + URLConstants.SSE_SUBPATH + '/' + streamName)
  64                 .build();
  65         }
  66     }
  67
  68     /**
  69      * A ListenersBroker working with WebSockets.
  70      */
  71     public static final class WebSockets extends ListenersBroker {
  72         @Override
  73         public URI prepareUriByStreamName(final UriInfo uriInfo, final String streamName) {
  74             final var scheme = switch (uriInfo.getAbsolutePath().getScheme()) {
  75                 // Secured HTTP goes to Secured WebSockets
  76                 case "https" -> "wss";
  77                 // Unsecured HTTP and others go to unsecured WebSockets
  78                 default -> "ws";
  79             };
  80
  81             return uriInfo.getBaseUriBuilder()
  82                 .scheme(scheme)
  83                 .replacePath(URLConstants.BASE_PATH + '/' + streamName)
  84                 .build();
  85         }
  86     }
  87
  88     private static final Logger LOG = LoggerFactory.getLogger(ListenersBroker.class);
  89
  90     private final StampedLock dataChangeListenersLock = new StampedLock();
  91     private final StampedLock notificationListenersLock = new StampedLock();
  92     private final StampedLock deviceNotificationListenersLock = new StampedLock();
  93     private final BiMap<String, ListenerAdapter> dataChangeListeners = HashBiMap.create();
  94     private final BiMap<String, NotificationListenerAdapter> notificationListeners = HashBiMap.create();
  95     private final BiMap<String, DeviceNotificationListenerAdaptor> deviceNotificationListeners = HashBiMap.create();
  96
  97     private ListenersBroker() {
  98         // Hidden on purpose
  99     }
 100
 101     /**
 102      * Gets {@link ListenerAdapter} specified by stream identification.
 103      *
 104      * @param streamName Stream name.
 105      * @return {@link ListenerAdapter} specified by stream name or {@code null} if listener with specified stream name
 106      *         does not exist.
 107      * @throws NullPointerException in {@code streamName} is {@code null}
 108      */
 109     public final @Nullable ListenerAdapter dataChangeListenerFor(final String streamName) {
 110         requireNonNull(streamName);
 111
 112         final long stamp = dataChangeListenersLock.readLock();
 113         try {
 114             return dataChangeListeners.get(streamName);
 115         } finally {
 116             dataChangeListenersLock.unlockRead(stamp);
 117         }
 118     }
 119
 120     /**
 121      * Gets {@link NotificationListenerAdapter} specified by stream name.
 122      *
 123      * @param streamName Stream name.
 124      * @return {@link NotificationListenerAdapter} specified by stream name or {@code null} if listener with specified
 125      *         stream name does not exist.
 126      * @throws NullPointerException in {@code streamName} is {@code null}
 127      */
 128     public final @Nullable NotificationListenerAdapter notificationListenerFor(final String streamName) {
 129         requireNonNull(streamName);
 130
 131         final long stamp = notificationListenersLock.readLock();
 132         try {
 133             return notificationListeners.get(streamName);
 134         } finally {
 135             notificationListenersLock.unlockRead(stamp);
 136         }
 137     }
 138
 139     /**
 140      * Get listener for device path.
 141      *
 142      * @param streamName name.
 143      * @return {@link DeviceNotificationListenerAdaptor} specified by stream name or {@code null} if listener with
 144      *         specified stream name does not exist.
 145      * @throws NullPointerException in {@code path} is {@code null}
 146      */
 147     public final @Nullable DeviceNotificationListenerAdaptor deviceNotificationListenerFor(final String streamName) {
 148         requireNonNull(streamName);
 149
 150         final long stamp = deviceNotificationListenersLock.readLock();
 151         try {
 152             return deviceNotificationListeners.get(streamName);
 153         } finally {
 154             deviceNotificationListenersLock.unlockRead(stamp);
 155         }
 156     }
 157
 158     /**
 159      * Get listener for stream-name.
 160      *
 161      * @param streamName Stream name.
 162      * @return {@link NotificationListenerAdapter} or {@link ListenerAdapter} object wrapped in {@link Optional}
 163      *     or {@link Optional#empty()} if listener with specified stream name doesn't exist.
 164      */
 165     public final @Nullable BaseListenerInterface listenerFor(final String streamName) {
 166         if (streamName.startsWith(RestconfStreamsConstants.NOTIFICATION_STREAM)) {
 167             return notificationListenerFor(streamName);
 168         } else if (streamName.startsWith(RestconfStreamsConstants.DATA_SUBSCRIPTION)) {
 169             return dataChangeListenerFor(streamName);
 170         } else if (streamName.startsWith(RestconfStreamsConstants.DEVICE_NOTIFICATION_STREAM)) {
 171             return deviceNotificationListenerFor(streamName);
 172         } else {
 173             return null;
 174         }
 175     }
 176
 177     /**
 178      * Creates new {@link ListenerAdapter} listener using input stream name and path if such listener
 179      * hasn't been created yet.
 180      *
 181      * @param path       Path to data in data repository.
 182      * @param outputType Specific type of output for notifications - XML or JSON.
 183      * @return Created or existing data-change listener adapter.
 184      */
 185     public final ListenerAdapter registerDataChangeListener(final EffectiveModelContext modelContext,
 186             final LogicalDatastoreType datastore, final YangInstanceIdentifier path, final Scope scope,
 187             final NotificationOutputType outputType) {
 188         final var sb = new StringBuilder(RestconfStreamsConstants.DATA_SUBSCRIPTION)
 189             .append('/').append(createStreamNameFromUri(IdentifierCodec.serialize(path, modelContext)))
 190             .append('/').append(RestconfStreamsConstants.DATASTORE_PARAM_NAME).append('=').append(datastore)
 191             .append('/').append(RestconfStreamsConstants.SCOPE_PARAM_NAME).append('=').append(scope);
 192         if (outputType != NotificationOutputType.XML) {
 193             sb.append('/').append(outputType.getName());
 194         }
 195
 196         final long stamp = dataChangeListenersLock.writeLock();
 197         try {
 198             return dataChangeListeners.computeIfAbsent(sb.toString(),
 199                 streamName -> new ListenerAdapter(datastore, path, streamName, outputType, this));
 200         } finally {
 201             dataChangeListenersLock.unlockWrite(stamp);
 202         }
 203     }
 204
 205     /**
 206      * Creates new {@link NotificationDefinition} listener using input stream name and schema path
 207      * if such listener haven't been created yet.
 208      *
 209      * @param refSchemaCtx reference {@link EffectiveModelContext}
 210      * @param notifications {@link QName}s of accepted YANG notifications
 211      * @param outputType Specific type of output for notifications - XML or JSON.
 212      * @return Created or existing notification listener adapter.
 213      */
 214     public final NotificationListenerAdapter registerNotificationListener(final EffectiveModelContext refSchemaCtx,
 215             final ImmutableSet<QName> notifications, final NotificationOutputType outputType) {
 216         final var sb = new StringBuilder(RestconfStreamsConstants.NOTIFICATION_STREAM).append('/');
 217         var haveFirst = false;
 218         for (var qname : notifications) {
 219             final var module = refSchemaCtx.findModuleStatement(qname.getModule())
 220                 .orElseThrow(() -> new RestconfDocumentedException(qname + " refers to an unknown module",
 221                     ErrorType.APPLICATION, ErrorTag.INVALID_VALUE));
 222             final var stmt = module.findSchemaTreeNode(qname)
 223                 .orElseThrow(() -> new RestconfDocumentedException(qname + " refers to an notification",
 224                     ErrorType.APPLICATION, ErrorTag.INVALID_VALUE));
 225             if (!(stmt instanceof NotificationEffectiveStatement)) {
 226                 throw new RestconfDocumentedException(qname + " refers to a non-notification",
 227                     ErrorType.APPLICATION, ErrorTag.INVALID_VALUE);
 228             }
 229
 230             if (haveFirst) {
 231                 sb.append(',');
 232             } else {
 233                 haveFirst = true;
 234             }
 235             sb.append(module.argument().getLocalName()).append(':').append(qname.getLocalName());
 236         }
 237         if (outputType != NotificationOutputType.XML) {
 238             sb.append('/').append(outputType.getName());
 239         }
 240
 241         final long stamp = notificationListenersLock.writeLock();
 242         try {
 243             return notificationListeners.computeIfAbsent(sb.toString(),
 244                 streamName -> new NotificationListenerAdapter(notifications, streamName, outputType, this));
 245         } finally {
 246             notificationListenersLock.unlockWrite(stamp);
 247         }
 248     }
 249
 250     /**
 251      * Creates new {@link DeviceNotificationListenerAdaptor} listener using input stream name and schema path
 252      * if such listener haven't been created yet.
 253      *
 254      * @param deviceName Device name.
 255      * @param outputType Specific type of output for notifications - XML or JSON.
 256      * @param refSchemaCtx Schema context of node
 257      * @param mountPointService Mount point service
 258      * @return Created or existing device notification listener adapter.
 259      */
 260     public final DeviceNotificationListenerAdaptor registerDeviceNotificationListener(final String deviceName,
 261             final NotificationOutputType outputType, final EffectiveModelContext refSchemaCtx,
 262             final DOMMountPointService mountPointService, final YangInstanceIdentifier path) {
 263         final var sb = new StringBuilder(RestconfStreamsConstants.DEVICE_NOTIFICATION_STREAM).append('/')
 264             .append(deviceName);
 265
 266         final long stamp = deviceNotificationListenersLock.writeLock();
 267         try {
 268             return deviceNotificationListeners.computeIfAbsent(sb.toString(),
 269                 streamName -> new DeviceNotificationListenerAdaptor(streamName, outputType, refSchemaCtx,
 270                     mountPointService, path, this));
 271         } finally {
 272             deviceNotificationListenersLock.unlockWrite(stamp);
 273         }
 274     }
 275
 276     /**
 277      * Removal and closing of all data-change-event and notification listeners.
 278      */
 279     public final synchronized void removeAndCloseAllListeners() {
 280         final long stampNotifications = notificationListenersLock.writeLock();
 281         final long stampDataChanges = dataChangeListenersLock.writeLock();
 282         try {
 283             removeAndCloseAllDataChangeListenersTemplate();
 284             removeAndCloseAllNotificationListenersTemplate();
 285         } finally {
 286             dataChangeListenersLock.unlockWrite(stampDataChanges);
 287             notificationListenersLock.unlockWrite(stampNotifications);
 288         }
 289     }
 290
 291     /**
 292      * Closes and removes all data-change listeners.
 293      */
 294     public final void removeAndCloseAllDataChangeListeners() {
 295         final long stamp = dataChangeListenersLock.writeLock();
 296         try {
 297             removeAndCloseAllDataChangeListenersTemplate();
 298         } finally {
 299             dataChangeListenersLock.unlockWrite(stamp);
 300         }
 301     }
 302
 303     @SuppressWarnings("checkstyle:IllegalCatch")
 304     private void removeAndCloseAllDataChangeListenersTemplate() {
 305         dataChangeListeners.values().forEach(listenerAdapter -> {
 306             try {
 307                 listenerAdapter.close();
 308             } catch (Exception e) {
 309                 LOG.error("Failed to close data-change listener {}.", listenerAdapter, e);
 310                 throw new IllegalStateException("Failed to close data-change listener %s.".formatted(listenerAdapter),
 311                     e);
 312             }
 313         });
 314         dataChangeListeners.clear();
 315     }
 316
 317     /**
 318      * Closes and removes all notification listeners.
 319      */
 320     public final void removeAndCloseAllNotificationListeners() {
 321         final long stamp = notificationListenersLock.writeLock();
 322         try {
 323             removeAndCloseAllNotificationListenersTemplate();
 324         } finally {
 325             notificationListenersLock.unlockWrite(stamp);
 326         }
 327     }
 328
 329     @SuppressWarnings("checkstyle:IllegalCatch")
 330     private void removeAndCloseAllNotificationListenersTemplate() {
 331         notificationListeners.values().forEach(listenerAdapter -> {
 332             try {
 333                 listenerAdapter.close();
 334             } catch (Exception e) {
 335                 LOG.error("Failed to close notification listener {}.", listenerAdapter, e);
 336                 throw new IllegalStateException("Failed to close notification listener %s.".formatted(listenerAdapter),
 337                     e);
 338             }
 339         });
 340         notificationListeners.clear();
 341     }
 342
 343     /**
 344      * Removes and closes data-change listener of type {@link ListenerAdapter} specified in parameter.
 345      *
 346      * @param listener Listener to be closed and removed.
 347      */
 348     @SuppressWarnings("checkstyle:IllegalCatch")
 349     public final void removeAndCloseDataChangeListener(final ListenerAdapter listener) {
 350         final long stamp = dataChangeListenersLock.writeLock();
 351         try {
 352             removeAndCloseDataChangeListenerTemplate(listener);
 353         } catch (Exception exception) {
 354             LOG.error("Data-change listener {} cannot be closed.", listener, exception);
 355         } finally {
 356             dataChangeListenersLock.unlockWrite(stamp);
 357         }
 358     }
 359
 360     /**
 361      * Removes and closes data-change listener of type {@link ListenerAdapter} specified in parameter.
 362      *
 363      * @param listener Listener to be closed and removed.
 364      */
 365     private void removeAndCloseDataChangeListenerTemplate(final ListenerAdapter listener) {
 366         try {
 367             requireNonNull(listener).close();
 368             if (dataChangeListeners.inverse().remove(listener) == null) {
 369                 LOG.warn("There isn't any data-change event stream that would match listener adapter {}.", listener);
 370             }
 371         } catch (InterruptedException | ExecutionException e) {
 372             LOG.error("Data-change listener {} cannot be closed.", listener, e);
 373             throw new IllegalStateException("Data-change listener %s cannot be closed.".formatted(listener), e);
 374         }
 375     }
 376
 377     /**
 378      * Removes and closes notification listener of type {@link NotificationListenerAdapter} specified in parameter.
 379      *
 380      * @param listener Listener to be closed and removed.
 381      */
 382     @SuppressWarnings("checkstyle:IllegalCatch")
 383     public final void removeAndCloseNotificationListener(final NotificationListenerAdapter listener) {
 384         final long stamp = notificationListenersLock.writeLock();
 385         try {
 386             removeAndCloseNotificationListenerTemplate(listener);
 387         } catch (Exception e) {
 388             LOG.error("Notification listener {} cannot be closed.", listener, e);
 389         } finally {
 390             notificationListenersLock.unlockWrite(stamp);
 391         }
 392     }
 393
 394     /**
 395      * Removes and closes device notification listener of type {@link NotificationListenerAdapter}
 396      * specified in parameter.
 397      *
 398      * @param listener Listener to be closed and removed.
 399      */
 400     @SuppressWarnings("checkstyle:IllegalCatch")
 401     public final void removeAndCloseDeviceNotificationListener(final DeviceNotificationListenerAdaptor listener) {
 402         final long stamp = deviceNotificationListenersLock.writeLock();
 403         try {
 404             requireNonNull(listener);
 405             if (deviceNotificationListeners.inverse().remove(listener) == null) {
 406                 LOG.warn("There isn't any device notification stream that would match listener adapter {}.", listener);
 407             }
 408         } catch (final Exception exception) {
 409             LOG.error("Device Notification listener {} cannot be closed.", listener, exception);
 410         } finally {
 411             deviceNotificationListenersLock.unlockWrite(stamp);
 412         }
 413     }
 414
 415     private void removeAndCloseNotificationListenerTemplate(final NotificationListenerAdapter listener) {
 416         try {
 417             requireNonNull(listener).close();
 418             if (notificationListeners.inverse().remove(listener) == null) {
 419                 LOG.warn("There isn't any notification stream that would match listener adapter {}.", listener);
 420             }
 421         } catch (InterruptedException | ExecutionException e) {
 422             LOG.error("Notification listener {} cannot be closed.", listener, e);
 423             throw new IllegalStateException("Notification listener %s cannot be closed.".formatted(listener), e);
 424         }
 425     }
 426
 427     /**
 428      * Removal and closing of general listener (data-change or notification listener).
 429      *
 430      * @param listener Listener to be closed and removed from cache.
 431      */
 432     final void removeAndCloseListener(final BaseListenerInterface listener) {
 433         requireNonNull(listener);
 434         if (listener instanceof ListenerAdapter) {
 435             removeAndCloseDataChangeListener((ListenerAdapter) listener);
 436         } else if (listener instanceof NotificationListenerAdapter) {
 437             removeAndCloseNotificationListener((NotificationListenerAdapter) listener);
 438         }
 439     }
 440
 441     /**
 442      * Creates string representation of stream name from URI. Removes slash from URI in start and end positions,
 443      * and optionally {@link URLConstants#BASE_PATH} prefix.
 444      *
 445      * @param uri URI for creation of stream name.
 446      * @return String representation of stream name.
 447      */
 448     public static String createStreamNameFromUri(final String uri) {
 449         String result = requireNonNull(uri);
 450         while (true) {
 451             if (result.startsWith(URLConstants.BASE_PATH)) {
 452                 result = result.substring(URLConstants.BASE_PATH.length());
 453             } else if (result.startsWith("/")) {
 454                 result = result.substring(1);
 455             } else {
 456                 break;
 457             }
 458         }
 459         if (result.endsWith("/")) {
 460             result = result.substring(0, result.length() - 1);
 461         }
 462         return result;
 463     }
 464
 465     /**
 466      * Prepare URL from base name and stream name.
 467      *
 468      * @param uriInfo base URL information
 469      * @param streamName name of stream for create
 470      * @return final URL
 471      */
 472     public abstract @NonNull URI prepareUriByStreamName(UriInfo uriInfo, String streamName);
 473
 474     /**
 475      * Register listener by streamName in identifier to listen to yang notifications, and put or delete information
 476      * about listener to DS according to ietf-restconf-monitoring.
 477      *
 478      * @param identifier              Name of the stream.
 479      * @param uriInfo                 URI information.
 480      * @param notificationQueryParams Query parameters of notification.
 481      * @param handlersHolder          Holder of handlers for notifications.
 482      * @return Stream location for listening.
 483      */
 484     public final @NonNull URI subscribeToYangStream(final String identifier, final UriInfo uriInfo,
 485             final NotificationQueryParams notificationQueryParams, final HandlersHolder handlersHolder) {
 486         final String streamName = createStreamNameFromUri(identifier);
 487         if (isNullOrEmpty(streamName)) {
 488             throw new RestconfDocumentedException("Stream name is empty.", ErrorType.PROTOCOL, ErrorTag.INVALID_VALUE);
 489         }
 490
 491         final var notificationListenerAdapter = notificationListenerFor(streamName);
 492         if (notificationListenerAdapter == null) {
 493             throw new RestconfDocumentedException("Stream with name %s was not found.".formatted(streamName),
 494                 ErrorType.PROTOCOL, ErrorTag.UNKNOWN_ELEMENT);
 495         }
 496
 497         final URI uri = prepareUriByStreamName(uriInfo, streamName);
 498         notificationListenerAdapter.setQueryParams(notificationQueryParams);
 499         notificationListenerAdapter.listen(handlersHolder.notificationService());
 500         final DOMDataBroker dataBroker = handlersHolder.dataBroker();
 501         notificationListenerAdapter.setCloseVars(dataBroker, handlersHolder.databindProvider());
 502         final MapEntryNode mapToStreams = RestconfStateStreams.notificationStreamEntry(streamName,
 503             notificationListenerAdapter.qnames(), notificationListenerAdapter.getStart(),
 504             notificationListenerAdapter.getOutputType(), uri);
 505
 506         // FIXME: how does this correlate with the transaction notificationListenerAdapter.close() will do?
 507         final DOMDataTreeWriteTransaction writeTransaction = dataBroker.newWriteOnlyTransaction();
 508         writeDataToDS(writeTransaction, mapToStreams);
 509         submitData(writeTransaction);
 510         return uri;
 511     }
 512
 513     /**
 514      * Register listener by streamName in identifier to listen to data change notifications, and put or delete
 515      * information about listener to DS according to ietf-restconf-monitoring.
 516      *
 517      * @param identifier              Identifier as stream name.
 518      * @param uriInfo                 Base URI information.
 519      * @param notificationQueryParams Query parameters of notification.
 520      * @param handlersHolder          Holder of handlers for notifications.
 521      * @return Location for listening.
 522      */
 523     public final URI subscribeToDataStream(final String identifier, final UriInfo uriInfo,
 524             final NotificationQueryParams notificationQueryParams, final HandlersHolder handlersHolder) {
 525         final var streamName = createStreamNameFromUri(identifier);
 526         final var listener = dataChangeListenerFor(streamName);
 527         if (listener == null) {
 528             throw new RestconfDocumentedException("No listener found for stream " + streamName,
 529                 ErrorType.APPLICATION, ErrorTag.DATA_MISSING);
 530         }
 531
 532         listener.setQueryParams(notificationQueryParams);
 533
 534         final var dataBroker = handlersHolder.dataBroker();
 535         final var schemaHandler = handlersHolder.databindProvider();
 536         listener.setCloseVars(dataBroker, schemaHandler);
 537         listener.listen(dataBroker);
 538
 539         final var uri = prepareUriByStreamName(uriInfo, streamName);
 540         final var schemaContext = schemaHandler.currentContext().modelContext();
 541         final var serializedPath = IdentifierCodec.serialize(listener.getPath(), schemaContext);
 542
 543         final var mapToStreams = RestconfStateStreams.dataChangeStreamEntry(listener.getPath(),
 544                 listener.getStart(), listener.getOutputType(), uri, schemaContext, serializedPath);
 545         final var writeTransaction = dataBroker.newWriteOnlyTransaction();
 546         writeDataToDS(writeTransaction, mapToStreams);
 547         submitData(writeTransaction);
 548         return uri;
 549     }
 550
 551     // FIXME: callers are utter duplicates, refactor them
 552     private static void writeDataToDS(final DOMDataTreeWriteOperations tx, final MapEntryNode mapToStreams) {
 553         // FIXME: use put() here
 554         tx.merge(LogicalDatastoreType.OPERATIONAL, RestconfStateStreams.restconfStateStreamPath(mapToStreams.name()),
 555             mapToStreams);
 556     }
 557
 558     private static void submitData(final DOMDataTreeWriteTransaction readWriteTransaction) {
 559         try {
 560             readWriteTransaction.commit().get();
 561         } catch (final InterruptedException | ExecutionException e) {
 562             throw new RestconfDocumentedException("Problem while putting data to DS.", e);
 563         }
 564     }
 565 }