Implementing Retry logic [Reliable Channel]

core/build.gradle

@@ -53,6 +53,7 @@ dependencies {
     compile ([group: 'org.apache.commons', name: 'commons-lang3', version: '3.7'])
     compile ([group: 'org.apache.httpcomponents', name: 'httpclient', version: '4.5.3'])
     compile ([group: 'com.google.guava', name: 'guava', version: '20.0'])
+    compile ([group: 'com.google.code.gson', name: 'gson', version: '2.8.2'])
     testCompile group: 'junit', name: 'junit', version: '4.12'
     testCompile group: 'org.mockito', name: 'mockito-all', version: '1.10.19'
     testCompile group: 'com.google.code.gson', name: 'gson', version: '2.8.2'
@@ -65,6 +66,7 @@ shadowJar {
     relocate 'org.apache.commons', 'com.microsoft.applicationinsights.core.dependencies.apachecommons'
     relocate 'com.google.common', 'com.microsoft.applicationinsights.core.dependencies.googlecommon'
     relocate 'javax.annotation', 'com.microsoft.applicationinsights.core.dependencies.javaxannotation'
+    relocate 'com.google.gson', 'com.microsoft.applicationinsights.core.dependencies.gson'
 }
 
 jar {

core/src/main/java/com/microsoft/applicationinsights/channel/TelemetryChannel.java

@@ -65,4 +65,5 @@ public interface TelemetryChannel {
      * @param telemetrySampler - The sampler
      */
     void setSampler(TelemetrySampler telemetrySampler);
+
 }

core/src/main/java/com/microsoft/applicationinsights/channel/concrete/inprocess/InProcessTelemetryChannel.java

@@ -22,7 +22,6 @@
 package com.microsoft.applicationinsights.channel.concrete.inprocess;
 
 import java.io.IOException;
-import java.io.PrintWriter;
 import java.io.StringWriter;
 import java.net.URI;
 import java.util.Map;
@@ -38,7 +37,6 @@
 import com.microsoft.applicationinsights.internal.util.LocalStringsUtils;
 import com.microsoft.applicationinsights.internal.util.Sanitizer;
 import com.microsoft.applicationinsights.telemetry.JsonTelemetryDataSerializer;
-import com.microsoft.applicationinsights.telemetry.SupportSampling;
 import com.microsoft.applicationinsights.telemetry.Telemetry;
 import com.microsoft.applicationinsights.channel.TelemetryChannel;
 

core/src/main/java/com/microsoft/applicationinsights/channel/concrete/inprocess/InProcessTelemetryChannelFactory.java

@@ -35,8 +35,11 @@
 final class InProcessTelemetryChannelFactory implements TransmitterFactory {
     @Override
     public TelemetriesTransmitter create(String endpoint, String maxTransmissionStorageCapacity, boolean throttlingIsEnabled) {
-        final TransmissionPolicyManager transmissionPolicyManager = new TransmissionPolicyManager(throttlingIsEnabled);
-
+        final TransmissionPolicyManager transmissionPolicyManager = new TransmissionPolicyManager(throttlingIsEnabled);     	
+        transmissionPolicyManager.addTransmissionHandler(new ErrorHandler(transmissionPolicyManager));
+        transmissionPolicyManager.addTransmissionHandler(new PartialSuccessHandler(transmissionPolicyManager));
+        transmissionPolicyManager.addTransmissionHandler(new ThrottlingHandler(transmissionPolicyManager));
+
         // An active object with the network sender
         TransmissionNetworkOutput actualNetworkSender = TransmissionNetworkOutput.create(endpoint, transmissionPolicyManager);
 
@@ -51,6 +54,7 @@ public TelemetriesTransmitter create(String endpoint, String maxTransmissionStor
         // The dispatcher works with the two active senders
         TransmissionDispatcher dispatcher = new NonBlockingDispatcher(new TransmissionOutput[] {networkSender, activeFileSystemOutput});
         actualNetworkSender.setTransmissionDispatcher(dispatcher);
+
 
         // The loader works with the file system loader as the active one does
         TransmissionsLoader transmissionsLoader = new ActiveTransmissionLoader(fileSystemSender, stateFetcher, dispatcher);

core/src/main/java/com/microsoft/applicationinsights/internal/channel/TransmissionHandler.java

@@ -0,0 +1,17 @@
+package com.microsoft.applicationinsights.internal.channel;
+
+/**
+ * An interface that is used to create a concrete class that is called by the the {@link TransmissionHandlerObserver}
+ * <p>
+ * This is used to implement classes like {@link ErrorHandler} and {@link PartialSuccessHandler}.
+ * @author jamdavi
+ * 
+ * 
+ */
+public interface TransmissionHandler {
+	/**
+	 * Called when a transmission is sent by the {@link TransmissionOutput}. 
+	 * @param args The {@link TransmissionHandlerArgs} for this handler.
+	 */
+	void onTransmissionSent(TransmissionHandlerArgs args);
+}

core/src/main/java/com/microsoft/applicationinsights/internal/channel/TransmissionHandlerArgs.java

@@ -0,0 +1,32 @@
+package com.microsoft.applicationinsights.internal.channel;
+
+import org.apache.http.Header;
+
+import com.microsoft.applicationinsights.internal.channel.common.Transmission;
+
+public class TransmissionHandlerArgs {
+	private String responseBody;
+	public void setResponseBody(String body) { this.responseBody = body;}
+	public String getResponseBody() { return this.responseBody;}
+
+
+	private TransmissionDispatcher transmissionDispatcher;
+	public void setTransmissionDispatcher(TransmissionDispatcher dispatcher) { this.transmissionDispatcher = dispatcher;}
+	public TransmissionDispatcher getTransmissionDispatcher() { return this.transmissionDispatcher;}
+
+	private Transmission transmission;
+	public void setTransmission(Transmission transmission) { this.transmission = transmission;}
+	public Transmission getTransmission() { return this.transmission;}
+
+	private int responseCode;
+	public void setResponseCode(int code) { this.responseCode = code;}
+	public int getResponseCode() { return this.responseCode;}
+
+	private Throwable exception;
+	public void setException(Throwable ex) { this.exception = ex;}
+	public Throwable getException() { return this.exception;}
+
+	private Header retryHeader;
+	public void setRetryHeader(Header head) { this.retryHeader = head;}
+	public Header getRetryHeader() { return this.retryHeader;}
+}

core/src/main/java/com/microsoft/applicationinsights/internal/channel/TransmissionHandlerObserver.java

@@ -0,0 +1,18 @@
+package com.microsoft.applicationinsights.internal.channel;
+
+
+/**
+ * Enables the {@link TransmissionPolicyManager} to handle transmission states.
+ * <p>
+ * This interface extends {@TransmissionHandler} to add the ability to observe when the transmission is completed.
+ * @author jamdavi
+ *
+ */
+public interface TransmissionHandlerObserver extends TransmissionHandler {
+
+	/**
+	 * Used to add a {@link TransmissionHandler} to the collection stored by the {@link TransmissionPolicyManager}
+	 * @param handler	The handler to add to the collection.
+	 */
+	void addTransmissionHandler(TransmissionHandler handler);
+}

core/src/main/java/com/microsoft/applicationinsights/internal/channel/common/ActiveTransmissionLoader.java

@@ -26,13 +26,11 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 
+import com.google.common.base.Preconditions;
 import com.microsoft.applicationinsights.internal.channel.TransmissionDispatcher;
 import com.microsoft.applicationinsights.internal.channel.TransmissionsLoader;
 import com.microsoft.applicationinsights.internal.logger.InternalLogger;
 
-import com.google.common.base.Preconditions;
-import com.microsoft.applicationinsights.internal.shutdown.Stoppable;
-
 /**
  * The class is responsible for loading transmission files that were saved to the disk
  *
@@ -108,7 +106,7 @@ public void run() {
                                 case UNBLOCKED:
                                     fetchNext(true);
                                     break;
-
+                                case BACKOFF:
                                 case BLOCKED_BUT_CAN_BE_PERSISTED:
                                     Thread.sleep(DEFAULT_SLEEP_INTERVAL_AFTER_DISPATCHING_IN_MILLS);
                                     break;

core/src/main/java/com/microsoft/applicationinsights/internal/channel/common/BackendResponse.java

@@ -0,0 +1,20 @@
+package com.microsoft.applicationinsights.internal.channel.common;
+
+/**
+ * Utility class used by the {@link PartialSuccessHandler}
+ * 
+ * @author jamdavi
+ *
+ */
+class BackendResponse {
+
+	int itemsReceived;
+	int itemsAccepted;
+	Error[] errors;
+
+	class Error {
+		public int index;
+		public int statusCode;
+		public String message;
+	}
+}

core/src/main/java/com/microsoft/applicationinsights/internal/channel/common/ErrorHandler.java

@@ -0,0 +1,66 @@
+package com.microsoft.applicationinsights.internal.channel.common;
+
+import org.apache.http.HttpStatus;
+
+import com.microsoft.applicationinsights.internal.channel.TransmissionHandler;
+import com.microsoft.applicationinsights.internal.channel.TransmissionHandlerArgs;
+import com.microsoft.applicationinsights.internal.logger.InternalLogger;
+
+/**
+ * This class implements the retry logic for transmissions with the results of a
+ * 408, 500, and 503 result.
+ * <p>
+ * It does not handle any error codes such as 400, 401, 403, 404, etc.
+ * 
+ * @author jamdavi
+ *
+ */
+public class ErrorHandler implements TransmissionHandler {
+
+	private TransmissionPolicyManager transmissionPolicyManager;
+
+	/**
+	 * Ctor
+	 * 
+	 * Constructs the ErrorHandler object.
+	 * 
+	 * @param policy
+	 *            The {@link TransmissionPolicyManager} object that is needed to
+	 *            control the back off policy
+	 */
+	public ErrorHandler(TransmissionPolicyManager policy) {
+		this.transmissionPolicyManager = policy;
+	}
+
+	@Override
+	public void onTransmissionSent(TransmissionHandlerArgs args) {
+
+		validateTransmissionAndSend(args);
+	}
+
+	boolean validateTransmissionAndSend(TransmissionHandlerArgs args) {
+		if (args.getTransmission() != null && args.getTransmissionDispatcher() != null) {
+			args.getTransmission().incrementNumberOfSends();
+			switch (args.getResponseCode()) {
+			case TransmissionSendResult.REQUEST_TIMEOUT:
+			case TransmissionSendResult.INTERNAL_SERVER_ERROR:
+			case TransmissionSendResult.SERVICE_UNAVAILABLE:
+				backoffAndSendTransmission(args);
+				return true;
+			default:
+				InternalLogger.INSTANCE.trace("Http response code %s not handled by %s", args.getResponseCode(),
+						this.getClass().getName());
+				return false;
+			}
+		} else if (args.getException() != null) {
+			backoffAndSendTransmission(args);
+			return true;
+		}
+		return false;
+	}
+
+	private void backoffAndSendTransmission(TransmissionHandlerArgs args) {
+		this.transmissionPolicyManager.backoff();
+		args.getTransmissionDispatcher().dispatch(args.getTransmission());
+	}
+}