Initial Contribution
Signed-off-by: Jan Supol <jan.supol@oracle.com>
diff --git a/examples/sse-item-store-jaxrs-webapp/README.MD b/examples/sse-item-store-jaxrs-webapp/README.MD
new file mode 100644
index 0000000..2948b07
--- /dev/null
+++ b/examples/sse-item-store-jaxrs-webapp/README.MD
@@ -0,0 +1,67 @@
+[//]: # " Copyright (c) 2015, 2018 Oracle and/or its affiliates. All rights reserved. "
+[//]: # " "
+[//]: # " This program and the accompanying materials are made available under the "
+[//]: # " terms of the Eclipse Distribution License v. 1.0, which is available at "
+[//]: # " http://www.eclipse.org/org/documents/edl-v10.php. "
+[//]: # " "
+[//]: # " SPDX-License-Identifier: BSD-3-Clause "
+
+JAX-RS SSE Item Store Jersey Example
+====================================
+
+An example demonstrating how new Server Sent Events (SSE) JAX-RS API can
+be used to notify clients about changes in server-side managed data
+(collection of items). The example also outlines how SSE events can be
+consumed using javascript browser-based clients.
+
+Contents
+--------
+
+The example consists of a web-based client application and a server-side
+application deployed on a [Jetty servlet container](http://www.eclipse.org/jetty/documentation/current/).
+
+The server side part of the application consists of a JAX-RS resource
+managing collection of string items and a simple HTML page that includes
+a browser SSE client written in Javascript as well as a basic CSS
+stylesheet. The SSE Javascript client connects to the JAX-RS resource
+and transforms the streamed messages into HTML code that is rendered by
+the browser. The javascript client also demonstrates how named and
+unnamed SSE events are handled by HTML5 `EventSource` component. The
+mapping of the URI path space of the server-side part of the application
+is presented in the following table:
+
+URI path | Resource class | HTTP methods
+------------------------------- | ------------------- | --------------
+**_/resources/items_** | ItemStoreResource | GET, POST
+**_/resources/items/events_** | ItemStoreResource | GET (SSE)
+
+Application is configured to run using Jetty maven plugin under base
+path `sse-item-store-jaxrs-webapp`.
+
+Running the Example
+-------------------
+
+> mvn clean compile jetty:run
+
+The command above deploys the current example. After successful
+deployment, you should be able to access the browser SSE client page at
+<http://localhost:8080/sse-item-store-jaxrs-webapp/index.html>. To see the raw
+SSE event stream, you may also point your browser directly at the
+[`ItemStoreResource`](http://localhost:8080/sse-item-store-jaxrs-webapp/resources/items/events).
+
+Deploying the example to another servlet container.
+---------------------------------------------------
+
+> mvn clean package
+
+The command above creates a Servlet 3.x compliant WAR located in the
+target directory. The WAR can be then deployed to your Servlet 3.x
+compliant container.
+
+Running Test Client
+-------------------
+
+After deploying the application into a Servlet 3.x compliant container,
+you can run the attached test by executing:
+
+> mvn test -Djersey.config.test.container.factory=org.glassfish.jersey.test.external.ExternalTestContainerFactory -Djersey.config.test.container.port=<port>
diff --git a/examples/sse-item-store-jaxrs-webapp/pom.xml b/examples/sse-item-store-jaxrs-webapp/pom.xml
new file mode 100644
index 0000000..afba6c1
--- /dev/null
+++ b/examples/sse-item-store-jaxrs-webapp/pom.xml
@@ -0,0 +1,101 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+ Copyright (c) 2017, 2018 Oracle and/or its affiliates. All rights reserved.
+
+ This program and the accompanying materials are made available under the
+ terms of the Eclipse Distribution License v. 1.0, which is available at
+ http://www.eclipse.org/org/documents/edl-v10.php.
+
+ SPDX-License-Identifier: BSD-3-Clause
+
+-->
+
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <groupId>org.glassfish.jersey.examples</groupId>
+ <artifactId>webapp-example-parent</artifactId>
+ <relativePath>../webapp-example-parent/pom.xml</relativePath>
+ <version>2.28-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>sse-item-store-jaxrs-webapp</artifactId>
+ <packaging>war</packaging>
+ <name>jersey-examples-sse-item-store-jaxrs-webapp</name>
+
+ <description>Jersey JAX-RS 2.1 SSE API-based item store example.</description>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.glassfish.jersey.containers</groupId>
+ <artifactId>jersey-container-servlet</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>org.glassfish.jersey.inject</groupId>
+ <artifactId>jersey-hk2</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>org.glassfish.jersey.media</groupId>
+ <artifactId>jersey-media-sse</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>org.glassfish.jersey.connectors</groupId>
+ <artifactId>jersey-apache-connector</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+
+ <build>
+ <plugins>
+ <!-- Run the application using "mvn jetty:run" -->
+ <plugin>
+ <!-- TODO unify Jetty in all examples -->
+ <groupId>org.eclipse.jetty</groupId>
+ <artifactId>jetty-maven-plugin</artifactId>
+ <configuration>
+ <scanIntervalSeconds>10</scanIntervalSeconds>
+ <stopPort>9999</stopPort>
+ <stopKey>STOP</stopKey>
+ <webApp>
+ <contextPath>/${project.artifactId}</contextPath>
+ </webApp>
+ <systemProperties>
+ <systemProperty>
+ <name>jetty.port</name>
+ <value>${jersey.config.test.container.port}</value>
+ </systemProperty>
+ </systemProperties>
+ </configuration>
+ </plugin>
+ </plugins>
+ </build>
+
+ <profiles>
+ <profile>
+ <id>release</id>
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.codehaus.mojo</groupId>
+ <artifactId>xml-maven-plugin</artifactId>
+ </plugin>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-assembly-plugin</artifactId>
+ </plugin>
+ </plugins>
+ </build>
+ </profile>
+ </profiles>
+
+ <properties>
+ <jersey.config.test.container.port>8080</jersey.config.test.container.port>
+ </properties>
+</project>
diff --git a/examples/sse-item-store-jaxrs-webapp/src/main/java/org/glassfish/jersey/examples/sseitemstore/jaxrs/JaxrsItemStoreApp.java b/examples/sse-item-store-jaxrs-webapp/src/main/java/org/glassfish/jersey/examples/sseitemstore/jaxrs/JaxrsItemStoreApp.java
new file mode 100644
index 0000000..ed76ecf
--- /dev/null
+++ b/examples/sse-item-store-jaxrs-webapp/src/main/java/org/glassfish/jersey/examples/sseitemstore/jaxrs/JaxrsItemStoreApp.java
@@ -0,0 +1,30 @@
+/*
+ * Copyright (c) 2017, 2018 Oracle and/or its affiliates. All rights reserved.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Distribution License v. 1.0, which is available at
+ * http://www.eclipse.org/org/documents/edl-v10.php.
+ *
+ * SPDX-License-Identifier: BSD-3-Clause
+ */
+
+package org.glassfish.jersey.examples.sseitemstore.jaxrs;
+
+import javax.ws.rs.ApplicationPath;
+
+import org.glassfish.jersey.server.ResourceConfig;
+
+/**
+ * SSE item store JAX-RS application class.
+ *
+ * @author Adam Lindenthal (adam.lindenthal at oracle.com)
+ */
+@ApplicationPath("resources")
+public class JaxrsItemStoreApp extends ResourceConfig {
+ /**
+ * Create new SSE Item Store Example JAX-RS application.
+ */
+ public JaxrsItemStoreApp() {
+ super(JaxrsItemStoreResource.class);
+ }
+}
diff --git a/examples/sse-item-store-jaxrs-webapp/src/main/java/org/glassfish/jersey/examples/sseitemstore/jaxrs/JaxrsItemStoreResource.java b/examples/sse-item-store-jaxrs-webapp/src/main/java/org/glassfish/jersey/examples/sseitemstore/jaxrs/JaxrsItemStoreResource.java
new file mode 100644
index 0000000..a8c193d
--- /dev/null
+++ b/examples/sse-item-store-jaxrs-webapp/src/main/java/org/glassfish/jersey/examples/sseitemstore/jaxrs/JaxrsItemStoreResource.java
@@ -0,0 +1,227 @@
+/*
+ * Copyright (c) 2017, 2018 Oracle and/or its affiliates. All rights reserved.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Distribution License v. 1.0, which is available at
+ * http://www.eclipse.org/org/documents/edl-v10.php.
+ *
+ * SPDX-License-Identifier: BSD-3-Clause
+ */
+
+package org.glassfish.jersey.examples.sseitemstore.jaxrs;
+
+import java.util.LinkedList;
+import java.util.ListIterator;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+import java.util.logging.Logger;
+
+import javax.ws.rs.BadRequestException;
+import javax.ws.rs.DefaultValue;
+import javax.ws.rs.FormParam;
+import javax.ws.rs.GET;
+import javax.ws.rs.HeaderParam;
+import javax.ws.rs.InternalServerErrorException;
+import javax.ws.rs.POST;
+import javax.ws.rs.Path;
+import javax.ws.rs.Produces;
+import javax.ws.rs.ServiceUnavailableException;
+import javax.ws.rs.core.Context;
+import javax.ws.rs.core.MediaType;
+import javax.ws.rs.sse.OutboundSseEvent;
+import javax.ws.rs.sse.Sse;
+import javax.ws.rs.sse.SseBroadcaster;
+import javax.ws.rs.sse.SseEventSink;
+
+import org.glassfish.jersey.media.sse.SseFeature;
+
+
+/**
+ * A resource for storing named items.
+ *
+ * @author Marek Potociar (marek.potociar at oracle.com)
+ */
+@Path("items")
+public class JaxrsItemStoreResource {
+ private static final Logger LOGGER = Logger.getLogger(JaxrsItemStoreResource.class.getName());
+
+ private static final ReentrantReadWriteLock storeLock = new ReentrantReadWriteLock();
+ private static final LinkedList<String> itemStore = new LinkedList<>();
+
+ private static final AtomicReference<SseBroadcaster> BROADCASTER = new AtomicReference<>(null);
+
+ @Context
+ private Sse sse;
+
+ private static volatile long reconnectDelay = 0;
+
+ /**
+ * List all stored items.
+ *
+ * @return list of all stored items.
+ */
+ @GET
+ @Produces(MediaType.TEXT_PLAIN)
+ public String listItems() {
+ try {
+ storeLock.readLock().lock();
+ return itemStore.toString();
+ } finally {
+ storeLock.readLock().unlock();
+ }
+ }
+
+ /**
+ * Receive & process commands sent by the test client that control the internal resource state.
+ * <p>
+ * Following is the list of recognized commands:
+ * <ul>
+ * <li><b>disconnect</b> - disconnect all registered event streams.</li>
+ * <li><b>reconnect now</b> - enable client reconnecting.</li>
+ * <li><b>reconnect <seconds></b> - disable client reconnecting.
+ * Reconnecting clients will receive a HTTP 503 response with
+ * {@value javax.ws.rs.core.HttpHeaders#RETRY_AFTER} set to the amount of
+ * milliseconds specified.</li>
+ * </ul>
+ *
+ * @param command command to be processed.
+ * @return message about processing result.
+ * @throws BadRequestException in case the command is not recognized or not specified.
+ */
+ @POST
+ @Path("commands")
+ public String processCommand(String command) {
+ if (command == null || command.isEmpty()) {
+ throw new BadRequestException("No command specified.");
+ }
+
+ if ("disconnect".equals(command)) {
+ closeBroadcaster();
+ return "Disconnected.";
+ } else if ("reconnect ".length() < command.length() && command.startsWith("reconnect ")) {
+ final String when = command.substring("reconnect ".length());
+ try {
+ reconnectDelay = "now".equals(when) ? 0 : Long.parseLong(when);
+ return "Reconnect strategy updated: " + when;
+ } catch (NumberFormatException ignore) {
+ // ignored
+ }
+ }
+
+ throw new BadRequestException("Command not recognized: '" + command + "'");
+ }
+
+ /**
+ * TODO - rewrite Connect or re-connect to SSE event stream.
+ *
+ * @param lastEventId Value of custom SSE HTTP <tt>{@value SseFeature#LAST_EVENT_ID_HEADER}</tt> header.
+ * Defaults to {@code -1} if not set.
+ * @throws InternalServerErrorException in case replaying missed events to the reconnected output stream fails.
+ * @throws ServiceUnavailableException in case the reconnect delay is set to a positive value.
+ */
+ @GET
+ @Path("events")
+ @Produces(MediaType.SERVER_SENT_EVENTS)
+ public void itemEvents(@HeaderParam(SseFeature.LAST_EVENT_ID_HEADER) @DefaultValue("-1") int lastEventId,
+ @Context SseEventSink eventSink) {
+
+ if (lastEventId >= 0) {
+ LOGGER.info("Received last event id :" + lastEventId);
+
+ // decide the reconnect handling strategy based on current reconnect delay value.
+ final long delay = reconnectDelay;
+ if (0 < delay) {
+ LOGGER.info("Non-zero reconnect delay [" + delay + "] - responding with HTTP 503.");
+ throw new ServiceUnavailableException(delay);
+ } else {
+ LOGGER.info("Zero reconnect delay - reconnecting.");
+ replayMissedEvents(lastEventId, eventSink);
+ }
+ }
+
+ getBroadcaster().register(eventSink);
+ }
+
+ private void replayMissedEvents(int lastEventId, SseEventSink eventSink) {
+ try {
+ storeLock.readLock().lock();
+ final int firstUnreceived = lastEventId + 1;
+ final int missingCount = itemStore.size() - firstUnreceived;
+ if (missingCount > 0) {
+ LOGGER.info("Replaying events - starting with id " + firstUnreceived);
+ final ListIterator<String> it = itemStore.subList(firstUnreceived, itemStore.size()).listIterator();
+ while (it.hasNext()) {
+ eventSink.send(createItemEvent(it.nextIndex() + firstUnreceived, it.next()));
+ }
+ } else {
+ LOGGER.info("No events to replay.");
+ }
+ } finally {
+ storeLock.readLock().unlock();
+ }
+ }
+
+ /**
+ * Add new item to the item store.
+ * <p>
+ * Invoking this method will fire 2 new SSE events - 1st about newly added item and 2nd about the new item store size.
+ *
+ * @param name item name.
+ */
+ @POST
+ public void addItem(@FormParam("name") String name) {
+ // Ignore if the request was sent without name parameter.
+ if (name == null) {
+ return;
+ }
+
+ final int eventId;
+ try {
+ storeLock.writeLock().lock();
+ eventId = itemStore.size();
+ itemStore.add(name);
+
+ SseBroadcaster sseBroadcaster = getBroadcaster();
+
+ // Broadcasting an un-named event with the name of the newly added item in data
+ sseBroadcaster.broadcast(createItemEvent(eventId, name));
+
+ // Broadcasting a named "size" event with the current size of the items collection in data
+ final OutboundSseEvent event = sse.newEventBuilder().name("size").data(Integer.class, eventId + 1).build();
+ sseBroadcaster.broadcast(event);
+ } finally {
+ storeLock.writeLock().unlock();
+ }
+ }
+
+ private OutboundSseEvent createItemEvent(final int eventId, final String name) {
+ Logger.getLogger(JaxrsItemStoreResource.class.getName())
+ .info("Creating event id [" + eventId + "] name [" + name + "]");
+ return sse.newEventBuilder().id("" + eventId).data(String.class, name).build();
+ }
+
+ /**
+ * Get stored broadcaster or create a new one and store it.
+ *
+ * @return broadcaster instance.
+ */
+ private SseBroadcaster getBroadcaster() {
+ SseBroadcaster sseBroadcaster = BROADCASTER.get();
+ if (sseBroadcaster == null) {
+ BROADCASTER.compareAndSet(null, sse.newBroadcaster());
+ }
+
+ return BROADCASTER.get();
+ }
+
+ /**
+ * Close currently stored broadcaster.
+ */
+ private void closeBroadcaster() {
+ SseBroadcaster sseBroadcaster = BROADCASTER.getAndSet(null);
+ if (sseBroadcaster == null) {
+ return;
+ }
+ sseBroadcaster.close();
+ }
+}
diff --git a/examples/sse-item-store-jaxrs-webapp/src/main/webapp/css/main.css b/examples/sse-item-store-jaxrs-webapp/src/main/webapp/css/main.css
new file mode 100644
index 0000000..5e31320
--- /dev/null
+++ b/examples/sse-item-store-jaxrs-webapp/src/main/webapp/css/main.css
@@ -0,0 +1,36 @@
+/*
+ * Copyright (c) 2017, 2018 Oracle and/or its affiliates. All rights reserved.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Distribution License v. 1.0, which is available at
+ * http://www.eclipse.org/org/documents/edl-v10.php.
+ *
+ * SPDX-License-Identifier: BSD-3-Clause
+ */
+
+div.message {
+ border-radius: 10px;
+ border: thin solid #444444;
+ margin: 10px;
+ width: 300px;
+ padding: 10px;
+ box-shadow: 5px 5px 5px #888888;
+ font-family: Helvetica, serif;
+ color: #333333;
+}
+
+div.items {
+ border-radius: 10px;
+ border: thin dotted #AAAAFF;
+ margin: 10px;
+ padding: 10px;
+ font-family: Helvetica, serif;
+ color: #4444FF;
+}
+
+div.messags {
+ border-radius: 10px;
+ border: thin dotted #888888;
+ margin: 10px;
+ padding: 10px;
+}
diff --git a/examples/sse-item-store-jaxrs-webapp/src/main/webapp/index.html b/examples/sse-item-store-jaxrs-webapp/src/main/webapp/index.html
new file mode 100644
index 0000000..0c488a2
--- /dev/null
+++ b/examples/sse-item-store-jaxrs-webapp/src/main/webapp/index.html
@@ -0,0 +1,27 @@
+<!--
+
+ Copyright (c) 2017, 2018 Oracle and/or its affiliates. All rights reserved.
+
+ This program and the accompanying materials are made available under the
+ terms of the Eclipse Distribution License v. 1.0, which is available at
+ http://www.eclipse.org/org/documents/edl-v10.php.
+
+ SPDX-License-Identifier: BSD-3-Clause
+
+-->
+
+<!DOCTYPE HTML PUBLIC "-//W3C//DTD HTML 4.01 Transitional//EN" "http://www.w3.org/TR/html4/loose.dtd">
+<html>
+<head>
+ <title>Jersey SSE Item Store Example</title>
+ <link rel="stylesheet" href="css/main.css"/>
+ <script src="js/engine.js" type="text/javascript"></script>
+</head>
+<body>
+<div>
+ Enter item: <input id="name" type="text"/><input value="Add" type="button" onclick="addItem()"/>
+</div>
+<div id="items" class="items"></div>
+<div id="messages" class="messages"></div>
+</body>
+</html>
diff --git a/examples/sse-item-store-jaxrs-webapp/src/main/webapp/js/engine.js b/examples/sse-item-store-jaxrs-webapp/src/main/webapp/js/engine.js
new file mode 100644
index 0000000..3b1f42d
--- /dev/null
+++ b/examples/sse-item-store-jaxrs-webapp/src/main/webapp/js/engine.js
@@ -0,0 +1,82 @@
+/*
+ * Copyright (c) 2017, 2018 Oracle and/or its affiliates. All rights reserved.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Distribution License v. 1.0, which is available at
+ * http://www.eclipse.org/org/documents/edl-v10.php.
+ *
+ * SPDX-License-Identifier: BSD-3-Clause
+ */
+
+"use strict";
+
+function addItem() {
+ var itemInput = document.getElementById("name");
+
+ var req = new XMLHttpRequest();
+ req.open("POST", "resources/items", true);
+ req.setRequestHeader("Content-Type", "application/x-www-form-urlencoded");
+ req.onreadystatechange = function () {
+ if (req.readyState == 4 && req.status == 204) {
+ //Call a function when the state changes.
+ itemInput.value = "";
+ getItems();
+ }
+ };
+ req.send("name=" + itemInput.value);
+}
+
+function getItems() {
+ var req = new XMLHttpRequest();
+ req.open("GET", "resources/items", true);
+ req.setRequestHeader("Accept", "text/plain");
+ req.onreadystatechange = function () {
+ //Call a function when the state changes.
+ if (req.readyState == 4 && req.status == 200) {
+ document.getElementById("items").innerHTML = req.responseText;
+ }
+ };
+ req.send();
+}
+
+function display(data, rgb) {
+ var msgSpan = document.createElement("span");
+ msgSpan.style.color = rgb;
+ msgSpan.innerHTML = data;
+ var msgDiv = document.createElement("div");
+ msgDiv.className = "message";
+ msgDiv.appendChild(msgSpan);
+
+ var messages = document.getElementById("messages");
+ messages.insertBefore(msgDiv, messages.firstChild);
+}
+
+function receiveMessages() {
+ if (typeof(EventSource) !== "undefined") {
+ // Yes! Server-sent events support!
+ var source = new EventSource("resources/items/events");
+ source.onmessage = function (event) {
+ console.log('Received unnamed event: ' + event.data);
+ display("Added new item: " + event.data, "#444444");
+ };
+
+ source.addEventListener("size", function(e) {
+ console.log('Received event ' + event.name + ': ' + event.data);
+ display("New items size: " + event.data, "#0000FF");
+ }, false);
+
+ source.onopen = function (event) {
+ console.log("event source opened");
+ };
+
+ source.onerror = function (event) {
+ console.log('Received error event: ' + event.data);
+ display(event.data, "#FF0000");
+ };
+ } else {
+ // Sorry! No server-sent events support..
+ display('SSE not supported by browser.', "#FF0000");
+ }
+}
+
+window.onload = receiveMessages;
diff --git a/examples/sse-item-store-jaxrs-webapp/src/test/java/org/glassfish/jersey/examples/sseitemstore/jaxrs/JaxrsItemStoreResourceTest.java b/examples/sse-item-store-jaxrs-webapp/src/test/java/org/glassfish/jersey/examples/sseitemstore/jaxrs/JaxrsItemStoreResourceTest.java
new file mode 100644
index 0000000..0f352a6
--- /dev/null
+++ b/examples/sse-item-store-jaxrs-webapp/src/test/java/org/glassfish/jersey/examples/sseitemstore/jaxrs/JaxrsItemStoreResourceTest.java
@@ -0,0 +1,301 @@
+/*
+ * Copyright (c) 2017, 2018 Oracle and/or its affiliates. All rights reserved.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Distribution License v. 1.0, which is available at
+ * http://www.eclipse.org/org/documents/edl-v10.php.
+ *
+ * SPDX-License-Identifier: BSD-3-Clause
+ */
+
+package org.glassfish.jersey.examples.sseitemstore.jaxrs;
+
+import java.net.URI;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Queue;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
+import javax.ws.rs.client.Client;
+import javax.ws.rs.client.ClientBuilder;
+import javax.ws.rs.client.Entity;
+import javax.ws.rs.client.WebTarget;
+import javax.ws.rs.core.Application;
+import javax.ws.rs.core.Form;
+import javax.ws.rs.core.Response;
+import javax.ws.rs.core.UriBuilder;
+import javax.ws.rs.sse.SseEventSource;
+
+import org.glassfish.jersey.apache.connector.ApacheClientProperties;
+import org.glassfish.jersey.apache.connector.ApacheConnectorProvider;
+import org.glassfish.jersey.client.ClientConfig;
+import org.glassfish.jersey.client.ClientProperties;
+import org.glassfish.jersey.internal.guava.ThreadFactoryBuilder;
+import org.glassfish.jersey.test.JerseyTest;
+import org.glassfish.jersey.test.external.ExternalTestContainerFactory;
+
+import org.apache.http.impl.conn.PoolingHttpClientConnectionManager;
+import org.junit.Test;
+
+import static org.hamcrest.CoreMatchers.containsString;
+import static org.hamcrest.CoreMatchers.describedAs;
+import static org.hamcrest.CoreMatchers.equalTo;
+import static org.hamcrest.CoreMatchers.hasItems;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertThat;
+import static org.junit.Assert.assertTrue;
+
+/**
+ * Item store test.
+ *
+ * @author Marek Potociar (marek.potociar at oracle.com)
+ */
+public class JaxrsItemStoreResourceTest extends JerseyTest {
+
+ private static final int RECONNECT_DEFAULT = 500;
+ private static final Logger LOGGER = Logger.getLogger(JaxrsItemStoreResourceTest.class.getName());
+ private static final int MAX_LISTENERS = 5;
+ private static final int MAX_ITEMS = 10;
+
+ private final ExecutorService executorService =
+ Executors.newCachedThreadPool(new ThreadFactoryBuilder().setNameFormat("JaxrsItemStoreResourceTest-%d").build());
+ private final AtomicReference<Client> client = new AtomicReference<>(null);
+
+ @Override
+ protected Application configure() {
+ return new JaxrsItemStoreApp();
+ }
+
+ protected void configureClient(ClientConfig config) {
+ // using AHC as a test client connector to avoid issues with HttpUrlConnection socket management.
+ PoolingHttpClientConnectionManager cm = new PoolingHttpClientConnectionManager();
+
+ // adjusting max. connections just to be safe - the testEventSourceReconnect is quite greedy...
+ cm.setMaxTotal(MAX_LISTENERS * MAX_ITEMS);
+ cm.setDefaultMaxPerRoute(MAX_LISTENERS * MAX_ITEMS);
+
+ config.property(ApacheClientProperties.CONNECTION_MANAGER, cm)
+ .property(ClientProperties.READ_TIMEOUT, 2000)
+ .connectorProvider(new ApacheConnectorProvider());
+ }
+
+ @Override
+ protected Client getClient() {
+ if (client.get() == null) {
+ ClientConfig clientConfig = new ClientConfig();
+ configureClient(clientConfig);
+ client.compareAndSet(null,
+ ClientBuilder.newBuilder()
+ .withConfig(clientConfig)
+ .executorService(executorService).build());
+ }
+
+ return client.get();
+ }
+
+ @Override
+ public void tearDown() throws Exception {
+ super.tearDown();
+ client.get().close();
+ executorService.shutdown();
+ }
+
+ @Override
+ protected URI getBaseUri() {
+ final UriBuilder baseUriBuilder = UriBuilder.fromUri(super.getBaseUri()).path("sse-item-store-jaxrs-webapp");
+ final boolean externalFactoryInUse = getTestContainerFactory() instanceof ExternalTestContainerFactory;
+ return externalFactoryInUse ? baseUriBuilder.path("resources").build() : baseUriBuilder.build();
+ }
+
+ /**
+ * Test the item addition, addition event broadcasting and item retrieval from {@link JaxrsItemStoreResource}.
+ *
+ * @throws Exception in case of a test failure.
+ */
+ @Test
+ public void testItemsStore() throws Exception {
+ final List<String> items = Collections.unmodifiableList(Arrays.asList("foo", "bar", "baz"));
+ final WebTarget itemsTarget = target("items");
+ final CountDownLatch latch = new CountDownLatch(items.size() * MAX_LISTENERS * 2); // countdown on all events
+ final List<Queue<Integer>> indexQueues = new ArrayList<>(MAX_LISTENERS);
+ final SseEventSource[] sources = new SseEventSource[MAX_LISTENERS];
+ final AtomicInteger sizeEventsCount = new AtomicInteger(0);
+
+ for (int i = 0; i < MAX_LISTENERS; i++) {
+ final int id = i;
+ final SseEventSource es = SseEventSource.target(itemsTarget.path("events")).build();
+ sources[id] = es;
+
+ final Queue<Integer> indexes = new ConcurrentLinkedQueue<>();
+ indexQueues.add(indexes);
+
+ es.register(inboundEvent -> {
+ try {
+ if (null == inboundEvent.getName()) {
+ final String data = inboundEvent.readData();
+ LOGGER.info("[-i-] SOURCE " + id + ": Received event id=" + inboundEvent.getId() + " data=" + data);
+ indexes.add(items.indexOf(data));
+ } else if ("size".equals(inboundEvent.getName())) {
+ sizeEventsCount.incrementAndGet();
+ }
+ } catch (Exception ex) {
+ LOGGER.log(Level.SEVERE, "[-x-] SOURCE " + id + ": Error getting event data.", ex);
+ indexes.add(-999);
+ } finally {
+ latch.countDown();
+ }
+ });
+ }
+
+ try {
+ open(sources);
+ items.forEach((item) -> postItem(itemsTarget, item));
+
+ assertTrue("Waiting to receive all events has timed out.",
+ latch.await((1000 + MAX_LISTENERS * RECONNECT_DEFAULT) * getAsyncTimeoutMultiplier(),
+ TimeUnit.MILLISECONDS));
+
+ // need to force disconnect on server in order for EventSource.close(...) to succeed with HttpUrlConnection
+ sendCommand(itemsTarget, "disconnect");
+ } finally {
+ close(sources);
+ }
+
+ String postedItems = itemsTarget.request().get(String.class);
+ items.forEach((item) -> assertTrue("Item '" + item + "' not stored on server.", postedItems.contains(item)));
+
+ final AtomicInteger queueId = new AtomicInteger(0);
+ indexQueues.forEach((indexes) -> {
+ for (int i = 0; i < items.size(); i++) {
+ assertTrue("Event for '" + items.get(i) + "' not received in queue " + queueId.get(), indexes.contains(i));
+ }
+ assertEquals("Not received the expected number of events in queue " + queueId.get(), items.size(), indexes.size());
+ queueId.incrementAndGet();
+ });
+
+ assertEquals("Number of received 'size' events does not match.", items.size() * MAX_LISTENERS, sizeEventsCount.get());
+ }
+
+ /**
+ * Test the {@link SseEventSource} reconnect feature.
+ *
+ * @throws Exception in case of a test failure.
+ */
+ @Test
+ public void testEventSourceReconnect() throws Exception {
+ final WebTarget itemsTarget = target("items");
+ final CountDownLatch latch = new CountDownLatch(MAX_ITEMS * MAX_LISTENERS * 2); // countdown only on new item events
+ final List<Queue<String>> receivedQueues = new ArrayList<>(MAX_LISTENERS);
+ final SseEventSource[] sources = new SseEventSource[MAX_LISTENERS];
+
+ for (int i = 0; i < MAX_LISTENERS; i++) {
+ final int id = i;
+ final SseEventSource es = SseEventSource.target(itemsTarget.path("events"))
+ .reconnectingEvery(1, TimeUnit.MILLISECONDS).build();
+ sources[id] = es;
+
+ final Queue<String> received = new ConcurrentLinkedQueue<>();
+ receivedQueues.add(received);
+
+ es.register(inboundEvent -> {
+ try {
+ if (null == inboundEvent.getName()) {
+ final String data = inboundEvent.readData();
+ LOGGER.info("[-i-] SOURCE " + id + ": Received event id=" + inboundEvent.getId() + " data=" + data);
+ received.add(data);
+ latch.countDown();
+ }
+ } catch (Exception ex) {
+ LOGGER.log(Level.SEVERE, "[-x-] SOURCE " + id + ": Error getting event data.", ex);
+ received.add("[data processing error]");
+ }
+ });
+ }
+
+ final String[] postedItems = new String[MAX_ITEMS * 2];
+ try {
+ open(sources);
+
+ for (int i = 0; i < MAX_ITEMS; i++) {
+ final String item = String.format("round-1-%02d", i);
+ postItem(itemsTarget, item);
+ postedItems[i] = item;
+ sendCommand(itemsTarget, "disconnect");
+ Thread.sleep(200);
+ }
+
+ final int reconnectDelay = 1;
+ sendCommand(itemsTarget, "reconnect " + reconnectDelay);
+ sendCommand(itemsTarget, "disconnect");
+
+ Thread.sleep(reconnectDelay * 1000);
+
+ for (int i = 0; i < MAX_ITEMS; i++) {
+ final String item = String.format("round-2-%02d", i);
+ postedItems[i + MAX_ITEMS] = item;
+ postItem(itemsTarget, item);
+ }
+
+ sendCommand(itemsTarget, "reconnect now");
+
+ assertTrue("Waiting to receive all events has timed out.",
+ latch.await((1 + MAX_LISTENERS * (MAX_ITEMS + 1) * reconnectDelay) * getAsyncTimeoutMultiplier(),
+ TimeUnit.SECONDS));
+
+ // need to force disconnect on server in order for EventSource.close(...) to succeed with HttpUrlConnection
+ sendCommand(itemsTarget, "disconnect");
+ } finally {
+ close(sources);
+ }
+
+ final String storedItems = itemsTarget.request().get(String.class);
+ for (String item : postedItems) {
+ assertThat("Posted item '" + item + "' stored on server", storedItems, containsString(item));
+ }
+
+ int sourceId = 0;
+ for (Queue<String> queue : receivedQueues) {
+ assertThat("Received events in source " + sourceId, queue,
+ describedAs("Collection containing %0", hasItems(postedItems), Arrays.asList(postedItems).toString()));
+ assertThat("Size of received queue for source " + sourceId, queue.size(), equalTo(postedItems.length));
+ sourceId++;
+ }
+ }
+
+ private static void postItem(final WebTarget itemsTarget, final String item) {
+ final Response response = itemsTarget.request().post(Entity.form(new Form("name", item)));
+ assertEquals("Posting new item has failed.", 204, response.getStatus());
+ LOGGER.info("[-i-] POSTed item: '" + item + "'");
+ }
+
+ private static void open(final SseEventSource[] sources) {
+ Arrays.stream(sources).forEach(SseEventSource::open);
+ }
+
+ private static void close(final SseEventSource[] sources) {
+ int i = 0;
+ for (SseEventSource source : sources) {
+ if (source.isOpen()) {
+ assertTrue("Waiting to close a source has timed out.", source.close(1, TimeUnit.SECONDS));
+// source.close(100, TimeUnit.MILLISECONDS);
+ LOGGER.info("[<--] SOURCE " + i++ + " closed.");
+ }
+ }
+ }
+
+ private static void sendCommand(final WebTarget itemsTarget, final String command) {
+ final Response response = itemsTarget.path("commands").request().post(Entity.text(command));
+ assertEquals("'" + command + "' command has failed.", 200, response.getStatus());
+ LOGGER.info("[-!-] COMMAND '" + command + "' has been processed.");
+ }
+}
