Skip to content

Commit 49c6008

Browse files
committed
Merge remote-tracking branch 'origin' into rlamb/streaming-synchronizer
2 parents 0aba424 + bef2500 commit 49c6008

27 files changed

+1258
-137
lines changed

lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/Components.java

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,7 @@
2626
import com.launchdarkly.sdk.server.integrations.PollingDataSourceBuilder;
2727
import com.launchdarkly.sdk.server.integrations.ServiceEndpointsBuilder;
2828
import com.launchdarkly.sdk.server.integrations.StreamingDataSourceBuilder;
29+
import com.launchdarkly.sdk.server.integrations.DataSystemModes;
2930
import com.launchdarkly.sdk.server.integrations.WrapperInfoBuilder;
3031
import com.launchdarkly.sdk.server.interfaces.HttpAuthentication;
3132
import com.launchdarkly.sdk.server.subsystems.BigSegmentStore;
@@ -476,4 +477,21 @@ public static PluginsConfigurationBuilder plugins() {
476477
* @since 7.1.0
477478
*/
478479
public static WrapperInfoBuilder wrapperInfo() { return new WrapperInfoBuilderImpl(); }
480+
481+
/**
482+
* This API is under active development. Do not use.
483+
*
484+
* Returns a set of builder options for configuring the SDK data system. When the data system configuration
485+
* is used it overrides {@link LDConfig.Builder#dataSource(ComponentConfigurer)} and
486+
* {@link LDConfig.Builder#dataStore(ComponentConfigurer)} in the configuration.
487+
* <p>
488+
* This class is not stable, and not subject to any backwards compatibility guarantees or semantic versioning.
489+
* It is in early access. If you want access to this feature please join the EAP. https://launchdarkly.com/docs/sdk/features/data-saving-mode
490+
* </p>
491+
*
492+
* @return a configuration builder
493+
*/
494+
public static DataSystemModes dataSystem() {
495+
return new DataSystemModes();
496+
}
479497
}

lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/DataSourceUpdatesImpl.java

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -210,6 +210,14 @@ private boolean hasFlagChangeEventListeners() {
210210
return flagChangeEventNotifier.hasListeners();
211211
}
212212

