Skip to content

Commit ee81f44

Browse files
chillaqchillaq
committed
Added immediate socket close option
1 parent 34fc67f commit ee81f44

File tree

12 files changed

+20
-73
lines changed

12 files changed

+20
-73
lines changed

‎client/pom.xml‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -5,9 +5,9 @@
55
<parent>
66
<groupId>io.split.client</groupId>
77
<artifactId>java-client-parent</artifactId>
8-
<version>4.18.1-rc3</version>
8+
<version>4.18.1-rc5</version>
99
</parent>
10-
<version>4.18.1-rc3</version>
10+
<version>4.18.1-rc5</version>
1111
<artifactId>java-client</artifactId>
1212
<packaging>jar</packaging>
1313
<name>Java Client</name>

‎client/src/main/java/io/split/client/SplitClientConfig.java‎

Lines changed: 2 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -112,8 +112,6 @@ private HttpScheme() {
112112
private final CustomHeaderDecorator _customHeaderDecorator;
113113
private final CustomHttpModule _alternativeHTTPModule;
114114

115-
private final int _streamingTokenRefreshRate;
116-
117115
public static Builder builder() {
118116
return new Builder();
119117
}
@@ -172,8 +170,7 @@ private SplitClientConfig(String endpoint,
172170
int invalidSets,
173171
CustomHeaderDecorator customHeaderDecorator,
174172
CustomHttpModule alternativeHTTPModule,
175-
FallbackTreatmentsConfiguration fallbackTreatments,
176-
int streamingTokenRefreshRate) {
173+
FallbackTreatmentsConfiguration fallbackTreatments) {
177174
_endpoint = endpoint;
178175
_eventsEndpoint = eventsEndpoint;
179176
_featuresRefreshRate = pollForFeatureChangesEveryNSeconds;
@@ -229,7 +226,6 @@ private SplitClientConfig(String endpoint,
229226
_customHeaderDecorator = customHeaderDecorator;
230227
_alternativeHTTPModule = alternativeHTTPModule;
231228
_fallbackTreatments = fallbackTreatments;
232-
_streamingTokenRefreshRate = streamingTokenRefreshRate;
233229

234230
Properties props = new Properties();
235231
try {
@@ -450,8 +446,6 @@ public boolean isSdkEndpointOverridden() {
450446

451447
public FallbackTreatmentsConfiguration fallbackTreatments() { return _fallbackTreatments; }
452448

453-
public int streamingTokenRefreshRate() { return _streamingTokenRefreshRate; }
454-
455449
public static final class Builder {
456450
private String _endpoint = SDK_ENDPOINT;
457451
private boolean _endpointSet = false;
@@ -511,7 +505,6 @@ public static final class Builder {
511505
private CustomHeaderDecorator _customHeaderDecorator = null;
512506
private CustomHttpModule _alternativeHTTPModule = null;
513507
private FallbackTreatmentsConfiguration _fallbackTreatments;
514-
private int _streamingTokenRefreshRate = 180;
515508

516509
public Builder() {
517510
}
@@ -1062,11 +1055,6 @@ public Builder threadFactory(ThreadFactory threadFactory) {
10621055
return this;
10631056
}
10641057

1065-
public Builder streamingTokenRefreshRate(int streamingTokenRefreshRate) {
1066-
_streamingTokenRefreshRate = streamingTokenRefreshRate;
1067-
return this;
1068-
}
1069-
10701058
private void verifyRates() {
10711059
if (_featuresRefreshRate < 5 ) {
10721060
throw new IllegalArgumentException("featuresRefreshRate must be >= 5: " + _featuresRefreshRate);
@@ -1087,10 +1075,6 @@ private void verifyRates() {
10871075
if(_telemetryRefreshRate < 60) {
10881076
throw new IllegalStateException("_telemetryRefreshRate must be >= 60");
10891077
}
1090-
1091-
if (_streamingTokenRefreshRate < 60) {
1092-
throw new IllegalStateException("_streamingTokenRefreshRate must be >= 60");
1093-
}
10941078
}
10951079

10961080
private void verifyEndPoints() {
@@ -1291,8 +1275,7 @@ public SplitClientConfig build() {
12911275
_invalidSetsCount,
12921276
_customHeaderDecorator,
12931277
_alternativeHTTPModule,
1294-
_fallbackTreatments,
1295-
_streamingTokenRefreshRate);
1278+
_fallbackTreatments);
12961279
}
12971280
}
12981281
}

‎client/src/main/java/io/split/engine/common/PushManagerImp.java‎

Lines changed: 4 additions & 26 deletions
Original file line numberDiff line numberDiff line change
@@ -52,7 +52,6 @@ public class PushManagerImp implements PushManager {
5252
private final ScheduledExecutorService _scheduledExecutorService;
5353
private AtomicLong _expirationTime;
5454
private final TelemetryRuntimeProducer _telemetryRuntimeProducer;
55-
private final int _streamingTokenRefreshRate;
5655

5756
@VisibleForTesting
5857
/* package private */ PushManagerImp(AuthApiClient authApiClient,
@@ -61,8 +60,7 @@ public class PushManagerImp implements PushManager {
6160
Worker<SegmentQueueDto> segmentWorker,
6261
PushStatusTracker pushStatusTracker,
6362
TelemetryRuntimeProducer telemetryRuntimeProducer,
64-
ThreadFactory threadFactory,
65-
int streamingTokenRefreshRate) {
63+
ThreadFactory threadFactory) {
6664

6765
_authApiClient = checkNotNull(authApiClient);
6866
_eventSourceClient = checkNotNull(eventSourceClient);
@@ -72,7 +70,6 @@ public class PushManagerImp implements PushManager {
7270
_expirationTime = new AtomicLong();
7371
_scheduledExecutorService = buildSingleThreadScheduledExecutor(threadFactory, "Split-SSERefreshToken-%d");
7472
_telemetryRuntimeProducer = checkNotNull(telemetryRuntimeProducer);
75-
_streamingTokenRefreshRate = streamingTokenRefreshRate;
7673
}
7774

7875
public static PushManagerImp build(Synchronizer synchronizer,
@@ -86,8 +83,7 @@ public static PushManagerImp build(Synchronizer synchronizer,
8683
SplitCacheProducer splitCacheProducer,
8784
FlagSetsFilter flagSetsFilter,
8885
RuleBasedSegmentCache ruleBasedSegmentCache,
89-
RuleBasedSegmentParser ruleBasedSegmentParser,
90-
int streamingTokenRefreshRate) {
86+
RuleBasedSegmentParser ruleBasedSegmentParser) {
9187
FeatureFlagsWorker featureFlagsWorker = new FeatureFlagWorkerImp(synchronizer, splitParser, ruleBasedSegmentParser, splitCacheProducer,
9288
ruleBasedSegmentCache, telemetryRuntimeProducer, flagSetsFilter);
9389
Worker<SegmentQueueDto> segmentWorker = new SegmentsWorkerImp(synchronizer);
@@ -100,26 +96,23 @@ public static PushManagerImp build(Synchronizer synchronizer,
10096
segmentWorker,
10197
pushStatusTracker,
10298
telemetryRuntimeProducer,
103-
threadFactory,
104-
streamingTokenRefreshRate);
99+
threadFactory);
105100
}
106101

107102
@Override
108103
public void start() {
109-
_log.debug("#1 - Start PushManagerImp");
110104
try {
111105
lock.lock();
112106
AuthenticationResponse response = _authApiClient.Authenticate();
113107
_log.debug(String.format("Auth service response pushEnabled: %s", response.isPushEnabled()));
114108
if (response.isPushEnabled() && startSse(response.getToken(), response.getChannels())) {
115109
_log.debug("#2 - PushManagerImp connected");
116-
_expirationTime.set(_streamingTokenRefreshRate);
110+
_expirationTime.set(response.getExpiration());
117111
_telemetryRuntimeProducer.recordStreamingEvents(new StreamingEvent(StreamEventsEnum.TOKEN_REFRESH.getType(),
118112
response.getExpiration(), System.currentTimeMillis()));
119113
return;
120114
}
121115

122-
_log.debug("#3 - PushManagerImp error");
123116
cleanUpResources();
124117
if (response.isRetry()) {
125118
_log.debug(String.format("Handling retry error response"));
@@ -128,7 +121,6 @@ public void start() {
128121
_log.debug(String.format("Auth service response is disabled: %s", response.getToken()));
129122
_pushStatusTracker.forcePushDisable();
130123
}
131-
_log.debug("#4 - PushManagerImp error");
132124
} catch (Exception e) {
133125
_log.debug("Exception in PushManager start: " + e.getMessage());
134126
} finally {
@@ -138,11 +130,9 @@ public void start() {
138130

139131
@Override
140132
public void stop() {
141-
_log.debug("#1 - Stopping PushManagerImp");
142133
try {
143134
lock.lock();
144135
cleanUpResources();
145-
_log.debug("#2 - Stopped PushManagerImp");
146136
} catch (Exception e) {
147137
_log.debug("Exception in stopping push manager: " + e.getMessage());
148138
} finally {
@@ -154,11 +144,8 @@ public void stop() {
154144
public void scheduleConnectionReset() {
155145
_log.debug(String.format("scheduleNextTokenRefresh in %s SECONDS", _expirationTime));
156146
_nextTokenRefreshTask = _scheduledExecutorService.schedule(() -> {
157-
_log.debug("#1 - Starting scheduleNextTokenRefresh ...");
158147
stop();
159-
_log.debug("#2 - Finished to stop all streaming engine");
160148
start();
161-
_log.debug("#3 - Finished to start streaming connection");
162149
}, _expirationTime.get(), TimeUnit.SECONDS);
163150
}
164151

@@ -175,9 +162,7 @@ private boolean startSse(String token, String channels) {
175162
@Override
176163
public void startWorkers() {
177164
try {
178-
_log.debug("Starting featureflag worker");
179165
_featureFlagsWorker.start();
180-
_log.debug("Starting segment worker");
181166
_segmentWorker.start();
182167
} catch (Exception e) {
183168
_log.debug("Exception in starting workers: " + e.getMessage());
@@ -187,25 +172,18 @@ public void startWorkers() {
187172
@Override
188173
public void stopWorkers() {
189174
try {
190-
_log.debug("Stopping featureflag worker");
191175
_featureFlagsWorker.stop();
192-
_log.debug("Stopping segment worker");
193176
_segmentWorker.stop();
194177
} catch (Exception e) {
195178
_log.debug("Exception in stopping workers: " + e.getMessage());
196179
}
197180
}
198181

199182
private void cleanUpResources() {
200-
_log.debug("Starting cleanUpResources - #1");
201183
_eventSourceClient.stop();
202-
_log.debug("cleanUpResources - #2");
203184
stopWorkers();
204185
if (_nextTokenRefreshTask != null) {
205-
_log.debug("Cancel nextTokenRefreshTask");
206186
_nextTokenRefreshTask.cancel(false);
207-
_log.debug("Finished cleanUpResources - #3 - Finished cancel nextTokenRefreshTask");
208187
}
209-
_log.debug("Finished cleanUpResources - #4");
210188
}
211189
}

‎client/src/main/java/io/split/engine/common/SyncManagerImp.java‎

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -116,8 +116,7 @@ public static SyncManagerImp build(SplitTasks splitTasks,
116116
splitCacheProducer,
117117
flagSetsFilter,
118118
ruleBasedSegmentCache,
119-
ruleBasedSegmentParser,
120-
config.streamingTokenRefreshRate());
119+
ruleBasedSegmentParser);
121120

122121
return new SyncManagerImp(splitTasks,
123122
config.streamingEnabled(),

‎client/src/main/java/io/split/engine/sse/client/SSEClient.java‎

Lines changed: 4 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,7 @@
99
import org.apache.hc.client5.http.classic.methods.HttpGet;
1010
import org.apache.hc.client5.http.impl.classic.CloseableHttpClient;
1111
import org.apache.hc.client5.http.impl.classic.CloseableHttpResponse;
12+
import org.apache.hc.core5.io.CloseMode;
1213
import org.slf4j.Logger;
1314
import org.slf4j.LoggerFactory;
1415

@@ -121,12 +122,8 @@ public void close() {
121122
_forcedStop.set(true);
122123
if (_state.compareAndSet(ConnectionState.OPEN, ConnectionState.CLOSED)) {
123124
if (_ongoingResponse.get() != null) {
124-
try {
125-
_ongoingRequest.get().abort();
126-
_ongoingResponse.get().close();
127-
} catch (IOException e) {
128-
_log.debug(String.format("SSEClient close forced: %s", e.getMessage()));
129-
}
125+
_ongoingRequest.get().abort();
126+
_ongoingResponse.get().close(CloseMode.IMMEDIATE);
130127
}
131128
}
132129
} catch (Exception e) {
@@ -195,14 +192,7 @@ private void connectAndLoop(URI uri, CountDownLatch signal) {
195192
_log.warn(e.getMessage(), e);
196193
_statusCallback.apply(StatusMessage.NONRETRYABLE_ERROR);
197194
} finally {
198-
_log.debug(String.format("Attempt to close SSE connection"));
199-
try {
200-
_ongoingResponse.get().close();
201-
} catch (IOException e) {
202-
_log.debug(String.format("SSE connection closing exception: %s", e.getMessage()));
203-
_log.debug(e.getMessage());
204-
}
205-
195+
_ongoingResponse.get().close(CloseMode.IMMEDIATE);
206196
_state.set(ConnectionState.CLOSED);
207197
_log.debug("SSEClient finished.");
208198
_forcedStop.set(false);

‎client/src/main/java/io/split/engine/sse/workers/Worker.java‎

Lines changed: 0 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -22,7 +22,6 @@ public Worker(String workerName) {
2222

2323
public void start() {
2424
if (_running.compareAndSet(false, true)) {
25-
_log.debug(String.format("%s Worker starting ...", _workerName));
2625
_queue.clear();
2726
_thread = new Thread( this);
2827
_thread.setName(String.format("%s-worker", _workerName));
@@ -36,7 +35,6 @@ public void start() {
3635

3736
public void stop() {
3837
if (_running.compareAndSet(true, false)) {
39-
_log.debug(String.format("%s stopping Worker", _workerName));
4038
_thread.interrupt();
4139
_log.debug(String.format("%s Worked stopped.", _workerName));
4240
} else {

‎client/src/test/java/io/split/engine/common/PushManagerTest.java‎

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -40,8 +40,7 @@ public void setUp() {
4040
_segmentsWorkerImp,
4141
_pushStatusTracker,
4242
_telemetryStorage,
43-
null,
44-
180);
43+
null);
4544
}
4645

4746
@Test

‎okhttp-modules/pom.xml‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -5,10 +5,10 @@
55
<parent>
66
<artifactId>java-client-parent</artifactId>
77
<groupId>io.split.client</groupId>
8-
<version>4.18.1-rc3</version>
8+
<version>4.18.1-rc5</version>
99
</parent>
1010
<modelVersion>4.0.0</modelVersion>
11-
<version>4.18.1-rc3</version>
11+
<version>4.18.1-rc5</version>
1212
<artifactId>okhttp-modules</artifactId>
1313
<packaging>jar</packaging>
1414
<name>http-modules</name>

‎pluggable-storage/pom.xml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -6,7 +6,7 @@
66
<parent>
77
<artifactId>java-client-parent</artifactId>
88
<groupId>io.split.client</groupId>
9-
<version>4.18.1-rc3</version>
9+
<version>4.18.1-rc5</version>
1010
</parent>
1111

1212
<version>2.1.0</version>

‎pom.xml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,7 @@
44
<modelVersion>4.0.0</modelVersion>
55
<groupId>io.split.client</groupId>
66
<artifactId>java-client-parent</artifactId>
7-
<version>4.18.1-rc3</version>
7+
<version>4.18.1-rc5</version>
88
<dependencyManagement>
99
<dependencies>
1010
<dependency>

0 commit comments

Comments
 (0)