chore: integration test add realtime update

SSebo · SSebo · commit cb877cd9767b · 2022-11-08T15:16:16.000+08:00
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/Cargo.toml b/Cargo.toml
@@ -70,4 +70,5 @@ criterion = "0.3"
 rusty-hook = "^0.11.2"
 tokio = { version = "1", features = ["full"] }
 tracing-subscriber = "0.3"
-feature-probe-server = "1.3.3"
+feature-probe-server = { version = "1.3.5", features = ["realtime"] }
+
diff --git a/Makefile b/Makefile
@@ -12,5 +12,6 @@ release-test:
 	cargo test --release --verbose --features async --no-default-features
 test:
 	cargo test --verbose && \
-	cargo test --verbose --features use_tokio --features internal --features event_tokio --no-default-features
+	cargo test --verbose --features use_tokio --features internal --features event_tokio --no-default-features && \
+	cargo test --verbose --features use_tokio --features internal --features event_tokio --features realtime --no-default-features
 
diff --git a/src/config.rs b/src/config.rs
@@ -75,7 +75,7 @@ impl FPConfig {
         };
 
         #[cfg(all(feature = "use_tokio", feature = "realtime"))]
-        let realtime_url = match self.realtime_url {
+        let realtime_url = match &self.realtime_url {
             None => {
                 Url::parse(&(remote_url.clone() + "api/realtime")).expect("invalid realtime url")
             }
diff --git a/src/feature_probe.rs b/src/feature_probe.rs
@@ -282,7 +282,7 @@ impl FeatureProbe {
     #[cfg(all(feature = "use_tokio", feature = "realtime"))]
     fn socket_on_connect(socket: socketio_rs::Socket, server_sdk_key: String) -> SocketCallback {
         let sdk_key = server_sdk_key;
-        trace!("on conect: {:?}", sdk_key);
+        trace!("socket_on_connect: {:?}", sdk_key);
         async move {
             if let Err(e) = socket
                 .emit("register", serde_json::json!({ "sdk_key": sdk_key }))
@@ -296,10 +296,12 @@ impl FeatureProbe {
 
     #[cfg(all(feature = "use_tokio", feature = "realtime"))]
     fn socket_on_update(slf: Self, payload: Option<socketio_rs::Payload>) -> SocketCallback {
-        trace!("on update: {:?}", payload);
+        use crate::sync::SyncType;
+
+        trace!("socket_on_update: {:?}", payload);
         async move {
             if let Some(syncer) = &slf.syncer {
-                let _ = syncer.sync_now().await;
+                let _ = syncer.sync_now(SyncType::Realtime).await;
             } else {
                 tracing::warn!("socket receive update event, but no synchronizer");
             }
diff --git a/src/lib.rs b/src/lib.rs
@@ -7,6 +7,7 @@ mod user;
 pub use crate::config::FPConfig;
 pub use crate::evaluate::{load_json, EvalDetail, Repository, Segment, Toggle};
 pub use crate::feature_probe::FeatureProbe;
+pub use crate::sync::SyncType;
 pub use crate::user::FPUser;
 use headers::{Error, Header, HeaderName, HeaderValue};
 use http::header::AUTHORIZATION;
diff --git a/src/sync.rs b/src/sync.rs
@@ -10,13 +10,19 @@ use tracing::trace;
 use tracing::{debug, error};
 use url::Url;
 
-pub type UpdateCallback = Box<dyn Fn(Repository, Repository) + Send>;
+pub type UpdateCallback = Box<dyn Fn(Repository, Repository, SyncType) + Send>;
 
 #[derive(Debug, Clone)]
 pub struct Synchronizer {
     inner: Arc<Inner>,
 }
 
+#[derive(Debug)]
+pub enum SyncType {
+    Realtime,
+    Polling,
+}
+
 struct Inner {
     toggles_url: Url,
     refresh_interval: Duration,
@@ -77,7 +83,9 @@ impl Synchronizer {
 
         let is_timeout = Self::init_timeout_fn(start_wait, interval_duration, start);
         std::thread::spawn(move || loop {
-            if let Some(r) = Self::should_send(inner.sync_now(), &is_timeout, is_send) {
+            if let Some(r) =
+                Self::should_send(inner.sync_now(SyncType::Polling), &is_timeout, is_send)
+            {
                 is_send = true;
                 let _ = tx.try_send(r);
             }
@@ -105,7 +113,7 @@ impl Synchronizer {
         tokio::spawn(async move {
             let mut interval = tokio::time::interval(inner.refresh_interval);
             loop {
-                let result = inner.sync_now().await;
+                let result = inner.sync_now(SyncType::Polling).await;
 
                 if let Some(r) = Self::should_send(result, &is_timeout, is_send) {
                     is_send = true;
@@ -135,8 +143,8 @@ impl Synchronizer {
     }
 
     #[cfg(test)]
-    fn notify_update(&self, old_repo: Repository, new_repo: Repository) {
-        self.inner.notify_update(old_repo, new_repo)
+    fn notify_update(&self, old_repo: Repository, new_repo: Repository, t: SyncType) {
+        self.inner.notify_update(old_repo, new_repo, t)
     }
 
     fn init_timeout_fn(
@@ -172,17 +180,17 @@ impl Synchronizer {
     }
 
     #[cfg(all(feature = "use_tokio", feature = "realtime"))]
-    pub async fn sync_now(&self) -> Result<(), FPError> {
-        self.inner.sync_now().await
+    pub async fn sync_now(&self, t: SyncType) -> Result<(), FPError> {
+        self.inner.sync_now(t).await
     }
 }
 
 impl Inner {
     #[cfg(feature = "use_tokio")]
-    pub async fn sync_now(&self) -> Result<(), FPError> {
+    pub async fn sync_now(&self, t: SyncType) -> Result<(), FPError> {
         use http::header::USER_AGENT;
 
-        trace!("sync now {:?}", self.auth);
+        trace!("sync_now {:?} {:?}", self.auth, t);
         let mut request = self
             .client
             .request(Method::GET, self.toggles_url.clone())
@@ -213,7 +221,7 @@ impl Inner {
                             let old = (*repo).clone();
                             let new = r.clone();
                             *repo = r;
-                            self.notify_update(old, new);
+                            self.notify_update(old, new, t);
                         }
                         let mut is_init = self.is_init.write();
                         *is_init = true;
@@ -225,8 +233,8 @@ impl Inner {
     }
 
     #[cfg(feature = "use_std")]
-    pub fn sync_now(&self) -> Result<(), FPError> {
-        trace!("sync_now {:?}", self.auth);
+    pub fn sync_now(&self, t: SyncType) -> Result<(), FPError> {
+        trace!("sync_now {:?}, {:?}", self.auth, t);
         //TODO: report failure
         let mut request = ureq::get(self.toggles_url.as_str())
             .set(
@@ -258,7 +266,7 @@ impl Inner {
                                 let old = (*repo).clone();
                                 let new = r.clone();
                                 *repo = r;
-                                self.notify_update(old, new);
+                                self.notify_update(old, new, t);
                             }
                             let mut is_init = self.is_init.write();
                             *is_init = true;
@@ -270,10 +278,10 @@ impl Inner {
         }
     }
 
-    fn notify_update(&self, old_repo: Repository, new_repo: Repository) {
+    fn notify_update(&self, old_repo: Repository, new_repo: Repository, t: SyncType) {
         let lock = self.update_callback.lock();
         if let Some(cb) = &*lock {
-            cb(old_repo, new_repo)
+            cb(old_repo, new_repo, t)
         }
     }
 }
@@ -291,10 +299,10 @@ mod tests {
         let mut syncer = build_synchronizer(9000);
         let (tx, rx) = channel();
 
-        syncer.set_update_callback(Box::new(move |_old, _new| tx.send(()).unwrap()));
+        syncer.set_update_callback(Box::new(move |_old, _new, _| tx.send(()).unwrap()));
         let old = Repository::default();
         let new = Repository::default();
-        syncer.notify_update(old, new);
+        syncer.notify_update(old, new, SyncType::Polling);
 
         assert!(rx.try_recv().is_ok())
     }
diff --git a/tests/integration_test.rs b/tests/integration_test.rs
@@ -1,29 +1,53 @@
 use std::{sync::Arc, time::Duration};
 
 use feature_probe_server::{
-    http::{serve_http, FpHttpHandler, LocalFileHttpHandler},
+    http::{serve_http, FpHttpHandler, LocalFileHttpHandlerForTest},
+    realtime::RealtimeSocket,
     repo::SdkRepository,
     ServerConfig,
 };
-use feature_probe_server_sdk::{FPConfig, FPUser, FeatureProbe, Url};
+use feature_probe_server_sdk::{FPConfig, FPUser, FeatureProbe, SyncType, Url};
+use parking_lot::Mutex;
+use tracing::info;
 
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn integration_test() {
-    // let _ = tracing_subscriber::fmt().init();
+    // let _ = tracing_subscriber::fmt()
+    //     .with_env_filter("feature_probe_server_sdk=trace,integration=trace")
+    //     .pretty()
+    //     .init();
 
     let api_port = 19980;
     let server_port = 19990;
-    setup_server(api_port, server_port).await;
+    let realtime_port = 19999;
+    setup_server(api_port, server_port, realtime_port).await;
 
     let config = FPConfig {
         remote_url: Url::parse(&format!("http://127.0.0.1:{}", server_port)).unwrap(),
         server_sdk_key: "server-sdk-key1".to_owned(),
-        refresh_interval: Duration::from_secs(1),
+        refresh_interval: Duration::from_secs(2),
         start_wait: Some(Duration::from_secs(5)),
+        #[cfg(feature = "realtime")]
+        realtime_url: Some(Url::parse(&format!("http://127.0.0.1:{}", realtime_port)).unwrap()),
         ..Default::default()
     };
 
-    let fp = FeatureProbe::new(config);
+    let mut fp = FeatureProbe::new(config);
+
+    let did_update = {
+        let did_update = Arc::new(Mutex::new((false, false)));
+        let did_update_clone = did_update.clone();
+
+        fp.set_update_callback(Box::new(move |_old, _new, t| {
+            let mut lock = did_update_clone.lock();
+            match t {
+                SyncType::Realtime => lock.0 = true,
+                SyncType::Polling => lock.1 = true,
+            };
+        }));
+
+        did_update
+    };
 
     let user = FPUser::new();
 
@@ -32,14 +56,22 @@ async fn integration_test() {
     assert!(fp.initialized());
 
     let b = fp.bool_detail("bool_toggle", &user, false);
-    assert_eq!(b.value, true);
+    assert!(b.value);
+
+    tokio::time::sleep(Duration::from_millis(3000)).await;
+    let lock = did_update.lock();
+    #[cfg(feature = "realtime")]
+    assert!(lock.0);
+    #[cfg(not(feature = "realtime"))]
+    assert!(lock.1);
 }
 
-async fn setup_server(api_port: u16, server_port: u16) {
+async fn setup_server(api_port: u16, server_port: u16, realtime_port: u16) {
+    let mut mock_api = LocalFileHttpHandlerForTest::default();
+    mock_api.version_update = true;
     // mock fp api
-    tokio::spawn(serve_http::<LocalFileHttpHandler>(
-        api_port,
-        LocalFileHttpHandler {},
+    tokio::spawn(serve_http::<LocalFileHttpHandlerForTest>(
+        api_port, mock_api,
     ));
 
     let server_sdk_key = "server-sdk-key1".to_owned();
@@ -55,15 +87,18 @@ async fn setup_server(api_port: u16, server_port: u16) {
         .parse()
         .unwrap();
     let refresh_interval = Duration::from_secs(1);
-    let repo = SdkRepository::new(ServerConfig {
+    let config = ServerConfig {
         toggles_url,
         server_port,
+        realtime_port,
         refresh_interval,
         keys_url: None,
         events_url: events_url.clone(),
         client_sdk_key: Some(client_sdk_key.clone()),
         server_sdk_key: Some(server_sdk_key.clone()),
-    });
+    };
+    let realtime_socket = RealtimeSocket::serve(config.realtime_port);
+    let repo = SdkRepository::new(config, realtime_socket);
     repo.sync(client_sdk_key, server_sdk_key, 1);
     let repo = Arc::new(repo);
     let feature_probe_server = FpHttpHandler {

Original file line number	Diff line number	Diff line change
`@@ -75,7 +75,7 @@ impl FPConfig {`
`75`	`75`	`};`
`76`	`76`
`77`	`77`	`#[cfg(all(feature = "use_tokio", feature = "realtime"))]`
`78`		`- let realtime_url = match self.realtime_url {`
	`78`	`+ let realtime_url = match &self.realtime_url {`
`79`	`79`	`None => {`
`80`	`80`	`Url::parse(&(remote_url.clone() + "api/realtime")).expect("invalid realtime url")`
`81`	`81`	`}`