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 java.io.UnsupportedEncodingException;
16 import java.lang.invoke.MethodHandles;
17 import java.lang.invoke.VarHandle;
19 import java.time.Instant;
21 import java.util.regex.Pattern;
22 import javax.xml.xpath.XPathExpressionException;
23 import org.checkerframework.checker.lock.qual.GuardedBy;
24 import org.eclipse.jdt.annotation.NonNull;
25 import org.eclipse.jdt.annotation.Nullable;
26 import org.opendaylight.restconf.common.errors.RestconfFuture;
27 import org.opendaylight.restconf.nb.rfc8040.ReceiveEventsParams;
28 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.restconf.monitoring.rev170126.restconf.state.streams.stream.Access;
29 import org.opendaylight.yangtools.concepts.Registration;
30 import org.opendaylight.yangtools.yang.model.api.EffectiveModelContext;
31 import org.slf4j.Logger;
32 import org.slf4j.LoggerFactory;
35 * A RESTCONF notification event stream. Each stream produces a number of events encoded in at least one encoding. The
36 * set of supported encodings is available through {@link #encodings()}.
38 * @param <T> Type of processed events
40 public final class RestconfStream<T> {
42 * An opinionated view on what values we can produce for {@link Access#getEncoding()}. The name can only be composed
43 * of one or more characters matching {@code [a-zA-Z]}.
45 * @param name Encoding name, as visible via the stream's {@code access} list
47 public record EncodingName(@NonNull String name) {
48 private static final Pattern PATTERN = Pattern.compile("[a-zA-Z]+");
51 * Well-known JSON encoding defined by RFC8040's {@code ietf-restconf-monitoring.yang}.
53 public static final @NonNull EncodingName RFC8040_JSON = new EncodingName("json");
55 * Well-known XML encoding defined by RFC8040's {@code ietf-restconf-monitoring.yang}.
57 public static final @NonNull EncodingName RFC8040_XML = new EncodingName("xml");
60 if (!PATTERN.matcher(name).matches()) {
61 throw new IllegalArgumentException("name must match " + PATTERN);
67 * A sink of events for a {@link RestconfStream}.
69 public interface Sink<T> {
71 * Publish a set of events generated from input data.
73 * @param modelContext An {@link EffectiveModelContext} used to format the input
74 * @param input Input data
75 * @param now Current time
76 * @throws NullPointerException if any argument is {@code null}
78 void publish(EffectiveModelContext modelContext, T input, Instant now);
81 * Called when the stream has reached its end.
87 * A source of events for a {@link RestconfStream}.
89 public abstract static class Source<T> {
90 // ImmutableMap because it retains iteration order
91 final @NonNull ImmutableMap<EncodingName, ? extends EventFormatterFactory<T>> encodings;
93 protected Source(final ImmutableMap<EncodingName, ? extends EventFormatterFactory<T>> encodings) {
94 if (encodings.isEmpty()) {
95 throw new IllegalArgumentException("A source must support at least one encoding");
97 this.encodings = encodings;
100 protected abstract @NonNull Registration start(Sink<T> sink);
103 public final String toString() {
104 return addToStringAttributes(MoreObjects.toStringHelper(this)).toString();
107 protected ToStringHelper addToStringAttributes(final ToStringHelper helper) {
108 return helper.add("encodings", encodings.keySet());
113 * Interface for session handler that is responsible for sending of data over established session.
115 public interface Sender {
117 * Interface for sending String message through one of implementation.
119 * @param data Message data to be send.
121 void sendDataMessage(String data);
124 * Called when the stream has reached its end. The handler should close all underlying resources.
130 * An entity managing allocation and lookup of {@link RestconfStream}s.
132 public interface Registry {
134 * Get a {@link RestconfStream} by its name.
136 * @param name Stream name.
137 * @return A {@link RestconfStream}, or {@code null} if the stream with specified name does not exist.
138 * @throws NullPointerException if {@code name} is {@code null}
140 @Nullable RestconfStream<?> lookupStream(String name);
143 * Create a {@link RestconfStream} with a unique name. This method will atomically generate a stream name,
144 * create the corresponding instance and register it.
146 * @param <T> Stream type
147 * @param restconfURI resolved {@code {+restconf}} resource name
148 * @param source Stream instance
149 * @param description Stream descriptiion
150 * @return A future {@link RestconfStream} instance
151 * @throws NullPointerException if any argument is {@code null}
153 <T> @NonNull RestconfFuture<RestconfStream<T>> createStream(URI restconfURI, Source<T> source,
157 private static final Logger LOG = LoggerFactory.getLogger(RestconfStream.class);
158 private static final VarHandle SUBSCRIBERS;
162 SUBSCRIBERS = MethodHandles.lookup().findVarHandle(RestconfStream.class, "subscribers", Subscribers.class);
163 } catch (NoSuchFieldException | IllegalAccessException e) {
164 throw new ExceptionInInitializerError(e);
168 private final @NonNull Sink<T> sink = new Sink<>() {
170 public void publish(final EffectiveModelContext modelContext, final T input, final Instant now) {
171 final var local = acquireSubscribers();
173 local.publish(modelContext, input, now);
175 LOG.debug("Ignoring publish() on terminated stream {}", RestconfStream.this);
180 public void endOfStream() {
181 // Atomic assertion we are ending plus guarded cleanup
182 final var local = (Subscribers<T>) SUBSCRIBERS.getAndSetRelease(RestconfStream.this, null);
189 private final @NonNull AbstractRestconfStreamRegistry registry;
190 private final @NonNull Source<T> source;
191 private final @NonNull String name;
193 // Accessed via SUBSCRIBERS, 'null' indicates we have been shut down
194 @SuppressWarnings("unused")
195 private volatile Subscribers<T> subscribers = Subscribers.empty();
198 private Registration registration;
200 RestconfStream(final AbstractRestconfStreamRegistry registry, final Source<T> source, final String name) {
201 this.registry = requireNonNull(registry);
202 this.source = requireNonNull(source);
203 this.name = requireNonNull(name);
207 * Get name of stream.
209 * @return Stream name.
211 public @NonNull String name() {
216 * Get supported {@link EncodingName}s. The set is guaranteed to contain at least one element and does not contain
219 * @return Supported encodings.
221 @SuppressWarnings("null")
222 @NonNull Set<EncodingName> encodings() {
223 return source.encodings.keySet();
227 * Registers {@link Sender} subscriber.
229 * @param handler SSE or WS session handler.
230 * @param encoding Requested event stream encoding
231 * @param params Reception parameters
232 * @return A new {@link Registration}, or {@code null} if the subscriber cannot be added
233 * @throws NullPointerException if any argument is {@code null}
234 * @throws UnsupportedEncodingException if {@code encoding} is not supported
235 * @throws XPathExpressionException if requested filter is not valid
237 public @Nullable Registration addSubscriber(final Sender handler, final EncodingName encoding,
238 final ReceiveEventsParams params) throws UnsupportedEncodingException, XPathExpressionException {
239 final var factory = source.encodings.get(requireNonNull(encoding));
240 if (factory == null) {
241 throw new UnsupportedEncodingException("Stream '" + name + "' does not support " + encoding);
244 final var startTime = params.startTime();
245 if (startTime != null) {
246 throw new IllegalArgumentException("Stream " + name + " does not support replay");
249 final var leafNodes = params.leafNodesOnly();
250 final var skipData = params.skipNotificationData();
251 final var changedLeafNodes = params.changedLeafNodesOnly();
252 final var childNodes = params.childNodesOnly();
254 final var textParams = new TextParameters(
255 leafNodes != null && leafNodes.value(),
256 skipData != null && skipData.value(),
257 changedLeafNodes != null && changedLeafNodes.value(),
258 childNodes != null && childNodes.value());
260 final var filter = params.filter();
261 final var filterValue = filter == null ? null : filter.paramValue();
262 final var formatter = filterValue == null || filterValue.isEmpty() ? factory.getFormatter(textParams)
263 : factory.getFormatter(textParams, filterValue);
266 // Lockless add of a subscriber. If we observe a null this stream is dead before the new subscriber could be
268 final var toAdd = new Subscriber<>(this, handler, formatter);
269 var observed = acquireSubscribers();
270 while (observed != null) {
271 final var next = observed.add(toAdd);
272 final var witness = (Subscribers<T>) SUBSCRIBERS.compareAndExchangeRelease(this, observed, next);
273 if (witness == observed) {
274 LOG.debug("Subscriber {} is added.", handler);
275 if (observed instanceof Subscribers.Empty) {
276 // We have became non-empty, start the source
282 // We have raced: retry the operation
290 * Removes a {@link Subscriber}. If this was the last subscriber also shut down this stream and initiate its removal
293 * @param subscriber The {@link Subscriber} to remove
294 * @throws NullPointerException if {@code subscriber} is {@code null}
296 void removeSubscriber(final Subscriber<T> subscriber) {
297 final var toRemove = requireNonNull(subscriber);
298 var observed = acquireSubscribers();
299 while (observed != null) {
300 final var next = observed.remove(toRemove);
301 final var witness = (Subscribers<T>) SUBSCRIBERS.compareAndExchangeRelease(this, observed, next);
302 if (witness == observed) {
303 LOG.debug("Subscriber {} is removed", subscriber);
305 // We have lost the last subscriber, terminate.
311 // We have raced: retry the operation
316 private Subscribers<T> acquireSubscribers() {
317 return (Subscribers<T>) SUBSCRIBERS.getAcquire(this);
320 private void startSource() {
321 // We have not started the stream yet, make sure that happens. This is a bit more involved, as the source may
322 // immediately issue endOfStream(), which in turn invokes terminate(). But at that point start() has not return
323 // and therefore registration is still null -- and thus we need to see if we are still on-line.
324 final var reg = source.start(sink);
325 synchronized (this) {
326 if (acquireSubscribers() == null) {
334 private void terminate() {
335 synchronized (this) {
336 if (registration != null) {
337 registration.close();
341 registry.removeStream(this);
345 public String toString() {
346 return MoreObjects.toStringHelper(this).add("name", name).add("source", source).toString();