✨ feat: support realtime synchronizer model (#36)

* feat: support realtime toggle update via socket.io

* cleanup code

* test: add it for socketio

* Update build.yml

---------

Co-authored-by: hezean <[email protected]>

Author: gangb-tech

.github/workflows/build.yml

@@ -24,7 +24,9 @@ jobs:
       run: mvn spotless:check
 
     - name: Build with Maven
-      run: mvn -B package --file pom.xml
+      run: |
+        mvn -B package --file pom.xml
+        mvn failsafe:integration-test
 
     - name: Upload coverage reports to Codecov with GitHub Action
-      uses: codecov/codecov-action@v2
+      uses: codecov/codecov-action@v2

pom.xml

@@ -19,6 +19,7 @@
         <version.jackson>2.14.0</version.jackson>
         <version.commons>3.11</version.commons>
         <version.maven-artifact>3.8.5</version.maven-artifact>
+        <version.socketio>2.1.0</version.socketio>
         <groovy.version>3.0.9</groovy.version>
     </properties>
 
@@ -45,6 +46,11 @@
             <artifactId>okhttp</artifactId>
             <version>${version.okhttp}</version>
         </dependency>
+        <dependency>
+            <groupId>io.socket</groupId>
+            <artifactId>socket.io-client</artifactId>
+            <version>${version.socketio}</version>
+        </dependency>
         <dependency>
             <groupId>org.apache.commons</groupId>
             <artifactId>commons-lang3</artifactId>
@@ -216,6 +222,19 @@
                     </markdown>
                 </configuration>
             </plugin>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-failsafe-plugin</artifactId>
+                <version>3.0.0-M4</version>
+                <executions>
+                    <execution>
+                        <goals>
+                            <goal>integration-test</goal>
+                            <goal>verify</goal>
+                        </goals>
+                    </execution>
+                </executions>
+            </plugin>
             <plugin>
                 <groupId>org.jacoco</groupId>
                 <artifactId>jacoco-maven-plugin</artifactId>

src/main/java/com/featureprobe/sdk/server/FPConfig.java

@@ -28,6 +28,8 @@ public final class FPConfig {
 
     static final Duration DEFAULT_INTERVAL = Duration.ofSeconds(5);
 
+    static final Duration DEFAULT_REALTIME_INTERVAL = Duration.ofSeconds(10);
+
     static final Long DEFAULT_START_WAIT = TimeUnit.SECONDS.toNanos(5);
 
     protected static final FPConfig DEFAULT = new Builder().build();
@@ -42,6 +44,8 @@ public final class FPConfig {
 
     URL eventUrl;
 
+    URI realtimeUri;
+
     final String location;
 
     final SynchronizerFactory synchronizerFactory;
@@ -57,14 +61,15 @@ protected FPConfig(Builder builder) {
         this.remoteUri = builder.remoteUri;
         this.location = builder.location;
         this.synchronizerFactory =
-                builder.synchronizer == null ? new PollingSynchronizerFactory() : builder.synchronizer;
+                builder.synchronizer == null ? new StreamingSynchronizerFactory() : builder.synchronizer;
         this.dataRepositoryFactory =
                 builder.dataRepository == null ? new MemoryDataRepositoryFactory() : builder.dataRepository;
         this.eventProcessorFactory = new DefaultEventProcessorFactory();
         this.httpConfiguration =
                 builder.httpConfiguration == null ? HttpConfiguration.DEFAULT : builder.httpConfiguration;
         this.synchronizerUrl = builder.synchronizerUrl;
         this.eventUrl = builder.eventUrl;
+        this.realtimeUri = builder.realtimeUri;
         this.startWait = builder.startWait == null ? DEFAULT_START_WAIT : builder.startWait;
     }
 
@@ -90,6 +95,8 @@ public static class Builder {
 
         private URL eventUrl;
 
+        private URI realtimeUri;
+
         private Long startWait;
 
         public Builder() {
@@ -105,6 +112,18 @@ public Builder pollingMode() {
             return this;
         }
 
+        public Builder streamingMode() {
+            this.refreshInterval = DEFAULT_REALTIME_INTERVAL;
+            this.synchronizer = new StreamingSynchronizerFactory();
+            return this;
+        }
+
+        public Builder streamingMode(Duration refreshInterval) {
+            this.refreshInterval = refreshInterval;
+            this.synchronizer = new StreamingSynchronizerFactory();
+            return this;
+        }
+
         public Builder pollingMode(Duration refreshInterval) {
             this.refreshInterval = refreshInterval;
             this.synchronizer = new PollingSynchronizerFactory();
@@ -143,6 +162,16 @@ public Builder eventUrl(URL eventUrl) {
             return this;
         }
 
+        public Builder realtimeUri(URI realtimeUri) {
+            this.realtimeUri = realtimeUri;
+            return this;
+        }
+
+        public Builder realtimeUri(String realtimeUri) {
+            this.realtimeUri = URI.create(realtimeUri);
+            return this;
+        }
+
         public Builder startWait(Long startWaitTime, TimeUnit unit) {
             this.startWait = unit.toNanos(startWaitTime);
             return this;

src/main/java/com/featureprobe/sdk/server/FPContext.java

@@ -22,6 +22,8 @@
 
 import java.io.InputStream;
 import java.net.MalformedURLException;
+import java.net.URI;
+import java.net.URISyntaxException;
 import java.net.URL;
 import java.time.Duration;
 import java.util.Objects;
@@ -43,8 +45,12 @@ final class FPContext {
 
     private static final String POST_EVENTS_DATA_API = "/api/events";
 
+    private static final String REALTIME_URI_PATH = "/realtime";
+
     private URL synchronizerUrl;
 
+    private URI realtimeUri;
+
     private URL eventUrl;
 
     private final String serverSdkKey;
@@ -69,8 +75,15 @@ final class FPContext {
             } else {
                 this.eventUrl = config.eventUrl;
             }
+            if (Objects.isNull(config.realtimeUri)) {
+                this.realtimeUri = new URI(config.remoteUri.toString() + REALTIME_URI_PATH);
+            } else {
+                this.realtimeUri = config.realtimeUri;
+            }
         } catch (MalformedURLException e) {
             logger.error("construction context error", e);
+        } catch (URISyntaxException e) {
+            logger.error("construction context error", e);
         }
         this.serverSdkKey = serverSdkKey;
         this.refreshInterval = config.refreshInterval;
@@ -119,7 +132,7 @@ private synchronized String getVersion() {
                     .getResourceAsStream("/META-INF/maven/com.featureprobe/server-sdk-java/pom.properties");
             if (is != null) {
                 p.load(is);
-                version = p.getProperty("version", "");
+                version = p.getProperty("version", DEFAULT_SDK_VERSION);
             }
         } catch (Exception e) {
 
@@ -134,8 +147,12 @@ private synchronized String getVersion() {
             }
         }
         if (version == null) {
-            version = "";
+            version = DEFAULT_SDK_VERSION;
         }
         return version;
     }
+
+    public URI getRealtimeUri() {
+        return realtimeUri;
+    }
 }

src/main/java/com/featureprobe/sdk/server/FeatureProbe.java

@@ -52,6 +52,7 @@ public final class FeatureProbe {
     @VisibleForTesting
     final DataRepository dataRepository;
 
+    @VisibleForTesting
     Synchronizer synchronizer;
 
     @VisibleForTesting

src/main/java/com/featureprobe/sdk/server/PollingSynchronizer.java

@@ -97,7 +97,7 @@ public void close() throws IOException {
         }
     }
 
-    private void poll() {
+    public void poll() {
         Request request = new Request.Builder()
                 .url(apiUrl.toString())
                 .headers(headers)

src/main/java/com/featureprobe/sdk/server/StreamingSynchronizer.java

@@ -0,0 +1,86 @@
+/*
+ * 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 com.featureprobe.sdk.server;
+
+import com.google.common.annotations.VisibleForTesting;
+import io.socket.client.IO;
+import io.socket.client.Socket;
+import io.socket.engineio.client.transports.WebSocket;
+import org.slf4j.Logger;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.Future;
+
+public class StreamingSynchronizer implements Synchronizer {
+
+    private static final Logger logger = Loggers.SYNCHRONIZER;
+
+    private PollingSynchronizer pollingSynchronizer;
+
+    @VisibleForTesting
+    Socket socket;
+
+    StreamingSynchronizer(FPContext context, DataRepository dataRepository) {
+        pollingSynchronizer = new PollingSynchronizer(context, dataRepository);
+        this.socket = connectSocket(context);
+    }
+
+    @Override
+    public Future<Void> sync() {
+        return pollingSynchronizer.sync();
+    }
+
+    @Override
+    public void close() throws IOException {
+        synchronized (this) {
+            if (socket != null) {
+                socket.close();
+                socket = null;
+            }
+        }
+    }
+
+    private Socket connectSocket(FPContext context) {
+        IO.Options sioOptions = IO.Options.builder()
+                .setTransports(new String[]{WebSocket.NAME})
+                .setPath(context.getRealtimeUri().getPath())
+                .build();
+        Socket sio = IO.socket(context.getRealtimeUri(), sioOptions);
+
+        sio.on("connect", objects -> {
+            logger.info("connect socket-io success");
+            Map<String, String> credentials = new HashMap<>(1);
+            credentials.put("key", context.getServerSdkKey());
+            sio.emit("register", credentials);
+        });
+
+        sio.on("update", objects -> {
+            logger.info("socket-io recv update event");
+            pollingSynchronizer.poll();
+        });
+
+        sio.on("disconnect", objects -> logger.info("socket-io disconnected"));
+
+        sio.on("connect_error", objects -> logger.error("socket-io error: {}", objects));
+
+        return sio.connect();
+    }
+
+}

src/main/java/com/featureprobe/sdk/server/StreamingSynchronizerFactory.java

@@ -0,0 +1,27 @@
+/*
+ * 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 com.featureprobe.sdk.server;
+
+public final class StreamingSynchronizerFactory implements SynchronizerFactory {
+
+    @Override
+    public Synchronizer createSynchronizer(FPContext context, DataRepository dataRepository) {
+        return new StreamingSynchronizer(context, dataRepository);
+    }
+
+}

src/test/groovy/com/featureprobe/sdk/server/PollingSynchronizerIT.groovy

@@ -0,0 +1,39 @@
+package com.featureprobe.sdk.server
+
+import ch.qos.logback.classic.Level
+import ch.qos.logback.classic.Logger
+import org.slf4j.LoggerFactory
+import spock.lang.Specification
+
+class PollingSynchronizerIT extends Specification {
+
+    def "Socketio realtime toggle update"() {
+
+        (LoggerFactory.getLogger(org.slf4j.Logger.ROOT_LOGGER_NAME) as Logger).setLevel(Level.DEBUG)
+
+        given:
+        def config = FPConfig.builder()
+                .streamingMode()
+                .remoteUri("https://featureprobe.io/server")
+                .realtimeUri("https://featureprobe.io/server/realtime")
+                .useMemoryRepository()
+                .build()
+        def featureProbe = new FeatureProbe("server-61db54ecea79824cae3ac38d73f1961d698d0477", config)
+        def repository = featureProbe.dataRepository
+        def socket = (featureProbe.synchronizer as StreamingSynchronizer).socket
+        def updateCnt = 0
+        socket.on("update", objects -> updateCnt++)
+
+        sleep(5000)
+
+        featureProbe.close()
+
+        sleep(5000)
+
+        expect:
+        repository.initialized()
+        !socket.connected()
+        updateCnt > 0
+    }
+}
+