From 40022f7e1517603a860faa57527c67c0c36def32 Mon Sep 17 00:00:00 2001
From: Mark Phelps <209477+markphelps@users.noreply.github.com>
Date: Thu, 17 Oct 2024 11:20:52 -0400
Subject: [PATCH] refactor(ffi): refactor engine for streaming (#430)

* refactor: decouple evaluator and parser to support streaming

Signed-off-by: Mark Phelps <209477+markphelps@users.noreply.github.com>

* chore: use channels for message passing

Signed-off-by: Mark Phelps <209477+markphelps@users.noreply.github.com>

* chore: handle errors from http

Signed-off-by: Mark Phelps <209477+markphelps@users.noreply.github.com>

* chore: return response from fetcher

Signed-off-by: Mark Phelps <209477+markphelps@users.noreply.github.com>

* feat: impl streaming

Signed-off-by: Mark Phelps <209477+markphelps@users.noreply.github.com>

* chore: rm streaming file

Signed-off-by: Mark Phelps <209477+markphelps@users.noreply.github.com>

* chore: fix async tests

Signed-off-by: Mark Phelps <209477+markphelps@users.noreply.github.com>

* chore: handle json parsing error

Signed-off-by: Mark Phelps <209477+markphelps@users.noreply.github.com>

* chore: rm unneeded test

Signed-off-by: Mark Phelps <209477+markphelps@users.noreply.github.com>

* chore: rm tokio from main deps

Signed-off-by: Mark Phelps <209477+markphelps@users.noreply.github.com>

* chore: fix sending async

Signed-off-by: Mark Phelps <209477+markphelps@users.noreply.github.com>

* chore: minor cleanup

Signed-off-by: Mark Phelps <209477+markphelps@users.noreply.github.com>

* chore: add some logging

Signed-off-by: Mark Phelps <209477+markphelps@users.noreply.github.com>

* chore: try to fix thread lifecycle management

Signed-off-by: Mark Phelps <209477+markphelps@users.noreply.github.com>

* chore: put polling in loop

Signed-off-by: Mark Phelps <209477+markphelps@users.noreply.github.com>

* chore: add some debug logging

Signed-off-by: Mark Phelps <209477+markphelps@users.noreply.github.com>

* chore: dont reset snapshot on error state

Signed-off-by: Mark Phelps <209477+markphelps@users.noreply.github.com>

* chore: dont print error on shutdown

Signed-off-by: Mark Phelps <209477+markphelps@users.noreply.github.com>

* chore: handle wait for async code to finish

Signed-off-by: Mark Phelps <209477+markphelps@users.noreply.github.com>

* chore: use println for debug

Signed-off-by: Mark Phelps <209477+markphelps@users.noreply.github.com>

* chore: more debug info

Signed-off-by: Mark Phelps <209477+markphelps@users.noreply.github.com>

* chore: perform initial fetch on engine start

Signed-off-by: Mark Phelps <209477+markphelps@users.noreply.github.com>

* chore: refactor initial fetch

Signed-off-by: Mark Phelps <209477+markphelps@users.noreply.github.com>

* chore: set error in snapshot

Signed-off-by: Mark Phelps <209477+markphelps@users.noreply.github.com>

* chore: test stream fetch

Signed-off-by: Mark Phelps <209477+markphelps@users.noreply.github.com>

* chore: error message

Signed-off-by: Mark Phelps <209477+markphelps@users.noreply.github.com>

* chore: clippyyyyyyyyy

Signed-off-by: Mark Phelps <209477+markphelps@users.noreply.github.com>

* chore: rm 303 handling

Signed-off-by: Mark Phelps <209477+markphelps@users.noreply.github.com>

* chore: expose fetch mode

Signed-off-by: Mark Phelps <209477+markphelps@users.noreply.github.com>

* chore: fix url for streaming

Signed-off-by: Mark Phelps <209477+markphelps@users.noreply.github.com>

* chore: fix streaming test

Signed-off-by: Mark Phelps <209477+markphelps@users.noreply.github.com>

* chore: expose fetch mode to other languages

Signed-off-by: Mark Phelps <209477+markphelps@users.noreply.github.com>

* chore: fix java build

Signed-off-by: Mark Phelps <209477+markphelps@users.noreply.github.com>

* chore: rename file cause java

Signed-off-by: Mark Phelps <209477+markphelps@users.noreply.github.com>

* chore: handle stream chunk parsing; rm timeout

Signed-off-by: Mark Phelps <209477+markphelps@users.noreply.github.com>

* chore: fmt java

Signed-off-by: Mark Phelps <209477+markphelps@users.noreply.github.com>

* chore: ruby fmt

Signed-off-by: Mark Phelps <209477+markphelps@users.noreply.github.com>

* chore: fmt python

Signed-off-by: Mark Phelps <209477+markphelps@users.noreply.github.com>

* chore: retry requests, ensure consistency in error handling

Signed-off-by: Mark Phelps <209477+markphelps@users.noreply.github.com>

* chore: assert expected retries for fetch error

Signed-off-by: Mark Phelps <209477+markphelps@users.noreply.github.com>

* chore: add more tests and documentation

Signed-off-by: Mark Phelps <209477+markphelps@users.noreply.github.com>

* chore: java fmt

Signed-off-by: Mark Phelps <209477+markphelps@users.noreply.github.com>

---------

Signed-off-by: Mark Phelps <209477+markphelps@users.noreply.github.com>
---
 .gitignore                                    |   3 +-
 flipt-client-dart/lib/src/models.dart         |  16 +
 flipt-client-dart/lib/src/models.g.dart       |   8 +
 flipt-client-go/evaluation.go                 |  10 +
 flipt-client-go/models.go                     |  16 +-
 flipt-client-java/.idea/gradle.xml            |   2 +-
 flipt-client-java/.idea/misc.xml              |   2 +-
 .../flipt/client/FliptEvaluationClient.java   |  89 ++-
 .../io/flipt/client/models/ClientOptions.java |  10 +-
 .../io/flipt/client/models/FetchMode.java     |  10 +
 flipt-client-python/flipt_client/models.py    |   7 +
 flipt-client-ruby/.rubocop.yml                |   3 +
 flipt-client-ruby/Gemfile.lock                |   2 +-
 flipt-client-ruby/lib/flipt_client.rb         |   7 +
 flipt-engine-ffi/Cargo.toml                   |   7 +-
 flipt-engine-ffi/examples/evaluation.rs       |  32 +-
 flipt-engine-ffi/src/evaluator.rs             | 175 ++----
 flipt-engine-ffi/src/http.rs                  | 527 ++++++++++++++----
 flipt-engine-ffi/src/lib.rs                   | 107 +++-
 flipt-evaluation/src/lib.rs                   |   1 -
 flipt-evaluation/src/models/flipt.rs          |  16 +-
 flipt-evaluation/src/models/source.rs         |  22 +-
 flipt-evaluation/src/parser.rs                |  46 --
 flipt-evaluation/src/store.rs                 |  47 +-
 24 files changed, 791 insertions(+), 374 deletions(-)
 create mode 100644 flipt-client-java/src/main/java/io/flipt/client/models/FetchMode.java
 delete mode 100644 flipt-evaluation/src/parser.rs

diff --git a/.gitignore b/.gitignore
index 4f0ee923..2efb1967 100644
--- a/.gitignore
+++ b/.gitignore
@@ -17,4 +17,5 @@ Cargo.lock
 tmp
 .envrc
 .vscode
-staging
\ No newline at end of file
+staging
+*.h
\ No newline at end of file
diff --git a/flipt-client-dart/lib/src/models.dart b/flipt-client-dart/lib/src/models.dart
index bae4f07e..b9c3f821 100644
--- a/flipt-client-dart/lib/src/models.dart
+++ b/flipt-client-dart/lib/src/models.dart
@@ -2,6 +2,14 @@ import 'package:json_annotation/json_annotation.dart';
 
 part 'models.g.dart';
 
+enum FetchMode {
+  @JsonValue('polling')
+  polling,
+  @JsonValue('streaming')
+  streaming,
+}
+
+/// Options for the Flipt client
 @JsonSerializable()
 class Options {
   final String? url;
@@ -9,11 +17,15 @@ class Options {
   final int? updateInterval;
   final Map<String, dynamic>? authentication;
 
+  /// Note: Streaming is currently only supported when using the SDK with Flipt Cloud (https://flipt.io/cloud).
+  final FetchMode? fetchMode;
+
   Options({
     this.url = 'http://localhost:8080',
     this.reference,
     this.updateInterval = 120,
     this.authentication,
+    this.fetchMode = FetchMode.polling,
   });
 
   factory Options.fromJson(Map<String, dynamic> json) =>
@@ -25,6 +37,7 @@ class Options {
     String? url,
     String? reference,
     int? updateInterval,
+    FetchMode? fetchMode,
   }) {
     return Options(
       url: url,
@@ -33,6 +46,7 @@ class Options {
       authentication: {
         'client_token': token,
       },
+      fetchMode: fetchMode,
     );
   }
 
@@ -41,6 +55,7 @@ class Options {
     String? url,
     String? reference,
     int? updateInterval,
+    FetchMode? fetchMode,
   }) {
     return Options(
       url: url,
@@ -49,6 +64,7 @@ class Options {
       authentication: {
         'jwt_token': token,
       },
+      fetchMode: fetchMode,
     );
   }
 }
diff --git a/flipt-client-dart/lib/src/models.g.dart b/flipt-client-dart/lib/src/models.g.dart
index d34f22e8..9145d6ce 100644
--- a/flipt-client-dart/lib/src/models.g.dart
+++ b/flipt-client-dart/lib/src/models.g.dart
@@ -11,6 +11,8 @@ Options _$OptionsFromJson(Map<String, dynamic> json) => Options(
       reference: json['reference'] as String?,
       updateInterval: (json['update_interval'] as num?)?.toInt() ?? 120,
       authentication: json['authentication'] as Map<String, dynamic>?,
+      fetchMode: $enumDecodeNullable(_$FetchModeEnumMap, json['fetch_mode']) ??
+          FetchMode.polling,
     );
 
 Map<String, dynamic> _$OptionsToJson(Options instance) => <String, dynamic>{
@@ -18,8 +20,14 @@ Map<String, dynamic> _$OptionsToJson(Options instance) => <String, dynamic>{
       'reference': instance.reference,
       'update_interval': instance.updateInterval,
       'authentication': instance.authentication,
+      'fetch_mode': _$FetchModeEnumMap[instance.fetchMode],
     };
 
+const _$FetchModeEnumMap = {
+  FetchMode.polling: 'polling',
+  FetchMode.streaming: 'streaming',
+};
+
 Flag _$FlagFromJson(Map<String, dynamic> json) => Flag(
       key: json['key'] as String,
       enabled: json['enabled'] as bool,
diff --git a/flipt-client-go/evaluation.go b/flipt-client-go/evaluation.go
index 39c76fb5..f9bc2711 100644
--- a/flipt-client-go/evaluation.go
+++ b/flipt-client-go/evaluation.go
@@ -29,6 +29,7 @@ type EvaluationClient struct {
 	authentication any
 	ref            string
 	updateInterval int
+	fetchMode      FetchMode
 }
 
 // NewEvaluationClient constructs a Client.
@@ -46,6 +47,7 @@ func NewEvaluationClient(opts ...clientOption) (*EvaluationClient, error) {
 		UpdateInterval: client.updateInterval,
 		Authentication: &client.authentication,
 		Reference:      client.ref,
+		FetchMode:      client.fetchMode,
 	}
 
 	b, err := json.Marshal(clientOpts)
@@ -106,6 +108,14 @@ func WithJWTAuthentication(token string) clientOption {
 	}
 }
 
+// WithFetchMode allows for specifying the fetch mode for the Flipt client (e.g. polling, streaming).
+// Note: Streaming is currently only supported when using the SDK with Flipt Cloud (https://flipt.io/cloud).
+func WithFetchMode(fetchMode FetchMode) clientOption {
+	return func(c *EvaluationClient) {
+		c.fetchMode = fetchMode
+	}
+}
+
 // EvaluateVariant performs evaluation for a variant flag.
 func (e *EvaluationClient) EvaluateVariant(_ context.Context, flagKey, entityID string, evalContext map[string]string) (*VariantEvaluationResponse, error) {
 	ereq, err := json.Marshal(evaluationRequest{
diff --git a/flipt-client-go/models.go b/flipt-client-go/models.go
index 3a9ed577..e47622aa 100644
--- a/flipt-client-go/models.go
+++ b/flipt-client-go/models.go
@@ -21,11 +21,19 @@ type jwtAuthentication struct {
 	Token string `json:"jwt_token"`
 }
 
+type FetchMode string
+
+const (
+	FetchModeStreaming FetchMode = "streaming"
+	FetchModePolling   FetchMode = "polling"
+)
+
 type clientOptions[T any] struct {
-	URL            string `json:"url,omitempty"`
-	Authentication *T     `json:"authentication,omitempty"`
-	UpdateInterval int    `json:"update_interval,omitempty"`
-	Reference      string `json:"reference,omitempty"`
+	URL            string    `json:"url,omitempty"`
+	Authentication *T        `json:"authentication,omitempty"`
+	UpdateInterval int       `json:"update_interval,omitempty"`
+	Reference      string    `json:"reference,omitempty"`
+	FetchMode      FetchMode `json:"fetch_mode,omitempty"`
 }
 
 type Flag struct {
diff --git a/flipt-client-java/.idea/gradle.xml b/flipt-client-java/.idea/gradle.xml
index 2a65317e..1ace7cf4 100644
--- a/flipt-client-java/.idea/gradle.xml
+++ b/flipt-client-java/.idea/gradle.xml
@@ -5,7 +5,7 @@
     <option name="linkedExternalProjectsSettings">
       <GradleProjectSettings>
         <option name="externalProjectPath" value="$PROJECT_DIR$" />
-        <option name="gradleHome" value="" />
+        <option name="gradleJvm" value="11" />
         <option name="modules">
           <set>
             <option value="$PROJECT_DIR$" />
diff --git a/flipt-client-java/.idea/misc.xml b/flipt-client-java/.idea/misc.xml
index 2dae0dde..b3ca6f2d 100644
--- a/flipt-client-java/.idea/misc.xml
+++ b/flipt-client-java/.idea/misc.xml
@@ -4,7 +4,7 @@
   <component name="FrameworkDetectionExcludesConfiguration">
     <file type="web" url="file://$PROJECT_DIR$" />
   </component>
-  <component name="ProjectRootManager" version="2" languageLevel="JDK_1_8" project-jdk-name="21" project-jdk-type="JavaSDK">
+  <component name="ProjectRootManager" version="2" languageLevel="JDK_1_8" project-jdk-name="11" project-jdk-type="JavaSDK">
     <output url="file://$PROJECT_DIR$/out" />
   </component>
 </project>
\ No newline at end of file
diff --git a/flipt-client-java/src/main/java/io/flipt/client/FliptEvaluationClient.java b/flipt-client-java/src/main/java/io/flipt/client/FliptEvaluationClient.java
index beaea84f..1d0b99d4 100644
--- a/flipt-client-java/src/main/java/io/flipt/client/FliptEvaluationClient.java
+++ b/flipt-client-java/src/main/java/io/flipt/client/FliptEvaluationClient.java
@@ -38,7 +38,7 @@ public interface CLibrary extends Library {
     void destroy_string(Pointer str);
   }
 
-  private FliptEvaluationClient(String namespace, ClientOptions clientOpts)
+  private FliptEvaluationClient(String namespace, ClientOptions clientOptions)
       throws EvaluationException {
 
     ObjectMapper objectMapper = new ObjectMapper();
@@ -46,7 +46,7 @@ private FliptEvaluationClient(String namespace, ClientOptions clientOpts)
 
     String clientOptionsSerialized;
     try {
-      clientOptionsSerialized = objectMapper.writeValueAsString(clientOpts);
+      clientOptionsSerialized = objectMapper.writeValueAsString(clientOptions);
     } catch (JsonProcessingException e) {
       throw new EvaluationException(e);
     }
@@ -61,40 +61,90 @@ public static FliptEvaluationClientBuilder builder() {
     return new FliptEvaluationClientBuilder();
   }
 
+  /** FliptEvaluationClientBuilder is a builder for creating a FliptEvaluationClient. */
   public static final class FliptEvaluationClientBuilder {
     private String namespace = "default";
     private String url = "http://localhost:8080";
     private AuthenticationStrategy authentication;
     private String reference;
     private Duration updateInterval;
+    private FetchMode fetchMode = FetchMode.POLLING;
 
     public FliptEvaluationClientBuilder() {}
 
+    /**
+     * url sets the URL for the Flipt server.
+     *
+     * @param url the URL for the Flipt server
+     * @return the FliptEvaluationClientBuilder
+     */
     public FliptEvaluationClientBuilder url(String url) {
       this.url = url;
       return this;
     }
 
+    /**
+     * namespace sets the namespace for the Flipt server.
+     *
+     * @param namespace the namespace for the Flipt server
+     * @return the FliptEvaluationClientBuilder
+     */
     public FliptEvaluationClientBuilder namespace(String namespace) {
       this.namespace = namespace;
       return this;
     }
 
+    /**
+     * authentication sets the authentication strategy for the Flipt server.
+     *
+     * @param authentication the authentication strategy for the Flipt server
+     * @return the FliptEvaluationClientBuilder
+     */
     public FliptEvaluationClientBuilder authentication(AuthenticationStrategy authentication) {
       this.authentication = authentication;
       return this;
     }
 
+    /**
+     * updateInterval sets the update interval for the Flipt server.
+     *
+     * @param updateInterval the update interval for the Flipt server
+     * @return the FliptEvaluationClientBuilder
+     */
     public FliptEvaluationClientBuilder updateInterval(Duration updateInterval) {
       this.updateInterval = updateInterval;
       return this;
     }
 
+    /**
+     * reference sets the reference for the Flipt server.
+     *
+     * @param reference the reference for the Flipt server
+     * @return the FliptEvaluationClientBuilder
+     */
     public FliptEvaluationClientBuilder reference(String reference) {
       this.reference = reference;
       return this;
     }
 
+    /**
+     * fetchMode sets the fetch mode for the Flipt server. Note: Streaming is currently only
+     * supported when using the SDK with Flipt Cloud (https://flipt.io/cloud).
+     *
+     * @param fetchMode the fetch mode for the Flipt server
+     * @return the FliptEvaluationClientBuilder
+     */
+    public FliptEvaluationClientBuilder fetchMode(FetchMode fetchMode) {
+      this.fetchMode = fetchMode;
+      return this;
+    }
+
+    /**
+     * build builds a new FliptEvaluationClient.
+     *
+     * @return the FliptEvaluationClient
+     * @throws EvaluationException if the FliptEvaluationClient could not be built
+     */
     public FliptEvaluationClient build() throws EvaluationException {
       return new FliptEvaluationClient(
           namespace,
@@ -102,7 +152,8 @@ public FliptEvaluationClient build() throws EvaluationException {
               Optional.of(url),
               Optional.ofNullable(updateInterval),
               Optional.ofNullable(authentication),
-              Optional.ofNullable(reference)));
+              Optional.ofNullable(reference),
+              Optional.ofNullable(fetchMode)));
     }
   }
 
@@ -136,6 +187,15 @@ public Map<String, String> getContext() {
     }
   }
 
+  /**
+   * evaluateVariant evaluates a variant flag.
+   *
+   * @param flagKey the key for the flag to evaluate
+   * @param entityId the ID for the entity to evaluate
+   * @param context the context for the evaluation
+   * @return the evaluation response
+   * @throws EvaluationException if the evaluation failed
+   */
   public VariantEvaluationResponse evaluateVariant(
       String flagKey, String entityId, Map<String, String> context) throws EvaluationException {
     InternalEvaluationRequest evaluationRequest =
@@ -160,6 +220,15 @@ public VariantEvaluationResponse evaluateVariant(
     return resp.getResult().get();
   }
 
+  /**
+   * evaluateBoolean evaluates a boolean flag.
+   *
+   * @param flagKey the key for the flag to evaluate
+   * @param entityId the ID for the entity to evaluate
+   * @param context the context for the evaluation
+   * @return the evaluation response
+   * @throws EvaluationException if the evaluation failed
+   */
   public BooleanEvaluationResponse evaluateBoolean(
       String flagKey, String entityId, Map<String, String> context) throws EvaluationException {
     InternalEvaluationRequest evaluationRequest =
@@ -183,6 +252,13 @@ public BooleanEvaluationResponse evaluateBoolean(
     return resp.getResult().get();
   }
 
+  /**
+   * evaluateBatch evaluates a batch of flags.
+   *
+   * @param batchEvaluationRequests the batch of flags to evaluate
+   * @return the evaluation response
+   * @throws EvaluationException if the evaluation failed
+   */
   public BatchEvaluationResponse evaluateBatch(EvaluationRequest[] batchEvaluationRequests)
       throws EvaluationException {
     ArrayList<InternalEvaluationRequest> evaluationRequests =
@@ -216,6 +292,12 @@ public BatchEvaluationResponse evaluateBatch(EvaluationRequest[] batchEvaluation
     return resp.getResult().get();
   }
 
+  /**
+   * listFlags lists all flags in the namespace.
+   *
+   * @return the list of flags
+   * @throws EvaluationException if the list of flags could not be retrieved
+   */
   public ArrayList<Flag> listFlags() throws EvaluationException {
     Pointer value = CLibrary.INSTANCE.list_flags(this.engine);
 
@@ -229,6 +311,7 @@ public ArrayList<Flag> listFlags() throws EvaluationException {
     return resp.getResult().get();
   }
 
+  /** close closes the FliptEvaluationClient. */
   public void close() {
     CLibrary.INSTANCE.destroy_engine(this.engine);
   }
diff --git a/flipt-client-java/src/main/java/io/flipt/client/models/ClientOptions.java b/flipt-client-java/src/main/java/io/flipt/client/models/ClientOptions.java
index 36237eb7..abfd8dd4 100644
--- a/flipt-client-java/src/main/java/io/flipt/client/models/ClientOptions.java
+++ b/flipt-client-java/src/main/java/io/flipt/client/models/ClientOptions.java
@@ -11,12 +11,14 @@ public class ClientOptions {
   private final Optional<Long> updateInterval;
   private final Optional<AuthenticationStrategy> authentication;
   private final Optional<String> reference;
+  private final Optional<FetchMode> fetchMode;
 
   public ClientOptions(
       Optional<String> url,
       Optional<Duration> updateInterval,
       Optional<AuthenticationStrategy> authentication,
-      Optional<String> reference) {
+      Optional<String> reference,
+      Optional<FetchMode> fetchMode) {
     this.url = url;
     this.authentication = authentication;
     this.reference = reference;
@@ -27,6 +29,7 @@ public ClientOptions(
     }
 
     this.updateInterval = setUpdateInterval;
+    this.fetchMode = fetchMode;
   }
 
   @JsonProperty("url")
@@ -48,4 +51,9 @@ public Optional<AuthenticationStrategy> getAuthentication() {
   public Optional<String> getReference() {
     return reference;
   }
+
+  @JsonProperty("fetch_mode")
+  public Optional<FetchMode> getFetchMode() {
+    return fetchMode;
+  }
 }
diff --git a/flipt-client-java/src/main/java/io/flipt/client/models/FetchMode.java b/flipt-client-java/src/main/java/io/flipt/client/models/FetchMode.java
new file mode 100644
index 00000000..377ca1dd
--- /dev/null
+++ b/flipt-client-java/src/main/java/io/flipt/client/models/FetchMode.java
@@ -0,0 +1,10 @@
+package io.flipt.client.models;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+public enum FetchMode {
+  @JsonProperty("polling")
+  POLLING,
+  @JsonProperty("streaming")
+  STREAMING
+}
diff --git a/flipt-client-python/flipt_client/models.py b/flipt-client-python/flipt_client/models.py
index b66d3b6a..5acb28e2 100644
--- a/flipt-client-python/flipt_client/models.py
+++ b/flipt-client-python/flipt_client/models.py
@@ -1,3 +1,4 @@
+from enum import Enum
 from pydantic import BaseModel, RootModel
 from typing import List, Optional
 
@@ -21,11 +22,17 @@ class JWTAuthentication(AuthenticationStrategy):
     jwt_token: str
 
 
+class FetchMode(Enum):
+    POLLING = "polling"
+    STREAMING = "streaming"
+
+
 class ClientOptions(BaseModel):
     url: Optional[str] = None
     update_interval: Optional[int] = None
     authentication: Optional[AuthenticationStrategy] = None
     reference: Optional[str] = None
+    fetch_mode: Optional[FetchMode] = None
 
 
 class VariantEvaluationResponse(BaseModel):
diff --git a/flipt-client-ruby/.rubocop.yml b/flipt-client-ruby/.rubocop.yml
index 460a64d9..d20805ec 100644
--- a/flipt-client-ruby/.rubocop.yml
+++ b/flipt-client-ruby/.rubocop.yml
@@ -27,3 +27,6 @@ Style/Documentation:
   Exclude:
     - "spec/**/*"
     - "test/**/*"
+
+Style/IfUnlessModifier:
+  Enabled: false
\ No newline at end of file
diff --git a/flipt-client-ruby/Gemfile.lock b/flipt-client-ruby/Gemfile.lock
index 06286831..331f4b82 100644
--- a/flipt-client-ruby/Gemfile.lock
+++ b/flipt-client-ruby/Gemfile.lock
@@ -1,7 +1,7 @@
 PATH
   remote: .
   specs:
-    flipt_client (0.5.0)
+    flipt_client (0.10.0)
 
 GEM
   remote: https://rubygems.org/
diff --git a/flipt-client-ruby/lib/flipt_client.rb b/flipt-client-ruby/lib/flipt_client.rb
index d20eb783..ba9f71b3 100644
--- a/flipt-client-ruby/lib/flipt_client.rb
+++ b/flipt-client-ruby/lib/flipt_client.rb
@@ -55,6 +55,8 @@ def self.libfile
     # @option opts [AuthenticationStrategy] :authentication strategy to authenticate with Flipt
     # @option opts [Integer] :update_interval interval in seconds to update the cache
     # @option opts [String] :reference reference to use for namespace data
+    # @option opts [Symbol] :fetch_mode fetch mode to use for the client (:polling or :streaming).
+    #   Note: Streaming is currently only supported when using the SDK with Flipt Cloud (https://flipt.io/cloud).
     def initialize(namespace = 'default', opts = {})
       @namespace = namespace
 
@@ -65,6 +67,11 @@ def initialize(namespace = 'default', opts = {})
               'invalid authentication strategy'
       end
 
+      fetch_mode = opts.fetch(:fetch_mode, :polling)
+      unless %i[polling streaming].include?(fetch_mode)
+        raise ArgumentError, 'invalid fetch mode'
+      end
+
       opts[:authentication] = authentication.strategy
 
       @engine = self.class.initialize_engine(namespace, opts.to_json)
diff --git a/flipt-engine-ffi/Cargo.toml b/flipt-engine-ffi/Cargo.toml
index 22db8aa0..90d5c0b1 100644
--- a/flipt-engine-ffi/Cargo.toml
+++ b/flipt-engine-ffi/Cargo.toml
@@ -12,10 +12,14 @@ repository = "https://github.com/flipt-io/flipt-client-sdks"
 libc = "0.2.150"
 serde = { version = "1.0.147", features = ["derive"] }
 serde_json = { version = "1.0.89", features = ["raw_value"] }
-reqwest = { version = "0.12.2", features = ["json", "blocking"] }
+reqwest = { version = "0.12.2", features = ["json", "stream"] }
+tokio = { version = "1.36.0", features = ["full"] }
+futures-util = "0.3.30"
 native-tls = { version = "0.2", features = ["vendored"] }
 futures = "0.3"
 thiserror = "1.0.63"
+reqwest-retry = "0.6.1"
+reqwest-middleware = "0.3.3"
 
 [dependencies.flipt-evaluation]
 path = "../flipt-evaluation"
@@ -27,4 +31,3 @@ mockito = "1.4.0"
 [lib]
 name = "fliptengine"
 crate-type = ["rlib", "dylib", "staticlib"]
-
diff --git a/flipt-engine-ffi/examples/evaluation.rs b/flipt-engine-ffi/examples/evaluation.rs
index d4f747a2..cd17b0a7 100644
--- a/flipt-engine-ffi/examples/evaluation.rs
+++ b/flipt-engine-ffi/examples/evaluation.rs
@@ -2,34 +2,28 @@
 
 use fliptengine::{
     evaluator::Evaluator,
-    http::{Authentication, HTTPParserBuilder},
+    http::{Authentication, HTTPFetcherBuilder},
 };
 use fliptevaluation::EvaluationRequest;
 use std::collections::HashMap;
 
 fn main() {
-    let evaluator = Evaluator::new_snapshot_evaluator(
-        "default".into(),
-        HTTPParserBuilder::new("http://localhost:8080")
-            .authentication(Authentication::with_client_token("secret".into()))
-            .build(),
-    )
-    .unwrap();
+    let namespace = "default";
+    let fetcher = HTTPFetcherBuilder::new("http://localhost:8080", namespace)
+        .authentication(Authentication::with_client_token("secret".into()))
+        .build();
+    let evaluator = Evaluator::new(namespace);
 
-    let eng = fliptengine::Engine::new(evaluator, Default::default());
+    let engine = fliptengine::Engine::new(namespace, fetcher, evaluator);
     let mut context: HashMap<String, String> = HashMap::new();
     context.insert("fizz".into(), "buzz".into());
 
-    let thread = std::thread::spawn(move || loop {
-        std::thread::sleep(std::time::Duration::from_millis(5000));
-        let variant = eng.variant(&EvaluationRequest {
-            flag_key: "flag1".into(),
-            entity_id: "entity".into(),
-            context: context.clone(),
-        });
-
-        println!("variant key {:?}", variant.unwrap().variant_key);
+    std::thread::sleep(std::time::Duration::from_millis(5000));
+    let variant = engine.variant(&EvaluationRequest {
+        flag_key: "flag1".into(),
+        entity_id: "entity".into(),
+        context: context.clone(),
     });
 
-    thread.join().expect("current thread panicked");
+    println!("variant key {:?}", variant.unwrap().variant_key);
 }
diff --git a/flipt-engine-ffi/src/evaluator.rs b/flipt-engine-ffi/src/evaluator.rs
index 171fbdfa..8098ce0f 100644
--- a/flipt-engine-ffi/src/evaluator.rs
+++ b/flipt-engine-ffi/src/evaluator.rs
@@ -3,86 +3,49 @@ use std::sync::{Arc, RwLock};
 use fliptevaluation::{
     batch_evaluation, boolean_evaluation,
     error::Error,
-    models::{flipt, source},
-    parser::Parser,
+    models::flipt,
     store::{Snapshot, Store},
     variant_evaluation, BatchEvaluationResponse, BooleanEvaluationResponse, EvaluationRequest,
     VariantEvaluationResponse,
 };
 
-pub struct Evaluator<P, S>
+pub struct Evaluator<S>
 where
-    P: Parser + Send,
     S: Store + Send,
 {
     namespace: String,
-    parser: P,
     store: S,
     mtx: Arc<RwLock<i32>>,
     error: Option<Error>,
 }
 
-impl<P> Evaluator<P, Snapshot>
-where
-    P: Parser + Send,
-{
-    pub fn new_snapshot_evaluator(namespace: &str, parser: P) -> Result<Self, Error> {
-        let snap = Snapshot::build(namespace, source::Document::default())?;
-        let mut e = Evaluator::new(namespace, parser, snap);
-        e.replace_snapshot();
-        Ok(e)
-    }
-
-    pub fn replace_snapshot(&mut self) {
-        match self.parser.parse(&self.namespace) {
-            Ok(doc) => {
-                // if doc is none then return, nothing to do
-                if doc.is_none() {
-                    return;
-                }
-                match Snapshot::build(&self.namespace, doc.unwrap()) {
-                    Ok(s) => {
-                        self.replace_store(s, None);
-                    }
-                    Err(err) => {
-                        // TODO: log::error!("error building snapshot: {}", e);
-                        self.replace_store(Snapshot::empty(&self.namespace), Some(err))
-                    }
-                };
-            }
-            Err(err) => {
-                // TODO: log::error!("error parsing document: {}"", e);
-                self.replace_store(Snapshot::empty(&self.namespace), Some(err))
-            }
-        };
-    }
-}
-
-impl<P, S> Evaluator<P, S>
-where
-    P: Parser + Send,
-    S: Store + Send,
-{
-    pub fn new(namespace: &str, parser: P, store: S) -> Self {
+impl Evaluator<Snapshot> {
+    pub fn new(namespace: &str) -> Self {
+        let snap = Snapshot::empty(namespace);
         Self {
             namespace: namespace.to_string(),
-            parser,
-            store,
+            store: snap,
             mtx: Arc::new(RwLock::new(0)),
             error: None,
         }
     }
 
-    pub fn replace_store(&mut self, store: S, err: Option<Error>) {
+    pub fn replace_snapshot(&mut self, res: Result<Snapshot, Error>) {
         let _w_lock = self.mtx.write().unwrap();
-        self.store = store;
-        self.error = err;
+        match res {
+            Ok(snap) => {
+                self.store = snap;
+                self.error = None;
+            }
+            Err(err) => {
+                self.error = Some(err);
+            }
+        }
     }
 
     pub fn list_flags(&self) -> Result<Vec<flipt::Flag>, Error> {
         let _r_lock = self.mtx.read().unwrap();
-        if self.error.is_some() {
-            let error = self.error.as_ref().expect("valid error");
+        if let Some(error) = &self.error {
             return Err(error.clone());
         }
         match self.store.list_flags(&self.namespace) {
@@ -99,8 +62,7 @@ where
         evaluation_request: &EvaluationRequest,
     ) -> Result<VariantEvaluationResponse, Error> {
         let _r_lock = self.mtx.read().unwrap();
-        if self.error.is_some() {
-            let error = self.error.as_ref().expect("valid error");
+        if let Some(error) = &self.error {
             return Err(error.clone());
         }
         variant_evaluation(&self.store, &self.namespace, evaluation_request)
@@ -111,8 +73,7 @@ where
         evaluation_request: &EvaluationRequest,
     ) -> Result<BooleanEvaluationResponse, Error> {
         let _r_lock = self.mtx.read().unwrap();
-        if self.error.is_some() {
-            let error = self.error.as_ref().expect("valid error");
+        if let Some(error) = &self.error {
             return Err(error.clone());
         }
         boolean_evaluation(&self.store, &self.namespace, evaluation_request)
@@ -123,8 +84,7 @@ where
         requests: Vec<EvaluationRequest>,
     ) -> Result<BatchEvaluationResponse, Error> {
         let _r_lock = self.mtx.read().unwrap();
-        if self.error.is_some() {
-            let error = self.error.as_ref().expect("valid error");
+        if let Some(error) = &self.error {
             return Err(error.clone());
         }
         batch_evaluation(&self.store, &self.namespace, requests)
@@ -136,17 +96,7 @@ mod tests {
     use std::collections::HashMap;
 
     use super::*;
-    use fliptevaluation::models::source;
-    use fliptevaluation::parser;
     use mockall::predicate::*;
-    use mockall::*;
-
-    mock! {
-        pub Parser{}
-        impl parser::Parser for Parser {
-            fn parse(&mut self, namespace: &str) -> Result<Option<source::Document>, Error>;
-        }
-    }
 
     fn assert_error_response<T>(response: Result<T, Error>, expected_error: &str) {
         match response {
@@ -160,39 +110,8 @@ mod tests {
     }
 
     #[test]
-    fn test_parser_with_error() {
-        let expected_error = "server error: can't connect";
-        let mut parser = MockParser::new();
-        parser
-            .expect_parse()
-            .returning(|_| Err(Error::Server("can't connect".to_owned())));
-        let evaluator =
-            Evaluator::new_snapshot_evaluator("namespace", parser).expect("expect valid evaluator");
-
-        let response = evaluator.list_flags();
-        assert_error_response(response, expected_error);
-        let requests = vec![];
-        let response = evaluator.batch(requests);
-        assert_error_response(response, expected_error);
-        let request = &EvaluationRequest {
-            flag_key: String::from("foo"),
-            entity_id: String::from("user@flipt.io"),
-            context: HashMap::new(),
-        };
-        let response = evaluator.boolean(request);
-        assert_error_response(response, expected_error);
-        let response = evaluator.variant(request);
-        assert_error_response(response, expected_error);
-    }
-
-    #[test]
-    fn test_parser_with_empty_snapshot() {
-        let mut parser = MockParser::new();
-        parser
-            .expect_parse()
-            .returning(|_| Ok(Some(source::Document::default())));
-        let evaluator =
-            Evaluator::new_snapshot_evaluator("namespace", parser).expect("expect valid evaluator");
+    fn test_empty_snapshot() {
+        let evaluator = Evaluator::new("namespace");
 
         let response = evaluator.list_flags();
         match response {
@@ -227,32 +146,38 @@ mod tests {
     }
 
     #[test]
-    fn test_parser_with_no_snapshot() {
-        let mut parser = MockParser::new();
-        parser.expect_parse().returning(|_| Ok(None));
-        let evaluator =
-            Evaluator::new_snapshot_evaluator("namespace", parser).expect("expect valid evaluator");
-
-        let response = evaluator.list_flags();
-        assert!(response.is_ok());
+    fn test_replace_snapshot() {
+        let mut evaluator = Evaluator::new("namespace");
+        let snapshot = Snapshot::empty("namespace");
+        evaluator.replace_snapshot(Ok(snapshot.clone()));
+        assert_eq!(evaluator.store, snapshot);
     }
 
     #[test]
-    fn test_list_flags_from_another_namespace() {
-        let mut parser = MockParser::new();
-        parser
-            .expect_parse()
-            .returning(|_| Ok(Some(source::Document::default())));
-        let mut evaluator =
-            Evaluator::new_snapshot_evaluator("namespace", parser).expect("expect valid evaluator");
-        evaluator.replace_store(Snapshot::empty("other"), None);
+    fn test_replace_snapshot_error() {
+        let mut evaluator = Evaluator::new("namespace");
+        evaluator.replace_snapshot(Err(Error::Unknown("error".to_string())));
+
         let response = evaluator.list_flags();
-        match response {
-            Err(err) => assert_eq!(
-                "unknown error: failed to get flags for namespace",
-                err.to_string()
-            ),
-            Ok(_) => panic!("unexpected error"),
+        assert_error_response(response, "unknown error: error");
+
+        let request = &EvaluationRequest {
+            flag_key: String::from("foo"),
+            entity_id: String::from("user@flipt.io"),
+            context: HashMap::new(),
         };
+        let response = evaluator.boolean(request);
+        assert_error_response(response, "unknown error: error");
+
+        let response = evaluator.variant(request);
+        assert_error_response(response, "unknown error: error");
+
+        let requests = vec![EvaluationRequest {
+            flag_key: String::from("foo"),
+            entity_id: String::from("user@flipt.io"),
+            context: HashMap::new(),
+        }];
+        let response = evaluator.batch(requests);
+        assert_error_response(response, "unknown error: error");
     }
 }
diff --git a/flipt-engine-ffi/src/http.rs b/flipt-engine-ffi/src/http.rs
index 5ae74d86..ec530141 100644
--- a/flipt-engine-ffi/src/http.rs
+++ b/flipt-engine-ffi/src/http.rs
@@ -1,9 +1,19 @@
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::Arc;
+use std::time::Duration;
+
+use futures_util::stream::StreamExt;
 use reqwest::header::{self, HeaderMap};
+use reqwest::Response;
+use reqwest_middleware::{ClientBuilder, ClientWithMiddleware};
+use reqwest_retry::policies::ExponentialBackoff;
+use reqwest_retry::RetryTransientMiddleware;
 use serde::Deserialize;
+use tokio::runtime::Handle;
+use tokio::sync::mpsc;
 
 use fliptevaluation::error::Error;
 use fliptevaluation::models::source;
-use fliptevaluation::parser::Parser;
 
 #[derive(Debug, Clone, Default, Deserialize)]
 #[cfg_attr(test, derive(PartialEq))]
@@ -57,26 +67,56 @@ impl From<Authentication> for HeaderMap {
     }
 }
 
-pub struct HTTPParser {
-    http_client: reqwest::blocking::Client,
-    http_url: String,
+#[derive(Debug, Clone, Default, Deserialize)]
+#[cfg_attr(test, derive(PartialEq))]
+#[serde(rename_all = "snake_case")]
+#[non_exhaustive]
+pub enum FetchMode {
+    #[default]
+    Polling,
+    Streaming,
+}
+
+#[derive(Clone)]
+pub struct HTTPFetcher {
+    http_client: ClientWithMiddleware,
+    base_url: String,
+    namespace: String,
     authentication: HeaderMap,
-    reference: Option<String>,
     etag: Option<String>,
+    reference: Option<String>,
+    update_interval: Duration,
+    mode: FetchMode,
 }
 
-pub struct HTTPParserBuilder {
-    http_url: String,
+pub struct HTTPFetcherBuilder {
+    base_url: String,
+    namespace: String,
     authentication: HeaderMap,
     reference: Option<String>,
+    update_interval: Duration,
+    mode: FetchMode,
+}
+
+#[derive(Deserialize)]
+struct StreamChunk {
+    result: StreamResult,
+}
+
+#[derive(Deserialize)]
+struct StreamResult {
+    namespaces: std::collections::HashMap<String, source::Document>,
 }
 
-impl HTTPParserBuilder {
-    pub fn new(http_url: &str) -> Self {
+impl HTTPFetcherBuilder {
+    pub fn new(base_url: &str, namespace: &str) -> Self {
         Self {
-            http_url: http_url.to_string(),
+            base_url: base_url.to_string(),
+            namespace: namespace.to_string(),
             authentication: HeaderMap::new(),
             reference: None,
+            update_interval: Duration::from_secs(120),
+            mode: FetchMode::default(),
         }
     }
 
@@ -90,53 +130,93 @@ impl HTTPParserBuilder {
         self
     }
 
-    pub fn build(self) -> HTTPParser {
-        HTTPParser {
-            http_client: reqwest::blocking::Client::builder()
-                .timeout(std::time::Duration::from_secs(10))
-                .build()
-                .unwrap(),
-            http_url: self.http_url,
+    pub fn update_interval(mut self, update_interval: Duration) -> Self {
+        self.update_interval = update_interval;
+        self
+    }
+
+    pub fn mode(mut self, mode: FetchMode) -> Self {
+        self.mode = mode;
+        self
+    }
+
+    pub fn build(self) -> HTTPFetcher {
+        let retry_policy = ExponentialBackoff::builder().build_with_max_retries(3);
+        HTTPFetcher {
+            base_url: self.base_url,
+            namespace: self.namespace,
+            http_client: ClientBuilder::new(reqwest::Client::new())
+                .with(RetryTransientMiddleware::new_with_policy(retry_policy))
+                .build(),
             authentication: self.authentication,
-            reference: self.reference,
             etag: None,
+            reference: self.reference,
+            update_interval: self.update_interval,
+            mode: self.mode,
         }
     }
 }
 
-impl HTTPParser {
-    fn url(&self, namespace: &str) -> String {
-        match &self.reference {
-            Some(reference) => {
-                format!(
-                    "{}/internal/v1/evaluation/snapshot/namespace/{}?reference={}",
-                    self.http_url, namespace, reference,
-                )
+type FetchResult = Result<source::Document, Error>;
+
+impl HTTPFetcher {
+    /// Start the fetcher and return a channel to receive updates on the snapshot changes
+    pub fn start(&mut self, stop_signal: Arc<AtomicBool>) -> mpsc::Receiver<FetchResult> {
+        let (tx, rx) = mpsc::channel(100);
+
+        let mut fetcher = self.clone();
+
+        tokio::spawn(async move {
+            while !stop_signal.load(Ordering::Relaxed) {
+                match fetcher.mode {
+                    FetchMode::Polling => {
+                        if fetcher.handle_polling(&tx).await.is_err() {
+                            // TODO: log error
+                            break;
+                        }
+                        tokio::time::sleep(fetcher.update_interval).await;
+                    }
+                    FetchMode::Streaming => {
+                        if fetcher.handle_streaming(&tx).await.is_err() {
+                            // TODO: log error
+                            break;
+                        }
+                    }
+                }
             }
-            None => {
-                format!(
-                    "{}/internal/v1/evaluation/snapshot/namespace/{}",
-                    self.http_url, namespace
-                )
+            // The channel will be closed when tx is dropped at the end of this function
+        });
+
+        rx
+    }
+
+    pub fn initial_fetch(&mut self) -> FetchResult {
+        match Handle::try_current() {
+            Ok(handle) => {
+                // We're already in a Tokio runtime, use the existing one
+                handle.block_on(self.initial_fetch_async())
+            }
+            Err(_) => {
+                // No runtime exists, create a new one
+                let runtime = tokio::runtime::Runtime::new()
+                    .map_err(|e| Error::Server(format!("failed to create runtime: {}", e)))?;
+                runtime.block_on(self.initial_fetch_async())
             }
         }
     }
-}
 
-impl Parser for HTTPParser {
-    fn parse(&mut self, namespace: &str) -> Result<Option<source::Document>, Error> {
+    fn build_headers(&self) -> HeaderMap {
         let mut headers = HeaderMap::new();
         headers.insert(
             reqwest::header::ACCEPT,
             reqwest::header::HeaderValue::from_static("application/json"),
         );
-        // version (or higher) that we can accept from the server
+
         headers.insert(
             "X-Flipt-Accept-Server-Version",
             reqwest::header::HeaderValue::from_static("1.47.0"),
         );
 
-        // add etag / if-none-match header if we have one
         if let Some(etag) = &self.etag {
             headers.insert(
                 reqwest::header::IF_NONE_MATCH,
@@ -148,95 +228,278 @@ impl Parser for HTTPParser {
             headers.insert(key, value.clone());
         }
 
-        let response = match self
+        headers
+    }
+
+    async fn fetch(&mut self) -> Result<Option<Response>, Error> {
+        let url = match &self.reference {
+            Some(reference) => {
+                format!(
+                    "{}/internal/v1/evaluation/snapshot/namespace/{}?reference={}",
+                    self.base_url, self.namespace, reference
+                )
+            }
+            None => {
+                format!(
+                    "{}/internal/v1/evaluation/snapshot/namespace/{}",
+                    self.base_url, self.namespace
+                )
+            }
+        };
+
+        let response = self
             .http_client
-            .get(self.url(namespace))
-            .headers(headers)
+            .get(url)
+            .headers(self.build_headers())
             .send()
-        {
-            Ok(resp) => match resp.error_for_status() {
-                Ok(resp) => resp,
-                Err(e) => return Err(Error::Server(format!("response: {}", e))),
-            },
-            Err(e) => return Err(Error::Server(format!("failed to make request: {}", e))),
-        };
+            .await
+            .map_err(|e| Error::Server(format!("failed to make request: {}", e)))?
+            .error_for_status()
+            .map_err(|e| Error::Server(format!("response: {}", e)))?;
+
         match response.status() {
-            // check if we have a 304 response
             reqwest::StatusCode::NOT_MODIFIED => Ok(None),
+
             reqwest::StatusCode::OK => {
                 // check if we have a new etag
                 if let Some(etag) = response.headers().get(reqwest::header::ETAG) {
                     self.etag = Some(etag.to_str().unwrap().to_string());
                 }
 
-                let response_text = match response.text() {
-                    Ok(t) => t,
-                    Err(e) => {
-                        return Err(Error::Server(format!("failed to get response body: {}", e)))
-                    }
-                };
-
-                match serde_json::from_str(&response_text) {
-                    Ok(doc) => Ok(Some(doc)),
-                    Err(e) => Err(Error::InvalidJSON(e.to_string())),
-                }
+                Ok(Some(response))
             }
+
             _ => Err(Error::Server(format!(
                 "unexpected http response: {} {}",
                 response.status(),
-                response.text().unwrap_or("".to_string())
+                response.text().await.unwrap_or("".to_string())
             ))),
         }
     }
+
+    async fn fetch_stream(&mut self) -> Result<Option<Response>, Error> {
+        let url = format!(
+            "{}/internal/v1/evaluation/snapshots?[]namespaces={}",
+            self.base_url, self.namespace
+        );
+
+        self.http_client
+            .get(url)
+            .headers(self.build_headers())
+            .send()
+            .await
+            .map_err(|e| Error::Server(format!("failed to make request: {}", e)))?
+            .error_for_status()
+            .map_err(|e| Error::Server(format!("response: {}", e)))
+            .map(Some)
+    }
+
+    async fn handle_polling(
+        &mut self,
+        sender: &mpsc::Sender<Result<source::Document, Error>>,
+    ) -> Result<(), Error> {
+        let result = match self.fetch().await {
+            Ok(Some(resp)) => match resp.text().await {
+                Ok(json) => match serde_json::from_str(&json) {
+                    Ok(doc) => Ok(doc),
+                    Err(e) => Err(Error::InvalidJSON(format!(
+                        "failed to parse response body: {}",
+                        e
+                    ))),
+                },
+                Err(e) => Err(Error::Server(format!(
+                    "failed to read response body: {}",
+                    e
+                ))),
+            },
+            Ok(None) => return Ok(()),
+            Err(e) => Err(e),
+        };
+
+        sender
+            .send(result)
+            .await
+            .map_err(|_| Error::Server("failed to send result".into()))
+    }
+
+    async fn handle_streaming(
+        &mut self,
+        sender: &mpsc::Sender<Result<source::Document, Error>>,
+    ) -> Result<(), Error> {
+        let stream_result = self.fetch_stream().await;
+
+        match stream_result {
+            Ok(Some(resp)) => {
+                let mut stream = resp.bytes_stream();
+                let mut buffer = Vec::new();
+
+                while let Some(bytes) = stream.next().await {
+                    let chunk = match bytes {
+                        Ok(chunk) => chunk,
+                        Err(e) => {
+                            let err = Error::Server(format!("failed to read stream chunk: {}", e));
+                            sender
+                                .send(Err(err.clone()))
+                                .await
+                                .map_err(|_| Error::Server("failed to send error".into()))?;
+                            continue; // Continue to the next iteration instead of returning
+                        }
+                    };
+
+                    for byte in chunk {
+                        if byte == b'\n' {
+                            let text = String::from_utf8_lossy(&buffer);
+                            let parse_result: Result<StreamChunk, Error> =
+                                serde_json::from_str(&text).map_err(|e| {
+                                    Error::InvalidJSON(format!(
+                                        "failed to parse response body: {}",
+                                        e
+                                    ))
+                                });
+
+                            match parse_result {
+                                Ok(stream_chunk) => {
+                                    match stream_chunk.result.namespaces.into_iter().next() {
+                                        Some((_, doc)) => {
+                                            sender.send(Ok(doc)).await.map_err(|_| {
+                                                Error::Server("failed to send result".into())
+                                            })?;
+                                        }
+                                        None => {
+                                            let err = Error::Server(
+                                                "no data received from server".into(),
+                                            );
+                                            sender.send(Err(err)).await.map_err(|_| {
+                                                Error::Server("failed to send error".into())
+                                            })?;
+                                        }
+                                    }
+                                }
+                                Err(e) => {
+                                    sender.send(Err(e)).await.map_err(|_| {
+                                        Error::Server("failed to send error".into())
+                                    })?;
+                                }
+                            }
+                            buffer.clear();
+                        } else {
+                            buffer.push(byte);
+                        }
+                    }
+                }
+                Ok(())
+            }
+            Ok(None) => Ok(()),
+            Err(e) => {
+                sender
+                    .send(Err(e.clone()))
+                    .await
+                    .map_err(|_| Error::Server("failed to send error".into()))?;
+                Ok(()) // Return Ok after sending the error through the channel
+            }
+        }
+    }
+
+    async fn initial_fetch_async(&mut self) -> FetchResult {
+        let response = self.fetch().await?;
+
+        match response {
+            Some(resp) => {
+                let body = resp
+                    .text()
+                    .await
+                    .map_err(|e| Error::Server(format!("failed to read response body: {}", e)))?;
+
+                let document: source::Document = serde_json::from_str(&body).map_err(|e| {
+                    Error::InvalidJSON(format!("failed to parse response body: {}", e))
+                })?;
+
+                Ok(document)
+            }
+            None => Err(Error::Server("no data received from server".into())),
+        }
+    }
 }
 
 #[cfg(test)]
 mod tests {
+    use mockito::Server;
+
     use crate::http::Authentication;
-    use crate::http::HTTPParserBuilder;
-    use fliptevaluation::parser::Parser;
+    use crate::http::FetchMode;
+    use crate::http::HTTPFetcherBuilder;
 
-    #[test]
-    fn test_http_parse() {
-        let mut server = mockito::Server::new();
+    #[tokio::test]
+    async fn test_http_fetch() {
+        let mut server = Server::new_async().await;
         let mock = server
             .mock("GET", "/internal/v1/evaluation/snapshot/namespace/default")
             .with_status(200)
             .with_header("content-type", "application/json")
             .with_header("etag", "etag")
             .with_body(r#"{"namespace": {"key": "default"}, "flags":[]}"#)
-            .create();
+            .create_async()
+            .await;
 
         let url = server.url();
-        let mut parser = HTTPParserBuilder::new(&url)
+        let mut fetcher = HTTPFetcherBuilder::new(&url, "default")
             .authentication(Authentication::None)
             .build();
 
-        let result = parser.parse("default");
+        let result = fetcher.fetch().await;
 
         assert!(result.is_ok());
         mock.assert();
 
-        assert_eq!(parser.etag, Some("etag".to_string()));
+        assert_eq!(fetcher.etag, Some("etag".to_string()));
     }
 
-    #[test]
-    fn test_http_parse_not_modified() {
-        let mut server = mockito::Server::new();
+    #[tokio::test]
+    async fn test_http_fetch_with_reference() {
+        let mut server = Server::new_async().await;
+        let mock = server
+            .mock(
+                "GET",
+                "/internal/v1/evaluation/snapshot/namespace/default?reference=123",
+            )
+            .with_status(200)
+            .with_header("content-type", "application/json")
+            .with_header("etag", "etag")
+            .with_body(r#"{"namespace": {"key": "default"}, "flags":[]}"#)
+            .create_async()
+            .await;
+
+        let url = server.url();
+        let mut fetcher = HTTPFetcherBuilder::new(&url, "default")
+            .reference("123")
+            .build();
+
+        let result = fetcher.fetch().await;
+
+        assert!(result.is_ok());
+        mock.assert();
+
+        assert_eq!(fetcher.etag, Some("etag".to_string()));
+    }
+
+    #[tokio::test]
+    async fn test_http_fetch_not_modified() {
+        let mut server = mockito::Server::new_async().await;
         let mock = server
             .mock("GET", "/internal/v1/evaluation/snapshot/namespace/default")
             .match_header("if-none-match", "etag")
             .with_status(304)
-            .create();
+            .create_async()
+            .await;
 
         let url = server.url();
-        let mut parser = HTTPParserBuilder::new(&url)
+        let mut fetcher = HTTPFetcherBuilder::new(&url, "default")
             .authentication(Authentication::None)
             .build();
 
-        parser.etag = Some("etag".to_string());
+        fetcher.etag = Some("etag".to_string());
 
-        let result = parser.parse("default");
+        let result = fetcher.fetch().await;
 
         assert!(result.is_ok());
         assert!(result.unwrap().is_none());
@@ -244,113 +507,147 @@ mod tests {
         mock.assert();
     }
 
-    #[test]
-    fn test_http_parse_error() {
-        let mut server = mockito::Server::new();
+    #[tokio::test]
+    async fn test_http_fetch_error() {
+        let mut server = mockito::Server::new_async().await;
         let mock = server
             .mock("GET", "/internal/v1/evaluation/snapshot/namespace/default")
             .with_status(500)
-            .create();
+            .expect_at_most(4)
+            .create_async()
+            .await;
 
         let url = server.url();
-        let mut parser = HTTPParserBuilder::new(&url)
+        let mut fetcher = HTTPFetcherBuilder::new(&url, "default")
             .authentication(Authentication::None)
             .build();
 
-        let result = parser.parse("default");
+        let result = fetcher.fetch().await;
 
         assert!(!result.is_ok());
         mock.assert();
     }
 
-    #[test]
-    fn test_http_parse_token_auth() {
-        let mut server = mockito::Server::new();
+    #[tokio::test]
+    async fn test_http_fetch_token_auth() {
+        let mut server = mockito::Server::new_async().await;
         let mock = server
             .mock("GET", "/internal/v1/evaluation/snapshot/namespace/default")
             .match_header("authorization", "Bearer foo")
             .with_status(200)
             .with_header("content-type", "application/json")
             .with_body(r#"{"namespace": {"key": "default"}, "flags":[]}"#)
-            .create();
+            .create_async()
+            .await;
 
         let url = server.url();
-        let mut parser = HTTPParserBuilder::new(&url)
+        let mut fetcher = HTTPFetcherBuilder::new(&url, "default")
             .authentication(Authentication::ClientToken("foo".to_string()))
             .build();
 
-        let result = parser.parse("default");
+        let result = fetcher.fetch().await;
 
         assert!(result.is_ok());
         mock.assert();
     }
 
-    #[test]
-    fn test_http_parse_jwt_auth() {
-        let mut server = mockito::Server::new();
+    #[tokio::test]
+    async fn test_http_fetch_jwt_auth() {
+        let mut server = mockito::Server::new_async().await;
         let mock = server
             .mock("GET", "/internal/v1/evaluation/snapshot/namespace/default")
             .match_header("authorization", "JWT foo")
             .with_status(200)
             .with_header("content-type", "application/json")
             .with_body(r#"{"namespace": {"key": "default"}, "flags":[]}"#)
-            .create();
+            .create_async()
+            .await;
 
         let url = server.url();
-        let mut parser = HTTPParserBuilder::new(&url)
+        let mut fetcher = HTTPFetcherBuilder::new(&url, "default")
             .authentication(Authentication::JwtToken("foo".to_string()))
             .build();
 
-        let result = parser.parse("default");
+        let result = fetcher.fetch().await;
 
         assert!(result.is_ok());
         mock.assert();
     }
 
-    #[test]
-    fn test_http_parse_jwt_auth_failure() {
-        let mut server = mockito::Server::new();
+    #[tokio::test]
+    async fn test_http_fetch_jwt_auth_failure() {
+        let mut server = mockito::Server::new_async().await;
         let mock = server
             .mock("GET", "/internal/v1/evaluation/snapshot/namespace/default")
             .match_header("authorization", "JWT foo")
             .with_status(401)
             .with_header("content-type", "application/json")
             .with_body(r#"{"code":16,"message":"request was not authenticated","details":[]}"#)
-            .create();
+            .create_async()
+            .await;
 
         let url = server.url();
-        let mut parser = HTTPParserBuilder::new(&url)
+        let mut fetcher = HTTPFetcherBuilder::new(&url, "default")
             .authentication(Authentication::JwtToken("foo".to_string()))
             .build();
 
-        let result = parser.parse("default");
+        let result = fetcher.fetch().await;
 
         assert!(result.is_err());
         mock.assert();
     }
 
-    #[test]
-    fn test_http_parser_url() {
-        use super::HTTPParserBuilder;
+    #[tokio::test]
+    async fn test_http_fetch_stream() {
+        let mut server = Server::new_async().await;
+        let mock = server
+            .mock("GET", "/internal/v1/evaluation/snapshots?[]namespaces=default")
+            .with_status(200)
+            .with_header(
+                "content-type",
+                "application/json",
+            )
+            .with_chunked_body(|w| {
+                w.write_all(b"{\"namespace\": {\"key\": \"default\"}, \"flags\":[]}\n")?;
+                w.write_all(
+                    b"{\"namespace\": {\"key\": \"default\"}, \"flags\":[{\"key\": \"new_flag\"}]}\n",
+                )?;
+                Ok(())
+            })
+            .create_async()
+            .await;
 
-        let parser = HTTPParserBuilder::new("http://localhost:8080")
-            .authentication(Authentication::with_client_token("secret".into()))
-            .reference("ref")
+        let url = server.url();
+        let mut fetcher = HTTPFetcherBuilder::new(&url, "default")
+            .authentication(Authentication::None)
+            .mode(FetchMode::Streaming)
             .build();
 
-        assert_eq!(
-            parser.url("default"),
-            "http://localhost:8080/internal/v1/evaluation/snapshot/namespace/default?reference=ref"
-        );
+        let result = fetcher.fetch_stream().await;
 
-        let parser = HTTPParserBuilder::new("http://localhost:8080")
-            .authentication(Authentication::with_client_token("secret".into()))
+        assert!(result.is_ok());
+        mock.assert();
+    }
+
+    #[test]
+    fn test_initial_fetch() {
+        let mut server = Server::new();
+        let mock = server
+            .mock("GET", "/internal/v1/evaluation/snapshot/namespace/default")
+            .with_status(200)
+            .with_header("content-type", "application/json")
+            .with_body(r#"{"namespace": {"key": "default"}, "flags":[]}"#)
+            .create();
+
+        let url = server.url();
+        let mut fetcher = HTTPFetcherBuilder::new(&url, "default")
+            .authentication(Authentication::None)
             .build();
 
-        assert_eq!(
-            parser.url("default"),
-            "http://localhost:8080/internal/v1/evaluation/snapshot/namespace/default"
-        );
+        let result = fetcher.initial_fetch();
+
+        assert!(result.is_ok());
+        mock.assert();
     }
 
     #[test]
diff --git a/flipt-engine-ffi/src/lib.rs b/flipt-engine-ffi/src/lib.rs
index 975311e1..83543984 100644
--- a/flipt-engine-ffi/src/lib.rs
+++ b/flipt-engine-ffi/src/lib.rs
@@ -2,8 +2,6 @@ pub mod evaluator;
 pub mod http;
 
 use evaluator::Evaluator;
-use http::{Authentication, HTTPParser, HTTPParserBuilder};
-
 use fliptevaluation::error::Error;
 use fliptevaluation::models::flipt;
 use fliptevaluation::store::Snapshot;
@@ -11,14 +9,18 @@ use fliptevaluation::{
     BatchEvaluationResponse, BooleanEvaluationResponse, EvaluationRequest,
     VariantEvaluationResponse,
 };
+use http::{Authentication, FetchMode, HTTPFetcher, HTTPFetcherBuilder};
 use libc::c_void;
 use serde::{Deserialize, Serialize};
 use serde_json::{Map, Value};
 use std::collections::HashMap;
 use std::ffi::{CStr, CString};
 use std::os::raw::c_char;
+use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::{Arc, Mutex};
+use std::time::Duration;
 use thiserror::Error;
+use tokio::runtime::Runtime;
 
 #[derive(Deserialize)]
 struct FFIEvaluationRequest {
@@ -82,6 +84,7 @@ pub struct EngineOpts {
     url: Option<String>,
     authentication: Option<Authentication>,
     update_interval: Option<u64>,
+    fetch_mode: Option<FetchMode>,
     reference: Option<String>,
 }
 
@@ -92,36 +95,59 @@ impl Default for EngineOpts {
             authentication: None,
             update_interval: Some(120),
             reference: None,
+            fetch_mode: Some(FetchMode::default()),
         }
     }
 }
 
 pub struct Engine {
-    pub opts: EngineOpts,
-    pub evaluator: Arc<Mutex<Evaluator<HTTPParser, Snapshot>>>,
+    evaluator: Arc<Mutex<Evaluator<Snapshot>>>,
+    runtime: Runtime,
+    stop_signal: Arc<AtomicBool>,
 }
 
 impl Engine {
-    pub fn new(evaluator: Evaluator<HTTPParser, Snapshot>, opts: EngineOpts) -> Self {
-        let mut engine = Self {
-            opts,
-            evaluator: Arc::new(Mutex::new(evaluator)),
-        };
+    pub fn new(namespace: &str, mut fetcher: HTTPFetcher, evaluator: Evaluator<Snapshot>) -> Self {
+        let stop_signal = Arc::new(AtomicBool::new(false));
 
-        engine.update();
+        let runtime = Runtime::new().expect("failed to create runtime");
 
-        engine
-    }
+        let evaluator = Arc::new(Mutex::new(evaluator));
+        let evaluator_clone = evaluator.clone();
 
-    fn update(&mut self) {
-        let evaluator = self.evaluator.clone();
-        let update_interval = self.opts.update_interval.unwrap_or(120);
+        let initial_snapshot = fetcher.initial_fetch();
+        match initial_snapshot {
+            Ok(doc) => {
+                let snap = Snapshot::build(namespace, doc);
+                evaluator_clone.lock().unwrap().replace_snapshot(snap);
+            }
+            Err(err) => {
+                evaluator_clone.lock().unwrap().replace_snapshot(Err(err));
+            }
+        }
 
-        std::thread::spawn(move || loop {
-            std::thread::sleep(std::time::Duration::from_secs(update_interval));
-            let mut lock = evaluator.lock().unwrap();
-            lock.replace_snapshot();
+        let mut rx = runtime.block_on(async { fetcher.start(stop_signal.clone()) });
+
+        let namespace_clone = namespace.to_string();
+        runtime.spawn(async move {
+            while let Some(res) = rx.recv().await {
+                match res {
+                    Ok(doc) => {
+                        let snap = Snapshot::build(&namespace_clone, doc);
+                        evaluator_clone.lock().unwrap().replace_snapshot(snap);
+                    }
+                    Err(err) => {
+                        evaluator_clone.lock().unwrap().replace_snapshot(Err(err));
+                    }
+                }
+            }
         });
+
+        Self {
+            evaluator,
+            runtime,
+            stop_signal,
+        }
     }
 
     pub fn variant(
@@ -195,23 +221,38 @@ pub unsafe extern "C" fn initialize_engine(
 
     let authentication = engine_opts.authentication.to_owned();
     let reference = engine_opts.reference.to_owned();
+    let update_interval = engine_opts.update_interval.to_owned();
+    let fetch_mode = engine_opts.fetch_mode.to_owned();
+
+    let mut fetcher_builder = HTTPFetcherBuilder::new(&http_url, namespace);
+
+    fetcher_builder = match update_interval {
+        Some(update_interval) => {
+            fetcher_builder.update_interval(Duration::from_secs(update_interval))
+        }
+        None => fetcher_builder,
+    };
 
-    let mut parser_builder = HTTPParserBuilder::new(&http_url);
+    fetcher_builder = match authentication {
+        Some(authentication) => fetcher_builder.authentication(authentication),
+        None => fetcher_builder,
+    };
 
-    parser_builder = match authentication {
-        Some(authentication) => parser_builder.authentication(authentication),
-        None => parser_builder,
+    fetcher_builder = match fetch_mode {
+        Some(fetch_mode) => fetcher_builder.mode(fetch_mode),
+        None => fetcher_builder,
     };
 
-    parser_builder = match reference {
-        Some(reference) => parser_builder.reference(&reference),
-        None => parser_builder,
+    fetcher_builder = match reference {
+        Some(reference) => fetcher_builder.reference(&reference),
+        None => fetcher_builder,
     };
 
-    let parser = parser_builder.build();
-    let evaluator = Evaluator::new_snapshot_evaluator(namespace, parser).unwrap();
+    let fetcher = fetcher_builder.build();
+    let evaluator = Evaluator::new(namespace);
+    let engine = Engine::new(namespace, fetcher, evaluator);
 
-    Box::into_raw(Box::new(Engine::new(evaluator, engine_opts))) as *mut c_void
+    Box::into_raw(Box::new(engine)) as *mut c_void
 }
 
 /// # Safety
@@ -327,7 +368,13 @@ pub unsafe extern "C" fn destroy_engine(engine_ptr: *mut c_void) {
         return;
     }
 
-    drop(Box::from_raw(engine_ptr as *mut Engine));
+    let engine = Box::from_raw(engine_ptr as *mut Engine);
+    engine.stop_signal.store(true, Ordering::Relaxed);
+
+    // Shutdown the runtime
+    engine.runtime.shutdown_background();
+
+    // The engine will be dropped here, cleaning up all resources
 }
 
 /// # Safety
diff --git a/flipt-evaluation/src/lib.rs b/flipt-evaluation/src/lib.rs
index 8a3a28e1..d065b8b5 100644
--- a/flipt-evaluation/src/lib.rs
+++ b/flipt-evaluation/src/lib.rs
@@ -5,7 +5,6 @@ use web_time::Instant;
 
 pub mod error;
 pub mod models;
-pub mod parser;
 pub mod store;
 
 use crate::error::Error;
diff --git a/flipt-evaluation/src/models/flipt.rs b/flipt-evaluation/src/models/flipt.rs
index 329380f2..9e2897a4 100644
--- a/flipt-evaluation/src/models/flipt.rs
+++ b/flipt-evaluation/src/models/flipt.rs
@@ -1,7 +1,7 @@
 use serde::{Deserialize, Serialize};
 use std::collections::HashMap;
 
-#[derive(Clone, Debug, Serialize)]
+#[derive(Clone, Debug, Serialize, PartialEq)]
 pub struct Flag {
     pub key: String,
     pub enabled: bool,
@@ -10,14 +10,14 @@ pub struct Flag {
     pub default_variant: Option<Variant>,
 }
 
-#[derive(Clone, Debug, Serialize)]
+#[derive(Clone, Debug, Serialize, PartialEq)]
 pub struct Variant {
     pub id: String,
     pub key: String,
     pub attachment: String,
 }
 
-#[derive(Clone, Debug)]
+#[derive(Clone, Debug, PartialEq)]
 pub struct Constraint {
     pub segment_key: String,
     pub r#type: ConstraintComparisonType,
@@ -26,7 +26,7 @@ pub struct Constraint {
     pub value: String,
 }
 
-#[derive(Clone, Debug)]
+#[derive(Clone, Debug, PartialEq)]
 pub struct EvaluationRule {
     pub id: String,
     pub flag_key: String,
@@ -35,7 +35,7 @@ pub struct EvaluationRule {
     pub segment_operator: SegmentOperator,
 }
 
-#[derive(Clone, Debug)]
+#[derive(Clone, Debug, PartialEq)]
 pub struct EvaluationDistribution {
     pub rule_id: String,
     pub rollout: f32,
@@ -43,7 +43,7 @@ pub struct EvaluationDistribution {
     pub variant_attachment: String,
 }
 
-#[derive(Clone, Debug)]
+#[derive(Clone, Debug, PartialEq)]
 pub struct EvaluationRollout {
     pub rollout_type: RolloutType,
     pub rank: usize,
@@ -51,13 +51,13 @@ pub struct EvaluationRollout {
     pub threshold: Option<RolloutThreshold>,
 }
 
-#[derive(Clone, Debug)]
+#[derive(Clone, Debug, PartialEq)]
 pub struct RolloutThreshold {
     pub percentage: f32,
     pub value: bool,
 }
 
-#[derive(Clone, Debug)]
+#[derive(Clone, Debug, PartialEq)]
 pub struct RolloutSegment {
     pub value: bool,
     pub segment_operator: SegmentOperator,
diff --git a/flipt-evaluation/src/models/source.rs b/flipt-evaluation/src/models/source.rs
index db481a0f..5e6b9ac6 100644
--- a/flipt-evaluation/src/models/source.rs
+++ b/flipt-evaluation/src/models/source.rs
@@ -2,7 +2,7 @@ use serde::{Deserialize, Serialize};
 
 use super::flipt;
 
-#[derive(Deserialize, Serialize)]
+#[derive(Deserialize, Serialize, Debug)]
 #[serde(rename_all = "camelCase")]
 pub struct Document {
     pub namespace: Namespace,
@@ -21,14 +21,14 @@ impl Default for Document {
     }
 }
 
-#[derive(Deserialize, Serialize)]
+#[derive(Deserialize, Serialize, Debug)]
 #[serde(rename_all = "camelCase")]
 pub struct Namespace {
     pub key: String,
     pub name: Option<String>,
 }
 
-#[derive(Deserialize, Serialize)]
+#[derive(Deserialize, Serialize, Debug)]
 #[serde(rename_all = "camelCase")]
 pub struct Flag {
     pub key: String,
@@ -41,7 +41,7 @@ pub struct Flag {
     pub default_variant: Option<Variant>,
 }
 
-#[derive(Deserialize, Serialize)]
+#[derive(Deserialize, Serialize, Debug)]
 #[serde(rename_all = "camelCase")]
 pub struct Variant {
     pub id: String,
@@ -49,7 +49,7 @@ pub struct Variant {
     pub attachment: String,
 }
 
-#[derive(Deserialize, Serialize)]
+#[derive(Deserialize, Serialize, Debug)]
 #[serde(rename_all = "camelCase")]
 pub struct Rule {
     pub distributions: Vec<Distribution>,
@@ -57,7 +57,7 @@ pub struct Rule {
     pub segment_operator: flipt::SegmentOperator,
 }
 
-#[derive(Deserialize, Serialize)]
+#[derive(Deserialize, Serialize, Debug)]
 #[serde(rename_all = "camelCase")]
 pub struct Distribution {
     pub variant_key: String,
@@ -65,7 +65,7 @@ pub struct Distribution {
     pub variant_attachment: String,
 }
 
-#[derive(Deserialize, Serialize)]
+#[derive(Deserialize, Serialize, Debug)]
 #[serde(rename_all = "camelCase")]
 pub struct Rollout {
     pub description: Option<String>,
@@ -73,7 +73,7 @@ pub struct Rollout {
     pub threshold: Option<Threshold>,
 }
 
-#[derive(Deserialize, Serialize)]
+#[derive(Deserialize, Serialize, Debug)]
 #[serde(rename_all = "camelCase")]
 pub struct SegmentRule {
     pub segment_operator: Option<flipt::SegmentOperator>,
@@ -81,14 +81,14 @@ pub struct SegmentRule {
     pub segments: Vec<Segment>,
 }
 
-#[derive(Deserialize, Serialize)]
+#[derive(Deserialize, Serialize, Debug)]
 #[serde(rename_all = "camelCase")]
 pub struct Threshold {
     pub percentage: f32,
     pub value: bool,
 }
 
-#[derive(Deserialize, Serialize)]
+#[derive(Deserialize, Serialize, Debug)]
 #[serde(rename_all = "camelCase")]
 pub struct Segment {
     pub key: String,
@@ -96,7 +96,7 @@ pub struct Segment {
     pub constraints: Vec<SegmentConstraint>,
 }
 
-#[derive(Deserialize, Serialize)]
+#[derive(Deserialize, Serialize, Debug)]
 #[serde(rename_all = "camelCase")]
 pub struct SegmentConstraint {
     pub r#type: flipt::ConstraintComparisonType,
diff --git a/flipt-evaluation/src/parser.rs b/flipt-evaluation/src/parser.rs
deleted file mode 100644
index 30c0ba5f..00000000
--- a/flipt-evaluation/src/parser.rs
+++ /dev/null
@@ -1,46 +0,0 @@
-pub trait Parser {
-    fn parse(&mut self, namespace: &str) -> Result<Option<source::Document>, Error>;
-}
-
-#[cfg(test)]
-use std::fs;
-#[cfg(test)]
-use std::path::PathBuf;
-
-use crate::error::Error;
-use crate::models::source;
-
-#[cfg(test)]
-pub struct TestParser {
-    path: Option<String>,
-}
-
-#[cfg(test)]
-impl TestParser {
-    pub fn new() -> Self {
-        Self { path: None }
-    }
-}
-
-#[cfg(test)]
-impl Parser for TestParser {
-    fn parse(&mut self, _: &str) -> Result<Option<source::Document>, Error> {
-        let f = match &self.path {
-            Some(path) => path.to_owned(),
-            None => {
-                let mut d = PathBuf::from(env!("CARGO_MANIFEST_DIR"));
-                d.push("src/testdata/state.json");
-                d.display().to_string()
-            }
-        };
-
-        let state = fs::read_to_string(f).expect("file should have read correctly");
-
-        let document: source::Document = match serde_json::from_str(&state) {
-            Ok(document) => document,
-            Err(e) => return Err(Error::InvalidJSON(e.to_string())),
-        };
-
-        Ok(Some(document))
-    }
-}
diff --git a/flipt-evaluation/src/store.rs b/flipt-evaluation/src/store.rs
index f8c9ccae..5a50b1a0 100644
--- a/flipt-evaluation/src/store.rs
+++ b/flipt-evaluation/src/store.rs
@@ -27,10 +27,12 @@ pub trait Store {
     ) -> Option<Vec<flipt::EvaluationRollout>>;
 }
 
+#[derive(Debug, PartialEq, Clone)]
 pub struct Snapshot {
     namespace: Namespace,
 }
 
+#[derive(Debug, PartialEq, Clone)]
 struct Namespace {
     _key: String,
     flags: HashMap<String, flipt::Flag>,
@@ -277,15 +279,50 @@ impl Store for Snapshot {
 
 #[cfg(test)]
 mod tests {
+    use std::{fs, path::PathBuf};
+
     use super::{Snapshot, Store};
-    use crate::models::flipt;
-    use crate::parser::Parser;
-    use crate::parser::TestParser;
+    use crate::{error::Error, models::flipt, models::source};
+
+    #[cfg(test)]
+    pub struct TestFetcher {
+        path: Option<String>,
+    }
+
+    #[cfg(test)]
+    impl TestFetcher {
+        pub fn new() -> Self {
+            Self { path: None }
+        }
+    }
+
+    #[cfg(test)]
+    impl TestFetcher {
+        fn fetch(&mut self, _: &str) -> Result<Option<source::Document>, Error> {
+            let f = match &self.path {
+                Some(path) => path.to_owned(),
+                None => {
+                    let mut d = PathBuf::from(env!("CARGO_MANIFEST_DIR"));
+                    d.push("src/testdata/state.json");
+                    d.display().to_string()
+                }
+            };
+
+            let state = fs::read_to_string(f).expect("file should have read correctly");
+
+            let document: source::Document = match serde_json::from_str(&state) {
+                Ok(document) => document,
+                Err(e) => return Err(Error::InvalidJSON(e.to_string())),
+            };
+
+            Ok(Some(document))
+        }
+    }
 
     #[test]
     fn test_snapshot() {
-        let mut tp = TestParser::new();
-        let doc = tp.parse("default").unwrap();
+        let mut tp = TestFetcher::new();
+        let doc = tp.fetch("default").unwrap();
 
         let snapshot = Snapshot::build("default", doc.unwrap()).unwrap();