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.api.query.PrettyPrintParam;
27 import org.opendaylight.restconf.server.api.EventStreamGetParams;
28 import org.opendaylight.restconf.server.api.QueryParams;
29 import org.opendaylight.restconf.server.spi.RestconfStream;
30 import org.opendaylight.restconf.server.spi.RestconfStream.EncodingName;
31 import org.slf4j.Logger;
32 import org.slf4j.LoggerFactory;
35 * Access to notification streams via Server-Sent Events.
38 // FIXME: integrate this service into JaxRsRestconf once we remove support for WebSockets
39 final class SSEStreamService {
40 private static final Logger LOG = LoggerFactory.getLogger(SSEStreamService.class);
42 private final RestconfStream.Registry streamRegistry;
43 private final PingExecutor pingExecutor;
44 private final int maximumFragmentLength;
45 private final int heartbeatInterval;
47 SSEStreamService(final RestconfStream.Registry streamRegistry, final PingExecutor pingExecutor,
48 final StreamsConfiguration configuration) {
49 this.streamRegistry = requireNonNull(streamRegistry);
50 this.pingExecutor = requireNonNull(pingExecutor);
51 heartbeatInterval = configuration.heartbeatInterval();
52 maximumFragmentLength = configuration.maximumFragmentLength();
56 * Attach to a particular notification stream.
58 * @param streamName path to target
61 @Path("/{encodingName:[a-zA-Z]+}/{streamName:.+}")
62 @Produces(MediaType.SERVER_SENT_EVENTS)
63 public void getSSE(@PathParam("encodingName") final EncodingName encodingName,
64 @PathParam("streamName") final String streamName, @Context final UriInfo uriInfo,
65 @Context final SseEventSink sink, @Context final Sse sse) {
66 final var stream = streamRegistry.lookupStream(streamName);
68 LOG.debug("Listener for stream with name {} was not found.", streamName);
69 throw new NotFoundException("No such stream: " + streamName);
72 final EventStreamGetParams getParams;
74 getParams = EventStreamGetParams.of(new QueryParams(
75 // FIXME: figure this out
76 QueryParameters.ofMultiValue(uriInfo.getQueryParameters()), PrettyPrintParam.FALSE));
77 } catch (IllegalArgumentException e) {
78 throw new BadRequestException(e.getMessage(), e);
81 LOG.debug("Listener for stream with name {} has been found, SSE session handler will be created.", streamName);
82 // FIXME: invert control here: we should call 'listener.addSession()', which in turn should call
83 // handler.init()/handler.close()
84 final var handler = new SSESender(pingExecutor, sink, sse, stream, encodingName, getParams,
85 maximumFragmentLength, heartbeatInterval);
89 } catch (UnsupportedEncodingException e) {
90 throw new NotFoundException("Unsupported encoding " + encodingName.name(), e);
91 } catch (IllegalArgumentException | XPathExpressionException e) {
92 throw new BadRequestException(e.getMessage(), e);