You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cxf.apache.org by re...@apache.org on 2021/05/20 22:34:37 UTC
[cxf] branch 3.4.x-fixes updated: CXF-7996 Jakarta EE TCKs and
compatibility. Address SSE failures,
missed 503 & Retry-After handling (#798)
This is an automated email from the ASF dual-hosted git repository.
reta pushed a commit to branch 3.4.x-fixes
in repository https://gitbox.apache.org/repos/asf/cxf.git
The following commit(s) were added to refs/heads/3.4.x-fixes by this push:
new 6fb764d CXF-7996 Jakarta EE TCKs and compatibility. Address SSE failures, missed 503 & Retry-After handling (#798)
6fb764d is described below
commit 6fb764dfe71296fb9705ec55cb130e6b538fbcab
Author: Andriy Redko <dr...@gmail.com>
AuthorDate: Thu May 20 17:17:58 2021 -0400
CXF-7996 Jakarta EE TCKs and compatibility. Address SSE failures, missed 503 & Retry-After handling (#798)
(cherry picked from commit 1bb0e460cfd2e76fef7471fcc29134bb5033c64f)
---
.../cxf/jaxrs/impl/RetryAfterHeaderProvider.java | 123 +++++++++++++++++++++
.../jaxrs/impl/RetryAfterHeaderProviderTest.java | 75 +++++++++++++
.../cxf/jaxrs/sse/client/SseEventSourceImpl.java | 24 ++++
.../jaxrs/sse/client/SseEventSourceImplTest.java | 76 ++++++++++++-
4 files changed, 297 insertions(+), 1 deletion(-)
diff --git a/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/impl/RetryAfterHeaderProvider.java b/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/impl/RetryAfterHeaderProvider.java
new file mode 100644
index 0000000..f4c97ae
--- /dev/null
+++ b/rt/frontend/jaxrs/src/main/java/org/apache/cxf/jaxrs/impl/RetryAfterHeaderProvider.java
@@ -0,0 +1,123 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.cxf.jaxrs.impl;
+
+import java.text.DateFormat;
+import java.text.ParseException;
+import java.text.SimpleDateFormat;
+import java.time.Clock;
+import java.time.Duration;
+import java.util.Date;
+import java.util.Locale;
+import java.util.logging.Logger;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+import javax.ws.rs.ext.RuntimeDelegate.HeaderDelegate;
+
+import org.apache.cxf.common.logging.LogUtils;
+
+/**
+ * Retry-After = "Retry-After" ":" ( HTTP-date | delta-seconds )
+ * HTTP-date = IMF-fixdate / obs-date
+ *
+ * An example of the preferred format is
+ * Sun, 06 Nov 1994 08:49:37 GMT ; IMF-fixdate (RFC 822)
+ *
+ * Examples of the two obsolete formats are
+ * Sunday, 06-Nov-94 08:49:37 GMT ; obsolete RFC 850 format
+ * Sun Nov 6 08:49:37 1994 ; ANSI C's asctime() format
+ *
+ */
+public class RetryAfterHeaderProvider implements HeaderDelegate<Duration> {
+ private static final Logger LOG = LogUtils.getL7dLogger(RetryAfterHeaderProvider.class);
+ private static final Pattern DELTA_SECONDS_PATTERN = Pattern.compile("^(\\d+)$");
+ private static final Pattern RFC_PATTERN = Pattern.compile("^\\s*([^,\\s]+)\\s*[,].+$");
+ private final Clock clock;
+
+ public RetryAfterHeaderProvider() {
+ this(Clock.systemUTC());
+ }
+
+ public RetryAfterHeaderProvider(Clock clock) {
+ this.clock = clock;
+ }
+
+ @Override
+ public Duration fromString(String value) {
+ if (value == null) {
+ return Duration.ZERO;
+ }
+
+ final Matcher matcher = DELTA_SECONDS_PATTERN.matcher(value);
+ if (matcher.matches()) {
+ return Duration.ofSeconds(Long.parseLong(matcher.group(0)));
+ } else {
+ final DateFormat formatter = tryFormatter(value);
+ if (formatter != null) {
+ try {
+ final Date date = formatter.parse(value);
+ final long retryAfter = date.getTime() - clock.millis();
+ if (retryAfter > 0) {
+ return Duration.ofMillis(retryAfter);
+ }
+ } catch (ParseException ex) {
+ LOG.fine("The format of '" + value + "' is not recognizable date: " + ex);
+ }
+ }
+ }
+
+ return Duration.ZERO;
+ }
+
+ private DateFormat tryFormatter(String value) {
+ final Matcher matcher = RFC_PATTERN.matcher(value);
+ if (matcher.matches()) {
+ final String dayOfWeek = matcher.group(1);
+ // RFC-1123 updates RFC-822 changing the year from two digits to four
+ if (dayOfWeek.length() == 3) {
+ return new SimpleDateFormat("EEE, dd MMM yyyy HH:mm:ss z", Locale.US);
+ } else {
+ // Obsolete RFC-850 format
+ return new SimpleDateFormat("EEEE, dd-MMM-yy HH:mm:ss z", Locale.US);
+ }
+ } else {
+ // ANSI C's asctime() format
+ return new SimpleDateFormat("EEE MMM d HH:mm:ss yyyy", Locale.US);
+ }
+ }
+
+ @Override
+ public String toString(Duration value) {
+ if (value == null) {
+ return null;
+ } else {
+ return Long.toString(value.getSeconds());
+ }
+ }
+
+ public static Duration valueOf(String value) {
+ return new RetryAfterHeaderProvider().fromString(value);
+ }
+
+ public static Duration valueOf(Clock clock, String value) {
+ return new RetryAfterHeaderProvider(clock).fromString(value);
+ }
+}
diff --git a/rt/frontend/jaxrs/src/test/java/org/apache/cxf/jaxrs/impl/RetryAfterHeaderProviderTest.java b/rt/frontend/jaxrs/src/test/java/org/apache/cxf/jaxrs/impl/RetryAfterHeaderProviderTest.java
new file mode 100644
index 0000000..df9d155
--- /dev/null
+++ b/rt/frontend/jaxrs/src/test/java/org/apache/cxf/jaxrs/impl/RetryAfterHeaderProviderTest.java
@@ -0,0 +1,75 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.cxf.jaxrs.impl;
+
+import java.time.Clock;
+import java.time.Duration;
+import java.time.OffsetDateTime;
+import java.time.ZoneId;
+import java.time.ZoneOffset;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.TimeZone;
+
+import org.junit.Assert;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+import static org.hamcrest.CoreMatchers.is;
+
+@RunWith(Parameterized.class)
+public class RetryAfterHeaderProviderTest {
+ private final String date;
+ private final Duration delay;
+ private final Clock clock;
+
+ public RetryAfterHeaderProviderTest(String date, Duration delay) {
+ this.date = date;
+ this.delay = delay;
+ this.clock = Clock.fixed(
+ OffsetDateTime.of(2000, 01, 01, 0, 0, 0, 0, ZoneOffset.UTC).toInstant(),
+ ZoneId.of("Z"));
+ }
+
+ @Parameterized.Parameters(name = "retry after {0} is duration of {1} from 2000-01-01 00:00:00.000")
+ public static Collection<Object[]> data() {
+ final int offset = TimeZone.getDefault().getRawOffset() / 1000;
+
+ return Arrays.asList(
+ new Object[] {"", Duration.ZERO},
+ new Object[] {null, Duration.ZERO},
+ new Object[] {"120", Duration.ofSeconds(120)},
+ new Object[] {"-120", Duration.ZERO},
+ new Object[] {"Sun, 03 Nov 2002 08:49:37 EST", Duration.ofSeconds(89646577)},
+ new Object[] {"Sat, 01 Jan 2000 01:00:00 GMT", Duration.ofSeconds(3600)},
+ new Object[] {"Sunday, 03-Nov-02 08:49:37 GMT", Duration.ofSeconds(89628577)},
+ new Object[] {"Sun Nov 3 08:49:37 2002", Duration.ofSeconds(89628577 - offset)},
+ new Object[] {"Sun Nov 03 08:49:37 2002", Duration.ofSeconds(89628577 - offset)},
+ new Object[] {"Sun, 06 Nov 1994 08:49:37 EST", Duration.ZERO}
+ );
+
+ }
+
+ @Test
+ public void test() {
+ Assert.assertThat(RetryAfterHeaderProvider.valueOf(clock, date), is(delay));
+ }
+}
diff --git a/rt/rs/sse/src/main/java/org/apache/cxf/jaxrs/sse/client/SseEventSourceImpl.java b/rt/rs/sse/src/main/java/org/apache/cxf/jaxrs/sse/client/SseEventSourceImpl.java
index c823e3c..7bda05c 100644
--- a/rt/rs/sse/src/main/java/org/apache/cxf/jaxrs/sse/client/SseEventSourceImpl.java
+++ b/rt/rs/sse/src/main/java/org/apache/cxf/jaxrs/sse/client/SseEventSourceImpl.java
@@ -20,6 +20,7 @@ package org.apache.cxf.jaxrs.sse.client;
import java.security.AccessController;
import java.security.PrivilegedAction;
+import java.time.Duration;
import java.util.Collection;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.Executors;
@@ -41,6 +42,7 @@ import javax.ws.rs.sse.SseEventSource;
import org.apache.cxf.common.logging.LogUtils;
import org.apache.cxf.endpoint.Endpoint;
import org.apache.cxf.jaxrs.client.WebClient;
+import org.apache.cxf.jaxrs.impl.RetryAfterHeaderProvider;
import org.apache.cxf.jaxrs.utils.ExceptionUtils;
/**
@@ -204,6 +206,28 @@ public class SseEventSourceImpl implements SseEventSource {
response.close();
return;
}
+
+ // In addition to handling the standard connection loss failures, JAX-RS SseEventSource automatically
+ // deals with any HTTP 503 Service Unavailable responses from an SSE endpoint, that contain a "Retry-After"
+ // HTTP header with a valid value. The HTTP 503 + "Retry-After" technique is often used by HTTP endpoints
+ // as a means of connection and traffic throttling. In case a HTTP 503 + "Retry-After" response is received
+ // in return to a connection request, JAX-RS SSE event source will automatically schedule a new reconnect
+ // attempt and use the received "Retry-After" HTTP header value as a one-time override of the reconnect
+ // delay.
+ if (status == 503) {
+ final Object retryAfterHeader = response.getHeaders().getFirst(HttpHeaders.RETRY_AFTER);
+ if (retryAfterHeader != null) {
+ final Duration retryAfter = RetryAfterHeaderProvider.valueOf(retryAfterHeader.toString());
+ final long retryAfterMillis = retryAfter.toMillis();
+ if (retryAfterMillis >= 0) {
+ LOG.fine("SSE endpoint " + target.getUri() + " is unavailable, retrying after "
+ + retryAfterMillis + "ms");
+ scheduleReconnect(retryAfterMillis, TimeUnit.MILLISECONDS, lastEventId);
+ response.close();
+ return;
+ }
+ }
+ }
// Convert unsuccessful responses to instances of WebApplicationException
if (status != 304 && status >= 300) {
diff --git a/rt/rs/sse/src/test/java/org/apache/cxf/jaxrs/sse/client/SseEventSourceImplTest.java b/rt/rs/sse/src/test/java/org/apache/cxf/jaxrs/sse/client/SseEventSourceImplTest.java
index 00e44a7..0bef971 100644
--- a/rt/rs/sse/src/test/java/org/apache/cxf/jaxrs/sse/client/SseEventSourceImplTest.java
+++ b/rt/rs/sse/src/test/java/org/apache/cxf/jaxrs/sse/client/SseEventSourceImplTest.java
@@ -64,7 +64,7 @@ import static org.hamcrest.MatcherAssert.assertThat;
public class SseEventSourceImplTest {
enum Type {
- NO_CONTENT, NO_SERVER, BUSY,
+ NO_CONTENT, NO_SERVER, BUSY, UNAVAILABLE, RETRY_AFTER,
EVENT, EVENT_JUST_DATA, EVENT_JUST_NAME, EVENT_MULTILINE_DATA, EVENT_NO_RETRY, EVENT_BAD_RETRY, EVENT_MIXED,
EVENT_BAD_NEW_LINES, EVENT_NOT_AUTHORIZED, EVENT_LAST_EVENT_ID, EVENT_RETRY_LAST_EVENT_ID;
}
@@ -131,6 +131,34 @@ public class SseEventSourceImplTest {
assertThat(events.size(), equalTo(0));
}
}
+
+ @Test
+ public void testNoReconnectWhenUnavailableIsReturned() {
+ try (SseEventSource eventSource = withNoReconnect(Type.UNAVAILABLE)) {
+ eventSource.open();
+ assertThat(eventSource.isOpen(), equalTo(false));
+
+ await()
+ .during(Duration.ofMillis(3000L))
+ .untilAsserted(() -> assertThat(eventSource.isOpen(), equalTo(false)));
+
+ assertThat(events.size(), equalTo(0));
+ }
+ }
+
+ @Test
+ public void testNoReconnectWhenRetryAfterIsReturned() {
+ try (SseEventSource eventSource = withNoReconnect(Type.RETRY_AFTER)) {
+ eventSource.open();
+ assertThat(eventSource.isOpen(), equalTo(false));
+
+ await()
+ .atMost(Duration.ofMillis(3000L))
+ .untilAsserted(() -> assertThat(eventSource.isOpen(), equalTo(true)));
+
+ assertThat(events.size(), equalTo(1));
+ }
+ }
@Test
public void testReuseSameEventSourceSeveralTimes() {
@@ -486,6 +514,8 @@ public class SseEventSourceImplTest {
startBusyServer(Type.BUSY);
startNotAuthorizedServer(Type.EVENT_NOT_AUTHORIZED);
+ startUnavailableServer(Type.UNAVAILABLE);
+ startUnavailableServer(Type.RETRY_AFTER);
startServer(Type.EVENT, EVENT);
startServer(Type.EVENT_JUST_DATA, EVENT_JUST_DATA);
@@ -522,7 +552,15 @@ public class SseEventSourceImplTest {
sf.setServiceBean(new BusyEventServer());
SERVERS.put(type, sf.create());
}
+
+ private static void startUnavailableServer(Type type) {
+ JAXRSServerFactoryBean sf = new JAXRSServerFactoryBean();
+ sf.setAddress(LOCAL_ADDRESS + type.name());
+ sf.setServiceBean(new StatusServer(503, type));
+ SERVERS.put(type, sf.create());
+ }
+
private static void startServer(Type type, String payload) {
JAXRSServerFactoryBean sf = new JAXRSServerFactoryBean();
sf.setAddress(LOCAL_ADDRESS + type.name());
@@ -544,6 +582,42 @@ public class SseEventSourceImplTest {
server.destroy();
}
}
+
+ public static class StatusServer {
+ private final int status;
+ private final Type type;
+ private volatile boolean triggered;
+
+ public StatusServer(int status, Type type) {
+ this.status = status;
+ this.type = type;
+ }
+
+ @GET
+ @Produces(MediaType.SERVER_SENT_EVENTS)
+ public Response event() {
+ try {
+ if (triggered) {
+ return Response.ok(EVENT).build();
+ } else if (status == 503) {
+ if (type == Type.RETRY_AFTER) {
+ return Response
+ .status(status)
+ .header(HttpHeaders.RETRY_AFTER, "2")
+ .build();
+ } else {
+ return Response
+ .status(status)
+ .build();
+ }
+ } else {
+ return Response.status(status).build();
+ }
+ } finally {
+ triggered = true;
+ }
+ }
+ }
public static class EventServer {
private final String payload;