restconf/restconf-nb/src/main/java/org/opendaylight/restconf/nb/rfc8040/streams/SSESessionHandler.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.streams;
   9
  10 import com.google.common.base.CharMatcher;
  11 import com.google.common.base.Strings;
  12 import java.util.concurrent.ScheduledExecutorService;
  13 import java.util.concurrent.ScheduledFuture;
  14 import java.util.concurrent.TimeUnit;
  15 import javax.ws.rs.sse.Sse;
  16 import javax.ws.rs.sse.SseEventSink;
  17 import org.opendaylight.restconf.nb.rfc8040.streams.listeners.BaseListenerInterface;
  18 import org.slf4j.Logger;
  19 import org.slf4j.LoggerFactory;
  20
  21 /**
  22  * SSE session handler that is responsible for controlling of session, managing subscription to data-change-event or
  23  * notification listener, and sending of data over established SSE session.
  24  */
  25 public final class SSESessionHandler implements StreamSessionHandler {
  26     private static final Logger LOG = LoggerFactory.getLogger(SSESessionHandler.class);
  27     private static final String PING_PAYLOAD = "ping";
  28
  29     private static final CharMatcher CR_OR_LF = CharMatcher.anyOf("\r\n");
  30
  31     private final ScheduledExecutorService executorService;
  32     private final BaseListenerInterface listener;
  33     private final int maximumFragmentLength;
  34     private final int heartbeatInterval;
  35     private final SseEventSink sink;
  36     private final Sse sse;
  37
  38     private ScheduledFuture<?> pingProcess;
  39
  40     /**
  41      * Creation of the new server-sent events session handler.
  42      *
  43      * @param executorService Executor that is used for periodical sending of SSE ping messages to keep session up even
  44      *            if the notifications doesn't flow from server to clients or clients don't implement ping-pong
  45      *            service.
  46      * @param listener YANG notification or data-change event listener to which client on this SSE session subscribes
  47      *            to.
  48      * @param maximumFragmentLength Maximum fragment length in number of Unicode code units (characters). If this
  49      *            parameter is set to 0, the maximum fragment length is disabled and messages up to 64 KB can be sent
  50      *            (exceeded notification length ends in error). If the parameter is set to non-zero positive value,
  51      *            messages longer than this parameter are fragmented into multiple SSE messages sent in one
  52      *            transaction.
  53      * @param heartbeatInterval Interval in milliseconds of sending of ping control frames to remote endpoint to keep
  54      *            session up. Ping control frames are disabled if this parameter is set to 0.
  55      */
  56     public SSESessionHandler(final ScheduledExecutorService executorService, final SseEventSink sink, final Sse sse,
  57             final BaseListenerInterface listener, final int maximumFragmentLength, final int heartbeatInterval) {
  58         this.executorService = executorService;
  59         this.sse = sse;
  60         this.sink = sink;
  61         this.listener = listener;
  62         this.maximumFragmentLength = maximumFragmentLength;
  63         this.heartbeatInterval = heartbeatInterval;
  64     }
  65
  66     /**
  67      * Initialization of SSE connection. SSE session handler is registered at data-change-event / YANG notification
  68      * listener and the heartbeat ping process is started if it is enabled.
  69      */
  70     public synchronized void init() {
  71         listener.addSubscriber(this);
  72         if (heartbeatInterval != 0) {
  73             pingProcess = executorService.scheduleWithFixedDelay(this::sendPingMessage, heartbeatInterval,
  74                     heartbeatInterval, TimeUnit.MILLISECONDS);
  75         }
  76     }
  77
  78     /**
  79      * Handling of SSE session close event. Removal of subscription at listener and stopping of the ping process.
  80      */
  81     public synchronized void close() {
  82         listener.removeSubscriber(this);
  83         stopPingProcess();
  84     }
  85
  86     /**
  87      * Sending of string message to outbound Server-Sent Events channel {@link SseEventSink}. SSE is automatically split
  88      * to fragments with new line character. If the maximum fragment length is set to non-zero positive value and input
  89      * message exceeds this value, message is manually fragmented to multiple message fragments which are send
  90      * individually. Previous fragmentation is removed.
  91      *
  92      * @param message Message data to be send over web-socket session.
  93      */
  94     @Override
  95     public synchronized void sendDataMessage(final String message) {
  96         if (Strings.isNullOrEmpty(message)) {
  97             // FIXME: should this be tolerated?
  98             return;
  99         }
 100         if (!sink.isClosed()) {
 101             final String toSend = maximumFragmentLength != 0 && message.length() > maximumFragmentLength
 102                 ? splitMessageToFragments(message) : message;
 103             sink.send(sse.newEvent(toSend));
 104         } else {
 105             close();
 106         }
 107     }
 108
 109     /**
 110      * Split message to fragments. SSE automatically fragment string with new line character.
 111      * For manual fragmentation we will remove all new line characters
 112      *
 113      * @param message Message data to be split.
 114      * @return splitted message
 115      */
 116     private String splitMessageToFragments(final String message) {
 117         StringBuilder outputMessage = new StringBuilder();
 118         String inputmessage = CR_OR_LF.removeFrom(message);
 119         int length = inputmessage.length();
 120         for (int i = 0; i < length; i += maximumFragmentLength) {
 121             outputMessage.append(inputmessage, i, Math.min(length, i + maximumFragmentLength)).append("\r\n");
 122         }
 123         return outputMessage.toString();
 124     }
 125
 126     private synchronized void sendPingMessage() {
 127         if (!sink.isClosed()) {
 128             LOG.debug("sending PING:{}", PING_PAYLOAD);
 129             sink.send(sse.newEventBuilder().comment(PING_PAYLOAD).build());
 130         } else {
 131             close();
 132         }
 133     }
 134
 135     private void stopPingProcess() {
 136         if (pingProcess != null && !pingProcess.isDone() && !pingProcess.isCancelled()) {
 137             pingProcess.cancel(true);
 138         }
 139     }
 140
 141     @Override
 142     public synchronized boolean isConnected() {
 143         return !sink.isClosed();
 144     }
 145
 146     // TODO:return some type of identification of connection
 147     @Override
 148     public String toString() {
 149         return sink.toString();
 150     }
 151 }