213+
void addFlagChangeListener(FlagChangeListener listener) {
214+
flagChangeEventNotifier.register(listener);
215+
}
216+
217+
void removeFlagChangeListener(FlagChangeListener listener) {
218+
flagChangeEventNotifier.unregister(listener);
219+
}
220+
213221
private void sendChangeEvents(Iterable<KindAndKey> affectedItems) {
214222
for (KindAndKey item: affectedItems) {
215223
if (item.kind == FEATURES) {
Lines changed: 120 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,120 @@
1+
package com.launchdarkly.sdk.server;
2+
3+
import com.launchdarkly.sdk.server.interfaces.DataSourceStatusProvider;
4+
import com.launchdarkly.sdk.server.interfaces.DataStoreStatusProvider;
5+
import com.launchdarkly.sdk.server.subsystems.DataStoreTypes.DataKind;
6+
import com.launchdarkly.sdk.server.subsystems.DataStoreTypes.ItemDescriptor;
7+
import com.launchdarkly.sdk.server.subsystems.DataStoreTypes.KeyedItems;
8+
9+
import java.util.concurrent.Future;
10+
11+
/**
12+
* Internal interface for the data system abstraction.
13+
* <p>
14+
* This interface is package-private and should not be used by application code.
15+
* <p>
16+
* This class is not stable, and not subject to any backwards compatibility guarantees or semantic versioning.
17+
* It is in early access. If you want access to this feature please join the EAP. https://launchdarkly.com/docs/sdk/features/data-saving-mode
18+
*/
19+
interface DataSystem {
20+
/**
21+
* Returns the read-only store interface.
22+
*
23+
* @return the read-only store
24+
*/
25+
ReadOnlyStore getStore();
26+
27+
/**
28+
* Starts the data system.
29+
*
30+
* @return a Future that completes when initialization is complete
31+
*/
32+
Future<Void> start();
33+
34+
/**
35+
* Returns whether the data system has been initialized.
36+
*
37+
* @return true if initialized
38+
*/
39+
boolean isInitialized();
40+
41+
/**
42+
* Returns the flag change notifier interface.
43+
*
44+
* @return the flag change notifier
45+
*/
46+
FlagChangeNotifier getFlagChanged();
47+
48+
/**
49+
* Returns the data source status provider.
50+
*
51+
* @return the data source status provider
52+
*/
53+
DataSourceStatusProvider getDataSourceStatusProvider();
54+
55+
/**
56+
* Returns the data store status provider.
57+
*
58+
* @return the data store status provider
59+
*/
60+
DataStoreStatusProvider getDataStoreStatusProvider();
61+
}
62+
63+
/**
64+
* Internal interface for read-only access to a data store.
65+
* <p>
66+
* This interface is package-private and should not be used by application code.
67+
*/
68+
interface ReadOnlyStore {
69+
/**
70+
* Retrieves an item from the specified collection, if available.
71+
* <p>
72+
* If the item has been deleted and the store contains a placeholder, it should
73+
* return that placeholder rather than null.
74+
*
75+
* @param kind specifies which collection to use
76+
* @param key the unique key of the item within that collection
77+
* @return a versioned item that contains the stored data (or placeholder for deleted data);
78+
* null if the key is unknown
79+
*/
80+
ItemDescriptor get(DataKind kind, String key);
81+
82+
/**
83+
* Retrieves all items from the specified collection.
84+
* <p>
85+
* If the store contains placeholders for deleted items, it should include them in
86+
* the results, not filter them out.
87+
*
88+
* @param kind specifies which collection to use
89+
* @return a collection of key-value pairs; the ordering is not significant
90+
*/
91+
KeyedItems<ItemDescriptor> getAll(DataKind kind);
92+
93+
/**
94+
* Checks whether this store has been initialized with any data yet.
95+
*
96+
* @return true if the store contains data
97+
*/
98+
boolean isInitialized();
99+
}
100+
101+
/**
102+
* Internal interface for flag change notifications.
103+
* <p>
104+
* This interface is package-private and should not be used by application code.
105+
*/
106+
interface FlagChangeNotifier {
107+
/**
108+
* Adds a listener for flag change events.
109+
*
110+
* @param listener the listener to add
111+
*/
112+
void addFlagChangeListener(com.launchdarkly.sdk.server.interfaces.FlagChangeListener listener);
113+
114+
/**
115+
* Removes a listener for flag change events.
116+
*
117+
* @param listener the listener to remove
118+
*/
119+
void removeFlagChangeListener(com.launchdarkly.sdk.server.interfaces.FlagChangeListener listener);
120+
}

lib/sdk/server/src/main/java/com/launchdarkly/sdk/server/DefaultFDv2Requestor.java

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@
33
import com.launchdarkly.logging.LDLogger;
44
import com.launchdarkly.sdk.internal.fdv2.payloads.FDv2Event;
55
import com.launchdarkly.sdk.internal.fdv2.sources.Selector;
6+
import com.launchdarkly.sdk.internal.http.HttpErrors;
67
import com.launchdarkly.sdk.internal.http.HttpHelpers;
78
import com.launchdarkly.sdk.internal.http.HttpProperties;
89
import com.launchdarkly.sdk.json.SerializationException;
@@ -117,7 +118,7 @@ public void onResponse(@Nonnull Call call, @Nonnull Response response) {
117118

118119
if (!response.isSuccessful()) {
119120
future.completeExceptionally(
120-
new IOException("FDv2 polling request failed with status code: " + response.code())
121+
new HttpErrors.HttpErrorException(response.code())
121122
);
122123
return;
123124
}
@@ -143,7 +144,7 @@ public void onResponse(@Nonnull Call call, @Nonnull Response response) {
143144
FDv2PayloadResponse pollingResponse = new FDv2PayloadResponse(events, response.headers());
144145
future.complete(pollingResponse);
145146

146-
} catch (IOException | SerializationException e) {
147+
} catch (Exception e) {
147148
future.completeExceptionally(e);
148149
} finally {
149150
response.close();
Lines changed: 176 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,176 @@
1+
package com.launchdarkly.sdk.server;
2+
3+
import com.launchdarkly.logging.LDLogger;
4+
import com.launchdarkly.sdk.server.interfaces.DataSourceStatusProvider;
5+
import com.launchdarkly.sdk.server.interfaces.DataStoreStatusProvider;
6+
import com.launchdarkly.sdk.server.interfaces.FlagChangeEvent;
7+
import com.launchdarkly.sdk.server.interfaces.FlagChangeListener;
8+
import com.launchdarkly.sdk.server.subsystems.ComponentConfigurer;
9+
import com.launchdarkly.sdk.server.subsystems.DataSource;
10+
import com.launchdarkly.sdk.server.subsystems.DataStore;
11+
import com.launchdarkly.sdk.server.subsystems.LoggingConfiguration;
12+
13+
import java.io.Closeable;
14+
import java.io.IOException;
15+
import java.util.concurrent.Future;
16+
17+
/**
18+
* Internal implementation of the FDv1 data system.
19+
* <p>
20+
* This class is package-private and should not be used by application code.
21+
*/
22+
final class FDv1DataSystem implements DataSystem, Closeable {
23+
private final DataSource dataSource;
24+
private final DataStore dataStore;
25+
private final ReadOnlyStore store;
26+
private final FlagChangeNotifier flagChanged;
27+
private final DataSourceStatusProvider dataSourceStatusProvider;
28+
private final DataStoreStatusProvider dataStoreStatusProvider;
29+
private boolean disposed = false;
30+
31+
/**
32+
* Testing access to internal components.
33+
*/
34+
static final class TestingAccess {
35+
final DataSource dataSource;
36+
37+
TestingAccess(DataSource dataSource) {
38+
this.dataSource = dataSource;
39+
}
40+
}
41+
42+
final TestingAccess testing;
43+
44+
/**
45+
* Gets the underlying data store. This is needed for the evaluator.
46+
* @return the underlying data store
47+
*/
48+
DataStore getUnderlyingStore() {
49+
return dataStore;
50+
}
51+
52+
private FDv1DataSystem(
53+
DataStore store,
54+
DataStoreStatusProvider dataStoreStatusProvider,
55+
DataSourceStatusProvider dataSourceStatusProvider,
56+
DataSource dataSource,
57+
FlagChangeNotifier flagChanged
58+
) {
59+
this.dataStoreStatusProvider = dataStoreStatusProvider;
60+
this.dataSourceStatusProvider = dataSourceStatusProvider;
61+
this.store = new ReadonlyStoreFacade(store);
62+
this.flagChanged = flagChanged;
63+
this.dataSource = dataSource;
64+
this.dataStore = store;
65+
this.testing = new TestingAccess(dataSource);
66+
}
67+
68+
/**
69+
* Creates a new FDv1DataSystem instance.
70+
*
71+
* @param logger the logger
72+
* @param config the SDK configuration
73+
* @param clientContext the client context
74+
* @param logConfig the logging configuration
75+
* @return a new FDv1DataSystem instance
76+
*/
77+
static FDv1DataSystem create(
78+
LDLogger logger,
79+
LDConfig config,
80+
ClientContextImpl clientContext,
81+
LoggingConfiguration logConfig
82+
) {
83+
DataStoreUpdatesImpl dataStoreUpdates = new DataStoreUpdatesImpl(
84+
EventBroadcasterImpl.forDataStoreStatus(clientContext.sharedExecutor, logger));
85+
86+
DataStore dataStore = (config.dataStore == null ? Components.inMemoryDataStore() : config.dataStore)
87+
.build(clientContext.withDataStoreUpdateSink(dataStoreUpdates));
88+
89+
DataStoreStatusProvider dataStoreStatusProvider = new DataStoreStatusProviderImpl(dataStore, dataStoreUpdates);
90+
91+
// Create a single flag change broadcaster to be shared between DataSourceUpdatesImpl and FlagTrackerImpl
92+
EventBroadcasterImpl<FlagChangeListener, FlagChangeEvent> flagChangeBroadcaster =
93+
EventBroadcasterImpl.forFlagChangeEvents(clientContext.sharedExecutor, logger);
94+
95+
// Create a single data source status broadcaster to be shared between DataSourceUpdatesImpl and DataSourceStatusProviderImpl
96+
EventBroadcasterImpl<DataSourceStatusProvider.StatusListener, DataSourceStatusProvider.Status> dataSourceStatusBroadcaster =
97+
EventBroadcasterImpl.forDataSourceStatus(clientContext.sharedExecutor, logger);
98+
99+
DataSourceUpdatesImpl dataSourceUpdates = new DataSourceUpdatesImpl(
100+
dataStore,
101+
dataStoreStatusProvider,
102+
flagChangeBroadcaster,
103+
dataSourceStatusBroadcaster,
104+
clientContext.sharedExecutor,
105+
logConfig.getLogDataSourceOutageAsErrorAfter(),
106+
logger
107+
);
108+
109+
ComponentConfigurer<DataSource> dataSourceFactory = config.offline
110+
? Components.externalUpdatesOnly()
111+
: (config.dataSource == null ? Components.streamingDataSource() : config.dataSource);
112+
DataSource dataSource = dataSourceFactory.build(clientContext.withDataSourceUpdateSink(dataSourceUpdates));
113+
DataSourceStatusProvider dataSourceStatusProvider = new DataSourceStatusProviderImpl(
114+
dataSourceStatusBroadcaster,
115+
dataSourceUpdates);
116+
117+
FlagChangeNotifier flagChanged = new FlagChangedFacade(dataSourceUpdates);
118+
119+
return new FDv1DataSystem(
120+
dataStore,
121+
dataStoreStatusProvider,
122+
dataSourceStatusProvider,
123+
dataSource,
124+
flagChanged
125+
);
126+
}
127+
128+
@Override
129+
public ReadOnlyStore getStore() {
130+
return store;
131+
}
132+
133+
@Override
134+
public Future<Void> start() {
135+
return dataSource.start();
136+
}
137+
138+
@Override
139+
public boolean isInitialized() {
140+
return dataSource.isInitialized();
141+
}
142+
143+
@Override
144+
public FlagChangeNotifier getFlagChanged() {
145+
return flagChanged;
146+
}
147+
148+
@Override
149+
public DataSourceStatusProvider getDataSourceStatusProvider() {
150+
return dataSourceStatusProvider;
151+
}
152+
153+
@Override
154+
public DataStoreStatusProvider getDataStoreStatusProvider() {
155+
return dataStoreStatusProvider;
156+
}
157+
158+
@Override
159+
public void close() throws IOException {
160+
if (disposed) {
161+
return;
162+
}
163+
try {
164+
if (dataSource instanceof Closeable) {
165+
((Closeable) dataSource).close();
166+
}
167+
if (dataStore instanceof Closeable) {
168+
((Closeable) dataStore).close();
169+
}
170+
// DataSourceUpdatesImpl doesn't implement Closeable
171+
} finally {
172+
disposed = true;
173+
}
174+
}
175+
}
176+

0 commit comments

Comments
 (0)