Skip to content

Commit fba5aa2

Browse files
kinyoklionkinyoklion
committed
Extract conditions from FDv2DataSource.
1 parent b46ac5e commit fba5aa2

4 files changed

Lines changed: 229 additions & 195 deletions

File tree

‎lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/FDv2DataSource.java‎

Lines changed: 5 additions & 193 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,11 @@
1818
import java.util.concurrent.atomic.AtomicBoolean;
1919
import java.util.stream.Collectors;
2020

21+
import static com.launchdarkly.sdk.server.FDv2DataSourceConditions.Condition;
22+
import static com.launchdarkly.sdk.server.FDv2DataSourceConditions.ConditionFactory;
23+
import static com.launchdarkly.sdk.server.FDv2DataSourceConditions.FallbackCondition;
24+
import static com.launchdarkly.sdk.server.FDv2DataSourceConditions.RecoveryCondition;
25+
2126
class FDv2DataSource implements DataSource {
2227
/**
2328
* Default fallback timeout of 2 minutes. The timeout is only configurable for testing.
@@ -50,199 +55,6 @@ class FDv2DataSource implements DataSource {
5055

5156
private final LDLogger logger;
5257

53-
/**
54-
* Package-private for testing.
55-
*/
56-
interface Condition {
57-
enum ConditionType {
58-
FALLBACK,
59-
RECOVERY,
60-
}
61-
CompletableFuture<Condition> execute();
62-
63-
void inform(FDv2SourceResult sourceResult);
64-
65-
void close() throws IOException;
66-
67-
ConditionType getType();
68-
}
69-
70-
interface ConditionFactory {
71-
Condition build();
72-
73-
Condition.ConditionType getType();
74-
}
75-
76-
77-
static abstract class TimedCondition implements Condition {
78-
protected final CompletableFuture<Condition> resultFuture = new CompletableFuture<>();
79-
80-
protected final ScheduledExecutorService sharedExecutor;
81-
82-
/**
83-
* Future for the timeout task, if any. Will be null when no timeout is active.
84-
*/
85-
protected ScheduledFuture<Void> timerFuture;
86-
87-
/**
88-
* Timeout duration for the fallback operation.
89-
*/
90-
protected final long timeoutSeconds;
91-
92-
public TimedCondition(ScheduledExecutorService sharedExecutor, long timeoutSeconds) {
93-
this.sharedExecutor = sharedExecutor;
94-
this.timeoutSeconds = timeoutSeconds;
95-
}
96-
97-
@Override
98-
public CompletableFuture<Condition> execute() {
99-
return resultFuture;
100-
}
101-
102-
@Override
103-
public void close() throws IOException {
104-
if (timerFuture != null) {
105-
timerFuture.cancel(false);
106-
timerFuture = null;
107-
}
108-
}
109-
110-
static abstract class Factory implements ConditionFactory {
111-
protected final ScheduledExecutorService sharedExecutor;
112-
protected final long timeoutSeconds;
113-
114-
public Factory(ScheduledExecutorService sharedExecutor, long timeout) {
115-
this.sharedExecutor = sharedExecutor;
116-
this.timeoutSeconds = timeout;
117-
}
118-
}
119-
}
120-
121-
/**
122-
* This condition is used to determine if a fallback should be performed. It is informed of each data source result
123-
* via {@link #inform(FDv2SourceResult)}. Based on the results, it updates its internal state. When the fallback
124-
* condition is met, then the {@link Future} returned by {@link #execute()} will complete.
125-
* <p>
126-
* This is package-private, instead of private, for ease of testing.
127-
*/
128-
static class FallbackCondition extends TimedCondition {
129-
static class Factory extends TimedCondition.Factory {
130-
public Factory(ScheduledExecutorService sharedExecutor, long timeout) {
131-
super(sharedExecutor, timeout);
132-
}
133-
@Override
134-
public Condition build() {
135-
return new FallbackCondition(sharedExecutor, timeoutSeconds);
136-
}
137-
138-
@Override
139-
public ConditionType getType() {
140-
return ConditionType.FALLBACK;
141-
}
142-
}
143-
144-
public FallbackCondition(ScheduledExecutorService sharedExecutor, long timeoutSeconds) {
145-
super(sharedExecutor, timeoutSeconds);
146-
}
147-
148-
@Override
149-
public void inform(FDv2SourceResult sourceResult) {
150-
if(sourceResult == null) {
151-
return;
152-
}
153-
if(sourceResult.getResultType() == FDv2SourceResult.ResultType.CHANGE_SET) {
154-
if(timerFuture != null) {
155-
timerFuture.cancel(false);
156-
timerFuture = null;
157-
}
158-
}
159-
if(sourceResult.getResultType() == FDv2SourceResult.ResultType.STATUS && sourceResult.getStatus().getState() == FDv2SourceResult.State.INTERRUPTED) {
160-
if (timerFuture == null) {
161-
timerFuture = sharedExecutor.schedule(() -> {
162-
resultFuture.complete(this);
163-
return null;
164-
}, timeoutSeconds, TimeUnit.SECONDS);
165-
}
166-
}
167-
}
168-
169-
@Override
170-
public ConditionType getType() {
171-
return ConditionType.FALLBACK;
172-
}
173-
}
174-
175-
static class RecoveryCondition extends TimedCondition {
176-
177-
static class Factory extends TimedCondition.Factory {
178-
public Factory(ScheduledExecutorService sharedExecutor, long timeout) {
179-
super(sharedExecutor, timeout);
180-
}
181-
@Override
182-
public Condition build() {
183-
return new RecoveryCondition(sharedExecutor, timeoutSeconds);
184-
}
185-
186-
@Override
187-
public ConditionType getType() {
188-
return ConditionType.RECOVERY;
189-
}
190-
}
191-
192-
public RecoveryCondition(ScheduledExecutorService sharedExecutor, long timeoutSeconds) {
193-
super(sharedExecutor, timeoutSeconds);
194-
timerFuture = sharedExecutor.schedule(() -> {
195-
resultFuture.complete(this);
196-
return null;
197-
}, timeoutSeconds, TimeUnit.SECONDS);
198-
}
199-
200-
@Override
201-
public void inform(FDv2SourceResult sourceResult) {
202-
// Time-based recovery.
203-
}
204-
205-
@Override
206-
public ConditionType getType() {
207-
return ConditionType.RECOVERY;
208-
}
209-
}
210-
211-
private static class SynchronizerFactoryWithState {
212-
public enum State {
213-
/**
214-
* This synchronizer is available to use.
215-
*/
216-
Available,
217-
218-
/**
219-
* This synchronizer is no longer available to use.
220-
*/
221-
Blocked
222-
}
223-
224-
private final DataSourceFactory<Synchronizer> factory;
225-
226-
private State state = State.Available;
227-
228-
229-
public SynchronizerFactoryWithState(DataSourceFactory<Synchronizer> factory) {
230-
this.factory = factory;
231-
}
232-
233-
public State getState() {
234-
return state;
235-
}
236-
237-
public void block() {
238-
state = State.Blocked;
239-
}
240-
241-
public Synchronizer build() {
242-
return factory.build();
243-
}
244-
}
245-
24658
public interface DataSourceFactory<T> {
24759
T build();
24860
}

‎lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/FDv2DataSourceConditions.java‎

Lines changed: 184 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,184 @@
1+
package com.launchdarkly.sdk.server;
2+
3+
import com.launchdarkly.sdk.server.datasources.FDv2SourceResult;
4+
5+
import java.io.Closeable;
6+
import java.io.IOException;
7+
import java.util.concurrent.CompletableFuture;
8+
import java.util.concurrent.ScheduledExecutorService;
9+
import java.util.concurrent.ScheduledFuture;
10+
import java.util.concurrent.TimeUnit;
11+
12+
/**
13+
* Container class for FDv2 data source conditions and related types.
14+
* <p>
15+
* This class is non-constructable and serves only as a namespace for condition-related types.
16+
* Package-private for internal use and testing.
17+
*/
18+
class FDv2DataSourceConditions {
19+
/**
20+
* Private constructor to prevent instantiation.
21+
*/
22+
private FDv2DataSourceConditions() {
23+
}
24+
25+
/**
26+
* Package-private for testing.
27+
*/
28+
interface Condition extends Closeable {
29+
enum ConditionType {
30+
FALLBACK,
31+
RECOVERY,
32+
}
33+
34+
CompletableFuture<Condition> execute();
35+
36+
void inform(FDv2SourceResult sourceResult);
37+
38+
void close() throws IOException;
39+
40+
ConditionType getType();
41+
}
42+
43+
interface ConditionFactory {
44+
Condition build();
45+
46+
Condition.ConditionType getType();
47+
}
48+
49+
static abstract class TimedCondition implements Condition {
50+
protected final CompletableFuture<Condition> resultFuture = new CompletableFuture<>();
51+
52+
protected final ScheduledExecutorService sharedExecutor;
53+
54+
/**
55+
* Future for the timeout task, if any. Will be null when no timeout is active.
56+
*/
57+
protected ScheduledFuture<Void> timerFuture;
58+
59+
/**
60+
* Timeout duration for the fallback operation.
61+
*/
62+
protected final long timeoutSeconds;
63+
64+
public TimedCondition(ScheduledExecutorService sharedExecutor, long timeoutSeconds) {
65+
this.sharedExecutor = sharedExecutor;
66+
this.timeoutSeconds = timeoutSeconds;
67+
}
68+
69+
@Override
70+
public CompletableFuture<Condition> execute() {
71+
return resultFuture;
72+
}
73+
74+
@Override
75+
public void close() throws IOException {
76+
if (timerFuture != null) {
77+
timerFuture.cancel(false);
78+
timerFuture = null;
79+
}
80+
}
81+
82+
static abstract class Factory implements ConditionFactory {
83+
protected final ScheduledExecutorService sharedExecutor;
84+
protected final long timeoutSeconds;
85+
86+
public Factory(ScheduledExecutorService sharedExecutor, long timeout) {
87+
this.sharedExecutor = sharedExecutor;
88+
this.timeoutSeconds = timeout;
89+
}
90+
}
91+
}
92+
93+
/**
94+
* This condition is used to determine if a fallback should be performed. It is informed of each data source result
95+
* via {@link #inform(FDv2SourceResult)}. Based on the results, it updates its internal state. When the fallback
96+
* condition is met, then the {@link java.util.concurrent.Future} returned by {@link #execute()} will complete.
97+
* <p>
98+
* This is package-private, instead of private, for ease of testing.
99+
*/
100+
static class FallbackCondition extends TimedCondition {
101+
static class Factory extends TimedCondition.Factory {
102+
public Factory(ScheduledExecutorService sharedExecutor, long timeout) {
103+
super(sharedExecutor, timeout);
104+
}
105+
106+
@Override
107+
public Condition build() {
108+
return new FallbackCondition(sharedExecutor, timeoutSeconds);
109+
}
110+
111+
@Override
112+
public ConditionType getType() {
113+
return ConditionType.FALLBACK;
114+
}
115+
}
116+
117+
public FallbackCondition(ScheduledExecutorService sharedExecutor, long timeoutSeconds) {
118+
super(sharedExecutor, timeoutSeconds);
119+
}
120+
121+
@Override
122+
public void inform(FDv2SourceResult sourceResult) {
123+
if (sourceResult == null) {
124+
return;
125+
}
126+
if (sourceResult.getResultType() == FDv2SourceResult.ResultType.CHANGE_SET) {
127+
if (timerFuture != null) {
128+
timerFuture.cancel(false);
129+
timerFuture = null;
130+
}
131+
}
132+
if (sourceResult.getResultType() == FDv2SourceResult.ResultType.STATUS && sourceResult.getStatus().getState() == FDv2SourceResult.State.INTERRUPTED) {
133+
if (timerFuture == null) {
134+
timerFuture = sharedExecutor.schedule(() -> {
135+
resultFuture.complete(this);
136+
return null;
137+
}, timeoutSeconds, TimeUnit.SECONDS);
138+
}
139+
}
140+
}
141+
142+
@Override
143+
public ConditionType getType() {
144+
return ConditionType.FALLBACK;
145+
}
146+
}
147+
148+
static class RecoveryCondition extends TimedCondition {
149+
150+
static class Factory extends TimedCondition.Factory {
151+
public Factory(ScheduledExecutorService sharedExecutor, long timeout) {
152+
super(sharedExecutor, timeout);
153+
}
154+
155+
@Override
156+
public Condition build() {
157+
return new RecoveryCondition(sharedExecutor, timeoutSeconds);
158+
}
159+
160+
@Override
161+
public ConditionType getType() {
162+
return ConditionType.RECOVERY;
163+
}
164+
}
165+
166+
public RecoveryCondition(ScheduledExecutorService sharedExecutor, long timeoutSeconds) {
167+
super(sharedExecutor, timeoutSeconds);
168+
timerFuture = sharedExecutor.schedule(() -> {
169+
resultFuture.complete(this);
170+
return null;
171+
}, timeoutSeconds, TimeUnit.SECONDS);
172+
}
173+
174+
@Override
175+
public void inform(FDv2SourceResult sourceResult) {
176+
// Time-based recovery.
177+
}
178+
179+
@Override
180+
public ConditionType getType() {
181+
return ConditionType.RECOVERY;
182+
}
183+
}
184+
}

0 commit comments

Comments
 (0)