proof of concept for message durability

ARQUITECTURE.md

@@ -0,0 +1,60 @@
+```mermaid
+
+sequenceDiagram    
+    box Data Consolidation
+        participant ConsolidationService
+    end
+
+        participant SegmentClient
+    box HTTP
+        participant QueueHttp        
+        participant LooperHttp        
+        participant SegmentAPI
+    end
+    box File
+        participant QueueFile
+        participant WriteFile
+        participant File
+        participant WatchFile
+    end
+
+    activate ConsolidationService
+    ConsolidationService->>+SegmentClient: enqueue    
+    SegmentClient<<->>QueueHttp: offer 
+    alt QueueHttp overflow
+        SegmentClient<<->>QueueFile: put
+    end
+    SegmentClient->>-ConsolidationService: 
+    deactivate ConsolidationService
+
+    loop consume QueueHttp
+        LooperHttp->>QueueHttp:take
+        activate LooperHttp       
+    end
+    LooperHttp->>SegmentAPI: batchUpload
+    note over LooperHttp,SegmentAPI: Batch
+    note over LooperHttp,SegmentAPI: CircuitBreaker and Retry
+    note over LooperHttp,SegmentAPI: HTTP requests submited to a pool
+    deactivate LooperHttp
+
+    alt retry exhausted or circuit open
+        note over LooperHttp: pool threads
+        LooperHttp->>QueueFile: put
+    end
+
+    loop consume QueueFile
+        WriteFile->>QueueFile:take
+        activate WriteFile       
+    end
+    WriteFile->>File: write
+    note over WriteFile: Batch and save file
+    deactivate WriteFile
+
+    note over WatchFile: check last written 
+    activate WatchFile
+    loop watch QueueFile        
+        WatchFile->>File: read and remove
+        WatchFile->>QueueHttp: offer
+    end
+    deactivate WatchFile
+```

analytics/pom.xml

@@ -39,9 +39,21 @@
       <artifactId>findbugs</artifactId>
       <scope>provided</scope>
     </dependency>
+
     <dependency>
-      <groupId>com.segment.backo</groupId>
-      <artifactId>backo</artifactId>
+      <groupId>dev.failsafe</groupId>
+      <artifactId>failsafe</artifactId>
+      <version>3.3.2</version>
+    </dependency>
+    <dependency>
+      <groupId>dev.failsafe</groupId>
+      <artifactId>failsafe-retrofit</artifactId>
+      <version>3.3.2</version>
+   </dependency>
+   <dependency>
+      <groupId>commons-io</groupId>
+      <artifactId>commons-io</artifactId>
+      <version>2.18.0</version>
     </dependency>
     <dependency>
       <groupId>junit</groupId>
@@ -67,7 +79,19 @@
       <groupId>org.mockito</groupId>
       <artifactId>mockito-core</artifactId>
       <scope>test</scope>
-    </dependency>
+    </dependency>    
+	<dependency>
+      <groupId>org.wiremock</groupId>
+	  <artifactId>wiremock-standalone</artifactId>
+	  <version>3.2.0</version>
+      <scope>test</scope>
+	</dependency>
+	<dependency>
+      <groupId>org.awaitility</groupId>
+      <artifactId>awaitility</artifactId>
+      <version>4.2.2</version>
+      <scope>test</scope>
+    </dependency>    
   </dependencies>
 
   <build>

analytics/src/main/java/com/segment/analytics/Analytics.java

@@ -90,11 +90,6 @@ public boolean offer(MessageBuilder builder) {
     return client.offer(message);
   }
 
-  /** Flush events in the message queue. */
-  public void flush() {
-    client.flush();
-  }
-
   /** Stops this instance from processing further requests. */
   public void shutdown() {
     client.shutdown();
@@ -144,9 +139,7 @@ public static class Builder {
     private ThreadFactory threadFactory;
     private int flushQueueSize;
     private int maximumFlushAttempts;
-    private int maximumQueueSizeInBytes;
     private long flushIntervalInMillis;
-    private List<Callback> callbacks;
     private int queueCapacity;
     private boolean forceTlsV1 = false;
     private GsonBuilder gsonBuilder;
@@ -273,17 +266,6 @@ public Builder flushQueueSize(int flushQueueSize) {
       return this;
     }
 
-    /** Set the queueSize at which flushes should be triggered. */
-    @Beta
-    public Builder maximumQueueSizeInBytes(int bytes) {
-      if (bytes < 1) {
-        throw new IllegalArgumentException("maximumQueueSizeInBytes must not be less than 1.");
-      }
-
-      this.maximumQueueSizeInBytes = bytes;
-      return this;
-    }
-
     /** Set the interval at which the queue should be flushed. */
     @Beta
     public Builder flushInterval(long flushInterval, TimeUnit unit) {
@@ -323,21 +305,6 @@ public Builder threadFactory(ThreadFactory threadFactory) {
       return this;
     }
 
-    /** Add a {@link Callback} to be notified when an event is processed. */
-    public Builder callback(Callback callback) {
-      if (callback == null) {
-        throw new NullPointerException("Null callback");
-      }
-      if (callbacks == null) {
-        callbacks = new ArrayList<>();
-      }
-      if (callbacks.contains(callback)) {
-        throw new IllegalStateException("Callback is already registered.");
-      }
-      callbacks.add(callback);
-      return this;
-    }
-
     /** Use a {@link Plugin} to configure the builder. */
     @Beta
     public Builder plugin(Plugin plugin) {
@@ -390,9 +357,6 @@ public Analytics build() {
       if (flushQueueSize == 0) {
         flushQueueSize = Platform.get().defaultFlushQueueSize();
       }
-      if (maximumQueueSizeInBytes == 0) {
-        maximumQueueSizeInBytes = MESSAGE_QUEUE_MAX_BYTE_SIZE;
-      }
       if (maximumFlushAttempts == 0) {
         maximumFlushAttempts = 3;
       }
@@ -412,11 +376,6 @@ public Analytics build() {
       if (threadFactory == null) {
         threadFactory = Platform.get().defaultThreadFactory();
       }
-      if (callbacks == null) {
-        callbacks = Collections.emptyList();
-      } else {
-        callbacks = Collections.unmodifiableList(callbacks);
-      }
 
       HttpLoggingInterceptor interceptor =
           new HttpLoggingInterceptor(
@@ -456,21 +415,17 @@ public void log(String message) {
 
       SegmentService segmentService = restAdapter.create(SegmentService.class);
 
-      AnalyticsClient analyticsClient =
-          AnalyticsClient.create(
-              endpoint,
-              segmentService,
-              queueCapacity,
-              flushQueueSize,
-              flushIntervalInMillis,
-              maximumFlushAttempts,
-              maximumQueueSizeInBytes,
-              log,
-              threadFactory,
-              networkExecutor,
-              callbacks,
-              writeKey,
-              gson);
+    AnalyticsClient analyticsClient = AnalyticsClient.create(
+            endpoint,
+            segmentService,
+            queueCapacity,
+            flushQueueSize,
+            flushIntervalInMillis,
+            log,
+            threadFactory,
+            networkExecutor,
+            writeKey,
+            gson);
 
       return new Analytics(analyticsClient, messageTransformers, messageInterceptors, log);
     }