Add multiplexed transport

sentry/api/sentry.api

@@ -1422,6 +1422,11 @@ public final class io/sentry/MonitorScheduleUnit : java/lang/Enum {
 	public static fun values ()[Lio/sentry/MonitorScheduleUnit;
 }
 
+public final class io/sentry/MultiplexedTransportFactory : io/sentry/ITransportFactory {
+	public fun <init> (Lio/sentry/ITransportFactory;Ljava/util/List;)V
+	public fun create (Lio/sentry/SentryOptions;Lio/sentry/RequestDetails;)Lio/sentry/transport/ITransport;
+}
+
 public final class io/sentry/NoOpCompositePerformanceCollector : io/sentry/CompositePerformanceCollector {
 	public fun close ()V
 	public static fun getInstance ()Lio/sentry/NoOpCompositePerformanceCollector;
@@ -6257,6 +6262,16 @@ public abstract interface class io/sentry/transport/ITransportGate {
 	public abstract fun isConnected ()Z
 }
 
+public final class io/sentry/transport/MultiplexedTransport : io/sentry/transport/ITransport {
+	public fun <init> (Ljava/util/List;)V
+	public fun close ()V
+	public fun close (Z)V
+	public fun flush (J)V
+	public fun getRateLimiter ()Lio/sentry/transport/RateLimiter;
+	public fun isHealthy ()Z
+	public fun send (Lio/sentry/SentryEnvelope;Lio/sentry/Hint;)V
+}
+
 public final class io/sentry/transport/NoOpEnvelopeCache : io/sentry/cache/IEnvelopeCache {
 	public fun <init> ()V
 	public fun discard (Lio/sentry/SentryEnvelope;)V

sentry/src/main/java/io/sentry/MultiplexedTransportFactory.java

@@ -0,0 +1,40 @@
+package io.sentry;
+
+import io.sentry.transport.ITransport;
+import io.sentry.transport.MultiplexedTransport;
+import io.sentry.util.Objects;
+import java.util.List;
+import java.util.stream.Collectors;
+import org.jetbrains.annotations.ApiStatus;
+import org.jetbrains.annotations.NotNull;
+
+@ApiStatus.Experimental
+public final class MultiplexedTransportFactory implements ITransportFactory {
+
+  private final @NotNull ITransportFactory transportFactory;
+  private final @NotNull List<Dsn> dsns;
+
+  public MultiplexedTransportFactory(
+      final @NotNull ITransportFactory transportFactory, final @NotNull List<String> dsns) {
+    Objects.requireNonNull(transportFactory, "transportFactory is required");
+    Objects.requireNonNull(dsns, "dsns is required");
+
+    this.transportFactory = transportFactory;
+    this.dsns = dsns.stream().map(Dsn::new).collect(Collectors.toList());
+  }
+
+  @Override
+  public @NotNull ITransport create(
+      final @NotNull SentryOptions options, final @NotNull RequestDetails requestDetails) {
+    final List<ITransport> transports =
+        dsns.stream().map(dsn -> createTransport(options, dsn)).collect(Collectors.toList());
+    return new MultiplexedTransport(transports);
+  }
+
+  private @NotNull ITransport createTransport(
+      final @NotNull SentryOptions options, final @NotNull Dsn dsn) {
+    final RequestDetails requestDetails =
+        new RequestDetailsResolver(dsn, options.getSentryClientName()).resolve();
+    return transportFactory.create(options, requestDetails);
+  }
+}

sentry/src/main/java/io/sentry/RequestDetailsResolver.java

@@ -5,6 +5,7 @@
 import java.util.HashMap;
 import java.util.Map;
 import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
 
 /** Resolves {@link RequestDetails}. */
 final class RequestDetailsResolver {
@@ -13,15 +14,23 @@ final class RequestDetailsResolver {
   /** HTTP Header for the authentication to Sentry. */
   private static final String SENTRY_AUTH = "X-Sentry-Auth";
 
-  private final @NotNull SentryOptions options;
+  private final @NotNull Dsn dsn;
+  private final @Nullable String sentryClientName;
+
+  public RequestDetailsResolver(final @NotNull Dsn dsn, final @Nullable String sentryClientName) {
+    this.dsn = Objects.requireNonNull(dsn, "dsn is required");
+    this.sentryClientName = sentryClientName;
+  }
 
   public RequestDetailsResolver(final @NotNull SentryOptions options) {
-    this.options = Objects.requireNonNull(options, "options is required");
+    Objects.requireNonNull(options, "options is required");
+
+    this.dsn = options.retrieveParsedDsn();
+    this.sentryClientName = options.getSentryClientName();
   }
 
   @NotNull
   RequestDetails resolve() {
-    final Dsn dsn = options.retrieveParsedDsn();
     final URI sentryUri = dsn.getSentryUri();
     final String envelopeUrl = sentryUri.resolve(sentryUri.getPath() + "/envelope/").toString();
 
@@ -33,15 +42,14 @@ RequestDetails resolve() {
             + SentryClient.SENTRY_PROTOCOL_VERSION
             + ","
             + "sentry_client="
-            + options.getSentryClientName()
+            + sentryClientName
             + ","
             + "sentry_key="
             + publicKey
             + (secretKey != null && secretKey.length() > 0 ? (",sentry_secret=" + secretKey) : "");
-    final String userAgent = options.getSentryClientName();
 
     final Map<String, String> headers = new HashMap<>();
-    headers.put(USER_AGENT, userAgent);
+    headers.put(USER_AGENT, sentryClientName);
     headers.put(SENTRY_AUTH, authHeader);
 
     return new RequestDetails(envelopeUrl, headers);

sentry/src/main/java/io/sentry/transport/MultiplexedTransport.java

@@ -0,0 +1,63 @@
+package io.sentry.transport;
+
+import io.sentry.Hint;
+import io.sentry.SentryEnvelope;
+import io.sentry.util.Objects;
+import java.io.IOException;
+import java.util.List;
+import java.util.Optional;
+import java.util.stream.Collectors;
+import org.jetbrains.annotations.ApiStatus;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+
+@ApiStatus.Internal
+public final class MultiplexedTransport implements ITransport {
+  private final @NotNull List<ITransport> transports;
+
+  public MultiplexedTransport(final @NotNull List<ITransport> transports) {
+    this.transports = Objects.requireNonNull(transports, "transports is required");
+  }
+
+  @Override
+  public void send(final @NotNull SentryEnvelope envelope, final @NotNull Hint hint)
+      throws IOException {
+    for (ITransport transport : transports) {
+      transport.send(envelope, hint);
+    }
+  }
+
+  @Override
+  public boolean isHealthy() {
+    return transports.stream().allMatch(ITransport::isHealthy);
+  }
+
+  @Override
+  public void flush(final long timeoutMillis) {
+    transports.forEach(transport -> transport.flush(timeoutMillis));
+  }
+
+  @Override
+  public @Nullable RateLimiter getRateLimiter() {
+    // Prefer one with rate limit active, else fall back to arbitrary one
+    final List<RateLimiter> rateLimiters =
+        this.transports.stream().map(ITransport::getRateLimiter).collect(Collectors.toList());
+    final Optional<RateLimiter> activeRateLimiter =
+        rateLimiters.stream().filter(RateLimiter::isAnyRateLimitActive).findAny();
+    return activeRateLimiter.orElse(rateLimiters.stream().findAny().orElse(null));
+  }
+
+  @Override
+  public void close(final boolean isRestarting) throws IOException {
+    for (ITransport transport : transports) {
+      transport.close(isRestarting);
+    }
+  }
+
+  @Override
+  public void close() throws IOException {
+    for (ITransport transport : transports) {
+      transport.close();
+    }
+  }
+}

sentry/src/test/java/io/sentry/MultiplexedTransportFactoryTest.kt

@@ -0,0 +1,52 @@
+package io.sentry
+import org.junit.Test
+import org.mockito.kotlin.argumentCaptor
+import org.mockito.kotlin.eq
+import org.mockito.kotlin.mock
+import org.mockito.kotlin.times
+import org.mockito.kotlin.verify
+import kotlin.test.assertContains
+import kotlin.test.assertEquals
+import kotlin.test.assertNotNull
+
+class MultiplexedTransportFactoryTest {
+    private class Fixture {
+        val dsn1 = "https://[email protected]:65535/2147483647"
+        var dsn2 = "https://[email protected]:65535/1234567890"
+        var transportFactory = mock<ITransportFactory>()
+        var sentryOptions: SentryOptions = SentryOptions().apply {
+            dsn = dsn1
+        }
+
+        fun getSUT(): MultiplexedTransportFactory {
+            return MultiplexedTransportFactory(transportFactory, listOf(dsn1, dsn2))
+        }
+    }
+
+    private val fixture = Fixture()
+
+    @Test
+    fun `create transport`() {
+        val requestDetailsResolver = RequestDetailsResolver(fixture.sentryOptions)
+
+        val transport = fixture.getSUT()
+            .create(fixture.sentryOptions, requestDetailsResolver.resolve())
+
+        assertNotNull(transport)
+
+        val captor = argumentCaptor<RequestDetails>()
+        verify(fixture.transportFactory, times(2))
+            .create(eq(fixture.sentryOptions), captor.capture())
+
+        assertEquals("/api/2147483647/envelope/", captor.firstValue.url.path)
+        assertEquals("/api/1234567890/envelope/", captor.secondValue.url.path)
+        assertContains(
+            captor.firstValue.headers["X-Sentry-Auth"].toString(),
+            "sentry_key=d4d82fc1c2c4032a83f3a29aa3a3aff"
+        )
+        assertContains(
+            captor.secondValue.headers["X-Sentry-Auth"].toString(),
+            "sentry_key=007508298f9048729e434faa9ae4f49"
+        )
+    }
+}

sentry/src/test/java/io/sentry/transport/MultiplexedTransportTest.kt

@@ -0,0 +1,95 @@
+package io.sentry.transport
+
+import io.sentry.Hint
+import io.sentry.SentryEnvelope
+import io.sentry.SentryOptions
+import io.sentry.Session
+import io.sentry.dsnString
+import io.sentry.protocol.User
+import org.junit.Test
+import org.mockito.kotlin.mock
+import org.mockito.kotlin.verify
+import org.mockito.kotlin.whenever
+import kotlin.test.assertEquals
+import kotlin.test.assertFalse
+import kotlin.test.assertTrue
+
+class MultiplexedTransportTest {
+    private class Fixture {
+        var transport1 = mock<ITransport>()
+        var transport2 = mock<ITransport>()
+        var rateLimiter1 = mock<RateLimiter>()
+        var rateLimiter2 = mock<RateLimiter>()
+        var sentryOptions: SentryOptions = SentryOptions().apply {
+            dsn = dsnString
+            setSerializer(mock())
+            setEnvelopeDiskCache(mock())
+        }
+
+        init {
+            whenever(transport1.rateLimiter).thenReturn(rateLimiter1)
+            whenever(transport2.rateLimiter).thenReturn(rateLimiter2)
+        }
+
+        fun getSUT(): MultiplexedTransport {
+            return MultiplexedTransport(listOf(transport1, transport2))
+        }
+    }
+    private val fixture = Fixture()
+
+    private fun createSession(): Session {
+        return Session("123", User(), "env", "release")
+    }
+
+    @Test
+    fun `send sends to all transports`() {
+        val envelope = SentryEnvelope
+            .from(fixture.sentryOptions.serializer, createSession(), null)
+        val hint = Hint()
+
+        fixture.getSUT().send(envelope, hint)
+
+        verify(fixture.transport1).send(envelope, hint)
+        verify(fixture.transport2).send(envelope, hint)
+    }
+
+    @Test
+    fun `healthy if all transports are healthy`() {
+        whenever(fixture.transport1.isHealthy).thenReturn(true)
+        whenever(fixture.transport2.isHealthy).thenReturn(true)
+
+        assertTrue(fixture.getSUT().isHealthy)
+    }
+
+    @Test
+    fun `not healthy if one transport is unhealthy`() {
+        whenever(fixture.transport1.isHealthy).thenReturn(true)
+        whenever(fixture.transport2.isHealthy).thenReturn(false)
+
+        assertFalse(fixture.getSUT().isHealthy)
+    }
+
+    @Test
+    fun `close closes all transports`() {
+        fixture.getSUT().close()
+
+        verify(fixture.transport1).close()
+        verify(fixture.transport2).close()
+    }
+
+    @Test
+    fun `close with isRestarting closes all transports with isRestarting`() {
+        fixture.getSUT().close(true)
+
+        verify(fixture.transport1).close(true)
+        verify(fixture.transport2).close(true)
+    }
+
+    @Test
+    fun `getRateLimiter returns active rate limiter if exists`() {
+        whenever(fixture.rateLimiter1.isAnyRateLimitActive).thenReturn(false)
+        whenever(fixture.rateLimiter2.isAnyRateLimitActive).thenReturn(true)
+
+        assertEquals(fixture.rateLimiter2, fixture.getSUT().rateLimiter)
+    }
+}