2 * Copyright (c) 2016 Cisco Systems, Inc. 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.spi;
10 import static java.util.Objects.requireNonNull;
12 import com.google.common.base.MoreObjects;
13 import com.google.common.base.MoreObjects.ToStringHelper;
14 import com.google.common.collect.ImmutableMap;
15 import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
16 import java.io.UnsupportedEncodingException;
17 import java.lang.invoke.MethodHandles;
18 import java.lang.invoke.VarHandle;
20 import java.time.Instant;
22 import java.util.regex.Pattern;
23 import javax.xml.xpath.XPathExpressionException;
24 import org.checkerframework.checker.lock.qual.GuardedBy;
25 import org.eclipse.jdt.annotation.NonNull;
26 import org.eclipse.jdt.annotation.Nullable;
27 import org.opendaylight.restconf.common.errors.RestconfFuture;
28 import org.opendaylight.restconf.server.api.EventStreamGetParams;
29 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.restconf.monitoring.rev170126.restconf.state.streams.stream.Access;
30 import org.opendaylight.yangtools.concepts.Registration;
31 import org.opendaylight.yangtools.yang.model.api.EffectiveModelContext;
32 import org.slf4j.Logger;
33 import org.slf4j.LoggerFactory;
36 * A RESTCONF notification event stream. Each stream produces a number of events encoded in at least one encoding. The
37 * set of supported encodings is available through {@link #encodings()}.
39 * @param <T> Type of processed events
41 public final class RestconfStream<T> {
43 * An opinionated view on what values we can produce for {@link Access#getEncoding()}. The name can only be composed
44 * of one or more characters matching {@code [a-zA-Z]}.
46 * @param name Encoding name, as visible via the stream's {@code access} list
48 public record EncodingName(@NonNull String name) {
49 private static final Pattern PATTERN = Pattern.compile("[a-zA-Z]+");
52 * Well-known JSON encoding defined by RFC8040's {@code ietf-restconf-monitoring.yang}.
54 public static final @NonNull EncodingName RFC8040_JSON = new EncodingName("json");
56 * Well-known XML encoding defined by RFC8040's {@code ietf-restconf-monitoring.yang}.
58 public static final @NonNull EncodingName RFC8040_XML = new EncodingName("xml");
61 if (!PATTERN.matcher(name).matches()) {
62 throw new IllegalArgumentException("name must match " + PATTERN);
68 * A sink of events for a {@link RestconfStream}.
70 public interface Sink<T> {
72 * Publish a set of events generated from input data.
74 * @param modelContext An {@link EffectiveModelContext} used to format the input
75 * @param input Input data
76 * @param now Current time
77 * @throws NullPointerException if any argument is {@code null}
79 void publish(EffectiveModelContext modelContext, T input, Instant now);
82 * Called when the stream has reached its end.
88 * A source of events for a {@link RestconfStream}.
90 public abstract static class Source<T> {
91 // ImmutableMap because it retains iteration order
92 final @NonNull ImmutableMap<EncodingName, ? extends EventFormatterFactory<T>> encodings;
94 protected Source(final ImmutableMap<EncodingName, ? extends EventFormatterFactory<T>> encodings) {
95 if (encodings.isEmpty()) {
96 throw new IllegalArgumentException("A source must support at least one encoding");
98 this.encodings = encodings;
101 protected abstract @NonNull Registration start(Sink<T> sink);
104 public final String toString() {
105 return addToStringAttributes(MoreObjects.toStringHelper(this)).toString();
108 protected ToStringHelper addToStringAttributes(final ToStringHelper helper) {
109 return helper.add("encodings", encodings.keySet());
114 * Interface for session handler that is responsible for sending of data over established session.
116 public interface Sender {
118 * Interface for sending String message through one of implementation.
120 * @param data Message data to be send.
122 void sendDataMessage(String data);
125 * Called when the stream has reached its end. The handler should close all underlying resources.
131 * An entity managing allocation and lookup of {@link RestconfStream}s.
133 public interface Registry {
135 * Get a {@link RestconfStream} by its name.
137 * @param name Stream name.
138 * @return A {@link RestconfStream}, or {@code null} if the stream with specified name does not exist.
139 * @throws NullPointerException if {@code name} is {@code null}
141 @Nullable RestconfStream<?> lookupStream(String name);
144 * Create a {@link RestconfStream} with a unique name. This method will atomically generate a stream name,
145 * create the corresponding instance and register it.
147 * @param <T> Stream type
148 * @param restconfURI resolved {@code {+restconf}} resource name
149 * @param source Stream instance
150 * @param description Stream descriptiion
151 * @return A future {@link RestconfStream} instance
152 * @throws NullPointerException if any argument is {@code null}
154 <T> @NonNull RestconfFuture<RestconfStream<T>> createStream(URI restconfURI, Source<T> source,
158 private static final Logger LOG = LoggerFactory.getLogger(RestconfStream.class);
159 private static final VarHandle SUBSCRIBERS;
163 SUBSCRIBERS = MethodHandles.lookup().findVarHandle(RestconfStream.class, "subscribers", Subscribers.class);
164 } catch (NoSuchFieldException | IllegalAccessException e) {
165 throw new ExceptionInInitializerError(e);
169 private final @NonNull Sink<T> sink = new Sink<>() {
171 public void publish(final EffectiveModelContext modelContext, final T input, final Instant now) {
172 final var local = acquireSubscribers();
174 local.publish(modelContext, input, now);
176 LOG.debug("Ignoring publish() on terminated stream {}", RestconfStream.this);
181 public void endOfStream() {
182 // Atomic assertion we are ending plus guarded cleanup
183 final var local = (Subscribers<T>) SUBSCRIBERS.getAndSetRelease(RestconfStream.this, null);
190 private final @NonNull AbstractRestconfStreamRegistry registry;
191 private final @NonNull Source<T> source;
192 private final @NonNull String name;
194 // Accessed via SUBSCRIBERS, 'null' indicates we have been shut down
195 @SuppressWarnings("unused")
196 @SuppressFBWarnings(value = "URF_UNREAD_FIELD", justification = "https://github.com/spotbugs/spotbugs/issues/2749")
197 private volatile Subscribers<T> subscribers = Subscribers.empty();
200 private Registration registration;
202 RestconfStream(final AbstractRestconfStreamRegistry registry, final Source<T> source, final String name) {
203 this.registry = requireNonNull(registry);
204 this.source = requireNonNull(source);
205 this.name = requireNonNull(name);
209 * Get name of stream.
211 * @return Stream name.
213 public @NonNull String name() {
218 * Get supported {@link EncodingName}s. The set is guaranteed to contain at least one element and does not contain
221 * @return Supported encodings.
223 @SuppressWarnings("null")
224 @NonNull Set<EncodingName> encodings() {
225 return source.encodings.keySet();
229 * Registers {@link Sender} subscriber.
231 * @param handler SSE or WS session handler.
232 * @param encoding Requested event stream encoding
233 * @param params Reception parameters
234 * @return A new {@link Registration}, or {@code null} if the subscriber cannot be added
235 * @throws NullPointerException if any argument is {@code null}
236 * @throws UnsupportedEncodingException if {@code encoding} is not supported
237 * @throws XPathExpressionException if requested filter is not valid
239 public @Nullable Registration addSubscriber(final Sender handler, final EncodingName encoding,
240 final EventStreamGetParams params) throws UnsupportedEncodingException, XPathExpressionException {
241 final var factory = source.encodings.get(requireNonNull(encoding));
242 if (factory == null) {
243 throw new UnsupportedEncodingException("Stream '" + name + "' does not support " + encoding);
246 final var startTime = params.startTime();
247 if (startTime != null) {
248 throw new IllegalArgumentException("Stream " + name + " does not support replay");
251 final var leafNodes = params.leafNodesOnly() != null && params.leafNodesOnly().value();
252 final var skipData = params.skipNotificationData() != null && params.skipNotificationData().value();
253 final var changedLeafNodes = params.changedLeafNodesOnly() != null && params.changedLeafNodesOnly().value();
254 final var childNodes = params.childNodesOnly() != null && params.childNodesOnly().value();
256 final var textParams = new TextParameters(leafNodes, skipData, changedLeafNodes, childNodes);
258 final var filter = params.filter();
259 final var filterValue = filter == null ? null : filter.paramValue();
260 final var formatter = filterValue == null || filterValue.isEmpty() ? factory.getFormatter(textParams)
261 : factory.getFormatter(textParams, filterValue);
264 // Lockless add of a subscriber. If we observe a null this stream is dead before the new subscriber could be
266 final var toAdd = new Subscriber<>(this, handler, formatter);
267 var observed = acquireSubscribers();
268 while (observed != null) {
269 final var next = observed.add(toAdd);
270 final var witness = (Subscribers<T>) SUBSCRIBERS.compareAndExchangeRelease(this, observed, next);
271 if (witness == observed) {
272 LOG.debug("Subscriber {} is added.", handler);
273 if (observed instanceof Subscribers.Empty) {
274 // We have became non-empty, start the source
280 // We have raced: retry the operation
288 * Removes a {@link Subscriber}. If this was the last subscriber also shut down this stream and initiate its removal
291 * @param subscriber The {@link Subscriber} to remove
292 * @throws NullPointerException if {@code subscriber} is {@code null}
294 void removeSubscriber(final Subscriber<T> subscriber) {
295 final var toRemove = requireNonNull(subscriber);
296 var observed = acquireSubscribers();
297 while (observed != null) {
298 final var next = observed.remove(toRemove);
299 final var witness = (Subscribers<T>) SUBSCRIBERS.compareAndExchangeRelease(this, observed, next);
300 if (witness == observed) {
301 LOG.debug("Subscriber {} is removed", subscriber);
303 // We have lost the last subscriber, terminate.
309 // We have raced: retry the operation
314 private Subscribers<T> acquireSubscribers() {
315 return (Subscribers<T>) SUBSCRIBERS.getAcquire(this);
318 private void startSource() {
319 // We have not started the stream yet, make sure that happens. This is a bit more involved, as the source may
320 // immediately issue endOfStream(), which in turn invokes terminate(). But at that point start() has not return
321 // and therefore registration is still null -- and thus we need to see if we are still on-line.
322 final var reg = source.start(sink);
323 synchronized (this) {
324 if (acquireSubscribers() == null) {
332 private void terminate() {
333 synchronized (this) {
334 if (registration != null) {
335 registration.close();
339 registry.removeStream(this);
343 public String toString() {
344 return MoreObjects.toStringHelper(this).add("name", name).add("source", source).toString();