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 java.io.UnsupportedEncodingException;
13 import javax.ws.rs.BadRequestException;
14 import javax.ws.rs.GET;
15 import javax.ws.rs.NotFoundException;
16 import javax.ws.rs.Path;
17 import javax.ws.rs.PathParam;
18 import javax.ws.rs.Produces;
19 import javax.ws.rs.core.Context;
20 import javax.ws.rs.core.MediaType;
21 import javax.ws.rs.core.UriInfo;
22 import javax.ws.rs.sse.Sse;
23 import javax.ws.rs.sse.SseEventSink;
24 import javax.xml.xpath.XPathExpressionException;
25 import org.opendaylight.restconf.api.QueryParameters;
26 import org.opendaylight.restconf.server.api.EventStreamGetParams;
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 // FIXME: integrate this service into JaxRsRestconf once we remove support for WebSockets
37 public final class SSEStreamService {
38 private static final Logger LOG = LoggerFactory.getLogger(SSEStreamService.class);
40 private final RestconfStream.Registry streamRegistry;
41 private final PingExecutor pingExecutor;
42 private final int maximumFragmentLength;
43 private final int heartbeatInterval;
45 SSEStreamService(final RestconfStream.Registry streamRegistry, final PingExecutor pingExecutor,
46 final StreamsConfiguration configuration) {
47 this.streamRegistry = requireNonNull(streamRegistry);
48 this.pingExecutor = requireNonNull(pingExecutor);
49 heartbeatInterval = configuration.heartbeatInterval();
50 maximumFragmentLength = configuration.maximumFragmentLength();
54 * Attach to a particular notification stream.
56 * @param streamName path to target
59 @Path("/{encodingName:[a-zA-Z]+}/{streamName:.+}")
60 @Produces(MediaType.SERVER_SENT_EVENTS)
61 public void getSSE(@PathParam("encodingName") final EncodingName encodingName,
62 @PathParam("streamName") final String streamName, @Context final UriInfo uriInfo,
63 @Context final SseEventSink sink, @Context final Sse sse) {
64 final var stream = streamRegistry.lookupStream(streamName);
66 LOG.debug("Listener for stream with name {} was not found.", streamName);
67 throw new NotFoundException("No such stream: " + streamName);
70 final EventStreamGetParams getParams;
72 getParams = EventStreamGetParams.of(QueryParameters.ofMultiValue(uriInfo.getQueryParameters()));
73 } catch (IllegalArgumentException e) {
74 throw new BadRequestException(e.getMessage(), e);
77 LOG.debug("Listener for stream with name {} has been found, SSE session handler will be created.", streamName);
78 // FIXME: invert control here: we should call 'listener.addSession()', which in turn should call
79 // handler.init()/handler.close()
80 final var handler = new SSESender(pingExecutor, sink, sse, stream, encodingName, getParams,
81 maximumFragmentLength, heartbeatInterval);
85 } catch (UnsupportedEncodingException e) {
86 throw new NotFoundException("Unsupported encoding " + encodingName.name(), e);
87 } catch (IllegalArgumentException | XPathExpressionException e) {
88 throw new BadRequestException(e.getMessage(), e);