Fixed JERSEY-2910 & JERSEY-2912.

JERSEY-2910 - SSE EventSource thread factory updated to produce daemon threads that can be terminated when JVM quits.
JERSEY-2912 - Added support for producing / consuming “comments only” SSE events.

Change-Id: Ic9bcc44997b05cc7dabf1c0c3ff3c5bee641296c
Signed-off-by: Marek Potociar <[email protected]>

Author: Marek Potociar

media/sse/src/main/java/org/glassfish/jersey/media/sse/EventOutput.java

@@ -1,7 +1,7 @@
 /*
  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS HEADER.
  *
- * Copyright (c) 2012-2014 Oracle and/or its affiliates. All rights reserved.
+ * Copyright (c) 2012-2015 Oracle and/or its affiliates. All rights reserved.
  *
  * The contents of this file are subject to the terms of either the GNU
  * General Public License Version 2 only ("GPL") or the Common Development
@@ -54,7 +54,7 @@
  */
 public class EventOutput extends ChunkedOutput<OutboundEvent> {
     // encoding does not matter for lower ASCII characters
-    private static final byte[] SSE_EVENT_DELIMITER = "\n\n".getBytes(Charset.forName("UTF-8"));
+    private static final byte[] SSE_EVENT_DELIMITER = "\n".getBytes(Charset.forName("UTF-8"));
 
     /**
      * Create new outbound Server-Sent Events channel.

media/sse/src/main/java/org/glassfish/jersey/media/sse/EventSource.java

@@ -49,7 +49,6 @@
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.logging.Level;
@@ -61,6 +60,8 @@
 
 import org.glassfish.jersey.internal.util.ExtendedLogger;
 
+import jersey.repackaged.com.google.common.util.concurrent.ThreadFactoryBuilder;
+
 /**
  * Client for reading and processing {@link InboundEvent incoming Server-Sent Events}.
  * <p>
@@ -139,7 +140,7 @@ public class EventSource implements EventListener {
      */
     public static final long RECONNECT_DEFAULT = 500;
 
-    private static enum State {
+    private enum State {
         READY, OPEN, CLOSED
     }
 
@@ -232,6 +233,7 @@ public Builder named(String name) {
          *
          * @return updated event source builder instance.
          */
+        @SuppressWarnings("unused")
         public Builder usePersistentConnections() {
             disableKeepAlive = false;
             return this;
@@ -249,6 +251,7 @@ public Builder usePersistentConnections() {
          * @param unit  time unit of the reconnect delay parameter.
          * @return updated event source builder instance.
          */
+        @SuppressWarnings("unused")
         public Builder reconnectingEvery(final long delay, TimeUnit unit) {
             reconnect = unit.toMillis(delay);
             return this;
@@ -380,12 +383,8 @@ private EventSource(final WebTarget target,
         this.disableKeepAlive = disableKeepAlive;
 
         final String esName = (name == null) ? createDefaultName(target) : name;
-        this.executor = Executors.newSingleThreadScheduledExecutor(new ThreadFactory() {
-            @Override
-            public Thread newThread(Runnable r) {
-                return new Thread(r, esName);
-            }
-        });
+        this.executor = Executors.newSingleThreadScheduledExecutor(
+                new ThreadFactoryBuilder().setNameFormat(esName + "-%d").setDaemon(true).build());
 
         if (open) {
             open();

media/sse/src/main/java/org/glassfish/jersey/media/sse/InboundEvent.java

@@ -60,10 +60,12 @@
  * @author Marek Potociar (marek.potociar at oracle.com)
  */
 public class InboundEvent {
-    private static final GenericType<String> STRING_AS_GENERIC_TYPE = new GenericType<String>(String.class);
+
+    private static final GenericType<String> STRING_AS_GENERIC_TYPE = new GenericType<>(String.class);
 
     private final String name;
     private final String id;
+    private final String comment;
     private final byte[] data;
     private final long reconnectDelay;
 
@@ -76,6 +78,7 @@ public class InboundEvent {
      * Inbound event builder. This implementation is not thread-safe.
      */
     static class Builder {
+
         private String name;
         private String id;
         private long reconnectDelay = SseFeature.RECONNECT_NOT_SET;
@@ -85,6 +88,7 @@ static class Builder {
         private final Annotation[] annotations;
         private final MediaType mediaType;
         private final MultivaluedMap<String, String> headers;
+        private final StringBuilder commentBuilder;
 
         /**
          * Create new inbound event builder.
@@ -106,12 +110,13 @@ public Builder(MessageBodyWorkers workers,
             this.mediaType = mediaType;
             this.headers = headers;
 
+            this.commentBuilder = new StringBuilder();
             this.dataStream = new ByteArrayOutputStream();
         }
 
         /**
          * Set inbound event name.
-         *
+         * <p/>
          * Value of the received SSE {@code "event"} field.
          *
          * @param name {@code "event"} field value.
@@ -124,7 +129,7 @@ public Builder name(String name) {
 
         /**
          * Set inbound event identifier.
-         *
+         * <p/>
          * Value of the received SSE {@code "id"} field.
          *
          * @param id {@code "id"} field value.
@@ -135,12 +140,32 @@ public Builder id(String id) {
             return this;
         }
 
+        /**
+         * Add a comment line to the event.
+         * <p>
+         * The comment line will be added to the received SSE event comment as a new line in the comment field.
+         * If the comment line parameter is {@code null}, the call will be ignored.
+         * </p>
+         *
+         * @param commentLine comment line to be added to the event comment.
+         * @return updated builder instance.
+         * @since 2.21
+         */
+        public Builder commentLine(final CharSequence commentLine) {
+            if (commentLine != null) {
+                commentBuilder.append(commentLine).append('\n');
+            }
+
+            return this;
+        }
+
         /**
          * Set reconnection delay (in milliseconds) that indicates how long the event receiver should wait
          * before attempting to reconnect in case a connection to SSE event source is lost.
          * <p>
          * Value of the received SSE {@code "retry"} field.
          * </p>
+         *
          * @param milliseconds reconnection delay in milliseconds. Negative values un-set the reconnection delay.
          * @return updated builder instance.
          * @since 2.3
@@ -181,6 +206,7 @@ public InboundEvent build() {
             return new InboundEvent(
                     name,
                     id,
+                    commentBuilder.length() > 0 ? commentBuilder.substring(0, commentBuilder.length() - 1) : null,
                     reconnectDelay,
                     dataStream.toByteArray(),
                     workers,
@@ -190,16 +216,18 @@ public InboundEvent build() {
         }
     }
 
-    private InboundEvent(String name,
-                         String id,
-                         long reconnectDelay,
-                         byte[] data,
-                         MessageBodyWorkers messageBodyWorkers,
-                         Annotation[] annotations,
-                         MediaType mediaType,
-                         MultivaluedMap<String, String> headers) {
+    private InboundEvent(final String name,
+                         final String id,
+                         final String comment,
+                         final long reconnectDelay,
+                         final byte[] data,
+                         final MessageBodyWorkers messageBodyWorkers,
+                         final Annotation[] annotations,
+                         final MediaType mediaType,
+                         final MultivaluedMap<String, String> headers) {
         this.name = name;
         this.id = id;
+        this.comment = comment;
         this.reconnectDelay = reconnectDelay;
         this.data = data;
         this.messageBodyWorkers = messageBodyWorkers;
@@ -235,6 +263,20 @@ public String getId() {
         return id;
     }
 
+    /**
+     * Get a comment string that accompanies the event.
+     * <p>
+     * Contains value of the comment associated with SSE event. This field is optional. Method may return {@code null},
+     * if the event comment is not specified.
+     * </p>
+     *
+     * @return comment associated with the event.
+     * @since 2.21
+     */
+    public String getComment() {
+        return comment;
+    }
+
     /**
      * Get new connection retry time in milliseconds the event receiver should wait before attempting to
      * reconnect after a connection to the SSE event source is lost.
@@ -273,8 +315,7 @@ public boolean isEmpty() {
      * Get the original event data string {@link String}.
      *
      * @return event data de-serialized into a string.
-     * @throws javax.ws.rs.ProcessingException
-     *          when provided type can't be read. The thrown exception wraps the original cause.
+     * @throws javax.ws.rs.ProcessingException when provided type can't be read. The thrown exception wraps the original cause.
      * @since 2.3
      */
     public String readData() {
@@ -286,8 +327,7 @@ public String readData() {
      *
      * @param type Java type to be used for event data de-serialization.
      * @return event data de-serialized as an instance of a given type.
-     * @throws javax.ws.rs.ProcessingException
-     *          when provided type can't be read. The thrown exception wraps the original cause.
+     * @throws javax.ws.rs.ProcessingException when provided type can't be read. The thrown exception wraps the original cause.
      * @since 2.3
      */
     public <T> T readData(Class<T> type) {
@@ -299,10 +339,10 @@ public <T> T readData(Class<T> type) {
      *
      * @param type generic type to be used for event data de-serialization.
      * @return event data de-serialized as an instance of a given type.
-     * @throws javax.ws.rs.ProcessingException
-     *          when provided type can't be read. The thrown exception wraps the original cause.
+     * @throws javax.ws.rs.ProcessingException when provided type can't be read. The thrown exception wraps the original cause.
      * @since 2.3
      */
+    @SuppressWarnings("unused")
     public <T> T readData(GenericType<T> type) {
         return readData(type, null);
     }
@@ -313,22 +353,21 @@ public <T> T readData(GenericType<T> type) {
      * @param messageType Java type to be used for event data de-serialization.
      * @param mediaType   {@link MediaType media type} to be used for event data de-serialization.
      * @return event data de-serialized as an instance of a given type.
-     * @throws javax.ws.rs.ProcessingException
-     *          when provided type can't be read. The thrown exception wraps the original cause.
+     * @throws javax.ws.rs.ProcessingException when provided type can't be read. The thrown exception wraps the original cause.
      * @since 2.3
      */
+    @SuppressWarnings("unused")
     public <T> T readData(Class<T> messageType, MediaType mediaType) {
         return readData(new GenericType<T>(messageType), mediaType);
     }
 
     /**
      * Read event data as a given generic type.
      *
-     * @param type generic type to be used for event data de-serialization.
-     * @param mediaType   {@link MediaType media type} to be used for event data de-serialization.
+     * @param type      generic type to be used for event data de-serialization.
+     * @param mediaType {@link MediaType media type} to be used for event data de-serialization.
      * @return event data de-serialized as an instance of a given type.
-     * @throws javax.ws.rs.ProcessingException
-     *          when provided type can't be read. The thrown exception wraps the original cause.
+     * @throws javax.ws.rs.ProcessingException when provided type can't be read. The thrown exception wraps the original cause.
      * @since 2.3
      */
     public <T> T readData(GenericType<T> type, MediaType mediaType) {
@@ -360,8 +399,9 @@ private <T> T readAndCast(GenericType<T> type, MediaType effectiveMediaType, Mes
      * Get the raw event data bytes.
      *
      * @return raw event data bytes. The returned byte array may be empty if the event does not
-     *         contain any data.
+     * contain any data.
      */
+    @SuppressWarnings("unused")
     public byte[] getRawData() {
         if (isEmpty()) {
             return data;
@@ -383,6 +423,7 @@ public String toString() {
         return "InboundEvent{"
                 + "name='" + name + '\''
                 + ", id='" + id + '\''
+                + ", comment=" + (comment == null ? "[no comments]" : '\'' + comment + '\'')
                 + ", data=" + s
                 + '}';
     }

media/sse/src/main/java/org/glassfish/jersey/media/sse/InboundEventReader.java

@@ -1,7 +1,7 @@
 /*
  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS HEADER.
  *
- * Copyright (c) 2012-2014 Oracle and/or its affiliates. All rights reserved.
+ * Copyright (c) 2012-2015 Oracle and/or its affiliates. All rights reserved.
  *
  * The contents of this file are subject to the terms of either the GNU
  * General Public License Version 2 only ("GPL") or the Common Development
@@ -74,7 +74,7 @@ class InboundEventReader implements MessageBodyReader<InboundEvent> {
     @Inject
     private Provider<MessageBodyWorkers> messageBodyWorkers;
 
-    private static enum State {
+    private enum State {
         NEW_LINE,
         COMMENT,
         FIELD,
@@ -100,6 +100,7 @@ public InboundEvent readFrom(final Class<InboundEvent> type,
          * last editors draft from 13 March 2012
          */
         final ByteArrayOutputStream tokenData = new ByteArrayOutputStream();
+        final String charsetName = MessageUtils.getCharset(mediaType).name();
         final InboundEvent.Builder eventBuilder =
                 new InboundEvent.Builder(messageBodyWorkers.get(), annotations, mediaType, headers);
 
@@ -123,13 +124,16 @@ public InboundEvent readFrom(final Class<InboundEvent> type,
                     break;
                 case COMMENT:
                     // skipping comment data
-                    b = readLineUntil(entityStream, '\n', null);
+                    b = readLineUntil(entityStream, '\n', tokenData);
+                    final String commentLine = tokenData.toString(charsetName);
+                    tokenData.reset();
+                    eventBuilder.commentLine(commentLine.trim());
                     currentState = State.NEW_LINE;
                     break;
                 case FIELD:
                     // read field name
                     b = readLineUntil(entityStream, ':', tokenData);
-                    final String fieldName = tokenData.toString(MessageUtils.getCharset(mediaType).name());
+                    final String fieldName = tokenData.toString(charsetName);
                     tokenData.reset();
 
                     if (b == ':') {

media/sse/src/main/java/org/glassfish/jersey/media/sse/OutboundEvent.java

@@ -1,7 +1,7 @@
 /*
  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS HEADER.
  *
- * Copyright (c) 2012-2013 Oracle and/or its affiliates. All rights reserved.
+ * Copyright (c) 2012-2015 Oracle and/or its affiliates. All rights reserved.
  *
  * The contents of this file are subject to the terms of either the GNU
  * General Public License Version 2 only ("GPL") or the Common Development
@@ -349,10 +349,10 @@ public boolean isReconnectDelaySet() {
      * serializing the {@link #getData() event data}.
      * </p>
      *
-     * @return data type.
+     * @return data type. May return {@code null}, if the event does not contain any data.
      */
     public Class<?> getType() {
-        return type.getRawType();
+        return type == null ? null : type.getRawType();
     }
 
     /**
@@ -362,11 +362,11 @@ public Class<?> getType() {
      * serializing the {@link #getData() event data}.
      * </p>
      *
-     * @return generic data type.
+     * @return generic data type. May return {@code null}, if the event does not contain any data.
      * @since 2.3
      */
     public Type getGenericType() {
-        return type.getType();
+        return type == null ? null : type.getType();
     }
 
     /**
@@ -406,7 +406,7 @@ public String getComment() {
      * {@link #getType() type}, {@link #getGenericType()} generic type} and {@link #getMediaType()} media type}.
      * </p>
      *
-     * @return event data.
+     * @return event data. May return {@code null}, if the event does not contain any data.
      */
     public Object getData() {
         return data;

media/sse/src/main/java/org/glassfish/jersey/media/sse/OutboundEventWriter.java

@@ -155,6 +155,7 @@ public void write(final int i) throws IOException {
                             }
                         }
                     });
+            entityStream.write(EOL);
         }
     }
 }