InboundEvent.java
/*
* Copyright (c) 2012, 2021 Oracle and/or its affiliates. All rights reserved.
*
* This program and the accompanying materials are made available under the
* terms of the Eclipse Public License v. 2.0, which is available at
* http://www.eclipse.org/legal/epl-2.0.
*
* This Source Code may also be made available under the following Secondary
* Licenses when the conditions for such availability set forth in the
* Eclipse Public License v. 2.0 are satisfied: GNU General Public License,
* version 2 with the GNU Classpath Exception, which is available at
* https://www.gnu.org/software/classpath/license.html.
*
* SPDX-License-Identifier: EPL-2.0 OR GPL-2.0 WITH Classpath-exception-2.0
*/
package org.glassfish.jersey.media.sse;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.lang.annotation.Annotation;
import java.util.Arrays;
import javax.ws.rs.ProcessingException;
import javax.ws.rs.core.GenericType;
import javax.ws.rs.core.MediaType;
import javax.ws.rs.core.MultivaluedMap;
import javax.ws.rs.ext.MessageBodyReader;
import javax.ws.rs.sse.InboundSseEvent;
import org.glassfish.jersey.message.MessageBodyWorkers;
import org.glassfish.jersey.message.internal.MessageBodyProviderNotFoundException;
/**
* Inbound event.
*
* @author Pavel Bucek
* @author Marek Potociar
*/
public class InboundEvent implements InboundSseEvent {
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;
private final MessageBodyWorkers messageBodyWorkers;
private final Annotation[] annotations;
private final MediaType mediaType;
private final MultivaluedMap<String, String> headers;
/**
* 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;
private final ByteArrayOutputStream dataStream;
private final MessageBodyWorkers workers;
private final Annotation[] annotations;
private final MediaType mediaType;
private final MultivaluedMap<String, String> headers;
private final StringBuilder commentBuilder;
/**
* Create new inbound event builder.
*
* @param workers configured client-side {@link MessageBodyWorkers entity providers} used for
* {@link javax.ws.rs.ext.MessageBodyReader} lookup.
* @param annotations annotations attached to the Java type to be read. Used for
* {@link javax.ws.rs.ext.MessageBodyReader} lookup.
* @param mediaType media type of the SSE event data.
* Used for {@link javax.ws.rs.ext.MessageBodyReader} lookup.
* @param headers response headers. Used for {@link javax.ws.rs.ext.MessageBodyWriter} lookup.
*/
public Builder(MessageBodyWorkers workers,
Annotation[] annotations,
MediaType mediaType,
MultivaluedMap<String, String> headers) {
this.workers = workers;
this.annotations = annotations;
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.
* @return updated builder instance.
*/
public Builder name(String name) {
this.name = name;
return this;
}
/**
* Set inbound event identifier.
* <p/>
* Value of the received SSE {@code "id"} field.
*
* @param id {@code "id"} field value.
* @return updated builder instance.
*/
public Builder id(String id) {
this.id = 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
*/
public Builder reconnectDelay(long milliseconds) {
if (milliseconds < 0) {
milliseconds = SseFeature.RECONNECT_NOT_SET;
}
this.reconnectDelay = milliseconds;
return this;
}
/**
* Add more inbound event data.
*
* @param data byte array containing data stored in the incoming event.
* @return updated builder instance.
*/
public Builder write(byte[] data) {
if (data == null || data.length == 0) {
return this;
}
try {
this.dataStream.write(data);
} catch (IOException ex) {
// ignore - this is not possible with ByteArrayOutputStream
}
return this;
}
/**
* Build a new inbound event instance using the supplied data.
*
* @return new inbound event instance.
*/
public InboundEvent build() {
return new InboundEvent(
name,
id,
commentBuilder.length() > 0 ? commentBuilder.substring(0, commentBuilder.length() - 1) : null,
reconnectDelay,
dataStream.toByteArray(),
workers,
annotations,
mediaType,
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 = stripLastLineBreak(data);
this.messageBodyWorkers = messageBodyWorkers;
this.annotations = annotations;
this.mediaType = mediaType;
this.headers = headers;
}
/**
* Get event name.
* <p>
* Contains value of SSE {@code "event"} field. This field is optional. Method may return {@code null}, if the event
* name is not specified.
* </p>
*
* @return event name, or {@code null} if not set.
*/
public String getName() {
return name;
}
/**
* Get event identifier.
* <p>
* Contains value of SSE {@code "id"} field. This field is optional. Method may return {@code null}, if the event
* identifier is not specified.
* </p>
*
* @return event id.
* @since 2.3
*/
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.
* <p>
* Contains value of SSE {@code "retry"} field. This field is optional. Method returns {@link SseFeature#RECONNECT_NOT_SET}
* if no value has been set.
* </p>
*
* @return reconnection delay in milliseconds or {@link SseFeature#RECONNECT_NOT_SET} if no value has been set.
* @since 2.3
*/
public long getReconnectDelay() {
return reconnectDelay;
}
/**
* Check if the connection retry time has been set in the event.
*
* @return {@code true} if new reconnection delay has been set in the event, {@code false} otherwise.
* @since 2.3
*/
public boolean isReconnectDelaySet() {
return reconnectDelay > SseFeature.RECONNECT_NOT_SET;
}
/**
* Check if the event is empty (i.e. does not contain any data).
*
* @return {@code true} if current instance does not contain any data, {@code false} otherwise.
*/
public boolean isEmpty() {
return data.length == 0;
}
/**
* 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.
* @since 2.3
*/
public String readData() {
return readData(STRING_AS_GENERIC_TYPE, null);
}
/**
* Read event data as a given Java type.
*
* @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.
* @since 2.3
*/
public <T> T readData(Class<T> type) {
return readData(new GenericType<T>(type), null);
}
/**
* Read event data as a given generic 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.
* @since 2.3
*/
@SuppressWarnings("unused")
public <T> T readData(GenericType<T> type) {
return readData(type, null);
}
/**
* Read event data as a given Java 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.
* @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.
* @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.
* @since 2.3
*/
public <T> T readData(GenericType<T> type, MediaType mediaType) {
final MediaType effectiveMediaType = mediaType == null ? this.mediaType : mediaType;
final MessageBodyReader reader =
messageBodyWorkers.getMessageBodyReader(type.getRawType(), type.getType(), annotations, mediaType);
if (reader == null) {
throw new MessageBodyProviderNotFoundException(LocalizationMessages.EVENT_DATA_READER_NOT_FOUND());
}
return readAndCast(type, effectiveMediaType, reader);
}
@SuppressWarnings("unchecked")
private <T> T readAndCast(GenericType<T> type, MediaType effectiveMediaType, MessageBodyReader reader) {
try {
return (T) reader.readFrom(
type.getRawType(),
type.getType(),
annotations,
effectiveMediaType,
headers,
new ByteArrayInputStream(data));
} catch (IOException ex) {
throw new ProcessingException(ex);
}
}
/**
* 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.
*/
@SuppressWarnings("unused")
public byte[] getRawData() {
if (isEmpty()) {
return data;
}
return Arrays.copyOf(data, data.length);
}
@Override
public String toString() {
String s;
try {
s = readData();
} catch (ProcessingException e) {
s = "<Error reading data into a string>";
}
return "InboundEvent{"
+ "name='" + name + '\''
+ ", id='" + id + '\''
+ ", comment=" + (comment == null ? "[no comments]" : '\'' + comment + '\'')
+ ", data=" + s
+ '}';
}
/**
* String last line break from data. (Last line-break should not be considered as part of received data).
*
* @param data data
* @return updated byte array.
*/
private static byte[] stripLastLineBreak(final byte[] data) {
if (data.length > 0 && data[data.length - 1] == '\n') {
return Arrays.copyOf(data, data.length - 1);
}
return data;
}
}