restconf/restconf-nb/src/main/java/org/opendaylight/restconf/nb/rfc8040/rests/services/impl/RestconfDataStreamServiceImpl.java

   1 /*
   2  * Copyright (c) 2020 Lumina Networks, Inc. All rights reserved.
   3  *
   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
   7  */
   8 package org.opendaylight.restconf.nb.rfc8040.rests.services.impl;
   9
  10 import static java.util.Objects.requireNonNull;
  11
  12 import com.google.common.collect.ImmutableMap;
  13 import java.io.UnsupportedEncodingException;
  14 import java.util.concurrent.ScheduledExecutorService;
  15 import javax.ws.rs.BadRequestException;
  16 import javax.ws.rs.GET;
  17 import javax.ws.rs.NotFoundException;
  18 import javax.ws.rs.Path;
  19 import javax.ws.rs.PathParam;
  20 import javax.ws.rs.Produces;
  21 import javax.ws.rs.core.Context;
  22 import javax.ws.rs.core.MediaType;
  23 import javax.ws.rs.core.UriInfo;
  24 import javax.ws.rs.sse.Sse;
  25 import javax.ws.rs.sse.SseEventSink;
  26 import javax.xml.xpath.XPathExpressionException;
  27 import org.opendaylight.controller.config.threadpool.ScheduledThreadPool;
  28 import org.opendaylight.restconf.nb.rfc8040.ReceiveEventsParams;
  29 import org.opendaylight.restconf.nb.rfc8040.streams.ListenersBroker;
  30 import org.opendaylight.restconf.nb.rfc8040.streams.RestconfStream.EncodingName;
  31 import org.opendaylight.restconf.nb.rfc8040.streams.SSESessionHandler;
  32 import org.opendaylight.restconf.nb.rfc8040.streams.StreamsConfiguration;
  33 import org.slf4j.Logger;
  34 import org.slf4j.LoggerFactory;
  35
  36 /**
  37  * Access to notification streams via Server-Sent Events.
  38  */
  39 @Path("/")
  40 public final class RestconfDataStreamServiceImpl {
  41     private static final Logger LOG = LoggerFactory.getLogger(RestconfDataStreamServiceImpl.class);
  42
  43     private final ListenersBroker listenersBroker;
  44     private final ScheduledExecutorService executorService;
  45     private final int maximumFragmentLength;
  46     private final int heartbeatInterval;
  47
  48     public RestconfDataStreamServiceImpl(final ScheduledThreadPool scheduledThreadPool,
  49             final ListenersBroker listenersBroker, final StreamsConfiguration configuration) {
  50         executorService = scheduledThreadPool.getExecutor();
  51         this.listenersBroker = requireNonNull(listenersBroker);
  52         heartbeatInterval = configuration.heartbeatInterval();
  53         maximumFragmentLength = configuration.maximumFragmentLength();
  54     }
  55
  56     /**
  57      * Attach to a particular notification stream.
  58      *
  59      * @param streamName path to target
  60      */
  61     @GET
  62     @Path("/{encodingName:[a-zA-Z]+}/{streamName:.+}")
  63     @Produces(MediaType.SERVER_SENT_EVENTS)
  64     public void getSSE(@PathParam("encodingName") final EncodingName encodingName,
  65             @PathParam("streamName") final String streamName, @Context final UriInfo uriInfo,
  66             @Context final SseEventSink sink, @Context final Sse sse) {
  67         final var stream = listenersBroker.getStream(streamName);
  68         if (stream == null) {
  69             LOG.debug("Listener for stream with name {} was not found.", streamName);
  70             throw new NotFoundException("No such stream: " + streamName);
  71         }
  72
  73         final var queryParameters = ImmutableMap.<String, String>builder();
  74         for (var entry : uriInfo.getQueryParameters().entrySet()) {
  75             final var values = entry.getValue();
  76             switch (values.size()) {
  77                 case 0:
  78                     // No-op
  79                     break;
  80                 case 1:
  81                     queryParameters.put(entry.getKey(), values.get(0));
  82                     break;
  83                 default:
  84                     throw new BadRequestException(
  85                         "Parameter " + entry.getKey() + " can appear at most once in request URI");
  86             }
  87         }
  88
  89         final ReceiveEventsParams params;
  90         try {
  91             params = ReceiveEventsParams.ofQueryParameters(queryParameters.build());
  92         } catch (IllegalArgumentException e) {
  93             throw new BadRequestException(e.getMessage(), e);
  94         }
  95
  96         LOG.debug("Listener for stream with name {} has been found, SSE session handler will be created.", streamName);
  97         // FIXME: invert control here: we should call 'listener.addSession()', which in turn should call
  98         //        handler.init()/handler.close()
  99         final var handler = new SSESessionHandler(executorService, sink, sse, stream, encodingName, params,
 100             maximumFragmentLength, heartbeatInterval);
 101
 102         try {
 103             handler.init();
 104         } catch (UnsupportedEncodingException e) {
 105             throw new NotFoundException("Unsupported encoding " + encodingName.name(), e);
 106         } catch (IllegalArgumentException | XPathExpressionException e) {
 107             throw new BadRequestException(e.getMessage(), e);
 108         }
 109     }
 110 }