2 * Copyright (c) 2020 Lumina Networks, Inc. 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.nb.rfc8040.streams;
10 import static java.util.Objects.requireNonNull;
12 import com.google.common.collect.ImmutableMap;
13 import java.io.UnsupportedEncodingException;
14 import javax.ws.rs.BadRequestException;
15 import javax.ws.rs.GET;
16 import javax.ws.rs.NotFoundException;
17 import javax.ws.rs.Path;
18 import javax.ws.rs.PathParam;
19 import javax.ws.rs.Produces;
20 import javax.ws.rs.core.Context;
21 import javax.ws.rs.core.MediaType;
22 import javax.ws.rs.core.UriInfo;
23 import javax.ws.rs.sse.Sse;
24 import javax.ws.rs.sse.SseEventSink;
25 import javax.xml.xpath.XPathExpressionException;
26 import org.opendaylight.restconf.nb.rfc8040.ReceiveEventsParams;
27 import org.opendaylight.restconf.server.spi.RestconfStream;
28 import org.opendaylight.restconf.server.spi.RestconfStream.EncodingName;
29 import org.slf4j.Logger;
30 import org.slf4j.LoggerFactory;
33 * Access to notification streams via Server-Sent Events.
36 final class SSEStreamService {
37 private static final Logger LOG = LoggerFactory.getLogger(SSEStreamService.class);
39 private final RestconfStream.Registry streamRegistry;
40 private final PingExecutor pingExecutor;
41 private final int maximumFragmentLength;
42 private final int heartbeatInterval;
44 SSEStreamService(final RestconfStream.Registry streamRegistry, final PingExecutor pingExecutor,
45 final StreamsConfiguration configuration) {
46 this.streamRegistry = requireNonNull(streamRegistry);
47 this.pingExecutor = requireNonNull(pingExecutor);
48 heartbeatInterval = configuration.heartbeatInterval();
49 maximumFragmentLength = configuration.maximumFragmentLength();
53 * Attach to a particular notification stream.
55 * @param streamName path to target
58 @Path("/{encodingName:[a-zA-Z]+}/{streamName:.+}")
59 @Produces(MediaType.SERVER_SENT_EVENTS)
60 public void getSSE(@PathParam("encodingName") final EncodingName encodingName,
61 @PathParam("streamName") final String streamName, @Context final UriInfo uriInfo,
62 @Context final SseEventSink sink, @Context final Sse sse) {
63 final var stream = streamRegistry.lookupStream(streamName);
65 LOG.debug("Listener for stream with name {} was not found.", streamName);
66 throw new NotFoundException("No such stream: " + streamName);
69 final var queryParameters = ImmutableMap.<String, String>builder();
70 for (var entry : uriInfo.getQueryParameters().entrySet()) {
71 final var values = entry.getValue();
72 switch (values.size()) {
77 queryParameters.put(entry.getKey(), values.get(0));
80 throw new BadRequestException(
81 "Parameter " + entry.getKey() + " can appear at most once in request URI");
85 final ReceiveEventsParams params;
87 params = ReceiveEventsParams.ofQueryParameters(queryParameters.build());
88 } catch (IllegalArgumentException e) {
89 throw new BadRequestException(e.getMessage(), e);
92 LOG.debug("Listener for stream with name {} has been found, SSE session handler will be created.", streamName);
93 // FIXME: invert control here: we should call 'listener.addSession()', which in turn should call
94 // handler.init()/handler.close()
95 final var handler = new SSESender(pingExecutor, sink, sse, stream, encodingName, params,
96 maximumFragmentLength, heartbeatInterval);
100 } catch (UnsupportedEncodingException e) {
101 throw new NotFoundException("Unsupported encoding " + encodingName.name(), e);
102 } catch (IllegalArgumentException | XPathExpressionException e) {
103 throw new BadRequestException(e.getMessage(), e);