Serialize shared HTTP envelope codecs

Author: BenCodez

SimpleAPI/src/main/java/com/bencodez/simpleapi/servercomm/http/HttpBackendTransportConnector.java

@@ -122,7 +122,8 @@ private HttpBackendTransportConnector(HttpClientCredentialStore.HttpClientProfil
 		if (profile == null || credential == null || onEnvelope == null || wireCodec == null)
 			throw new IllegalArgumentException("HTTP backend transport configuration is invalid");
 		if (!matchesCredential(profile, credential)) throw new IllegalArgumentException("HTTP client certificate does not match transport profile");
-		this.profile = profile; this.serverId = profile.serverId(); this.onEnvelope = onEnvelope; this.wireCodec = wireCodec;
+		this.profile = profile; this.serverId = profile.serverId(); this.onEnvelope = onEnvelope;
+		this.wireCodec = HttpEnvelopeWireCodec.serialized(wireCodec);
 		this.credential = credential;
 		this.credentialDirectory = credentialDirectory;
 		HttpInboundDeliveryStore loadedInbound = null, loadedAcknowledgements = null;

SimpleAPI/src/main/java/com/bencodez/simpleapi/servercomm/http/HttpEnvelopeWireCodec.java

@@ -1,5 +1,7 @@
 package com.bencodez.simpleapi.servercomm.http;
 
+import java.util.Objects;
+
 import com.bencodez.simpleapi.servercomm.codec.JsonEnvelope;
 
 /**
@@ -20,6 +22,12 @@ static HttpEnvelopeWireCodec identity() {
 		return Identity.INSTANCE;
 	}
 
+	/** Serializes a codec shared by concurrent HTTP sessions. */
+	static HttpEnvelopeWireCodec serialized(HttpEnvelopeWireCodec codec) {
+		Objects.requireNonNull(codec, "wireCodec");
+		return codec == Identity.INSTANCE ? codec : new Serialized(codec);
+	}
+
 	final class Identity implements HttpEnvelopeWireCodec {
 		private static final Identity INSTANCE = new Identity();
 
@@ -29,4 +37,26 @@ private Identity() { }
 
 		@Override public JsonEnvelope decode(JsonEnvelope envelope) { return envelope; }
 	}
+
+	final class Serialized implements HttpEnvelopeWireCodec {
+		private final HttpEnvelopeWireCodec delegate;
+
+		private Serialized(HttpEnvelopeWireCodec delegate) {
+			this.delegate = delegate;
+		}
+
+		@Override
+		public JsonEnvelope encode(JsonEnvelope envelope) {
+			synchronized (delegate) {
+				return delegate.encode(envelope);
+			}
+		}
+
+		@Override
+		public JsonEnvelope decode(JsonEnvelope envelope) {
+			synchronized (delegate) {
+				return delegate.decode(envelope);
+			}
+		}
+	}
 }

SimpleAPI/src/main/java/com/bencodez/simpleapi/servercomm/http/HttpProxyTransportServer.java

@@ -133,7 +133,9 @@ public HttpProxyTransportServer(InetSocketAddress bind, HttpTlsIdentity identity
 			throw new IllegalArgumentException("HTTP transport configuration is required");
 		if (wireCodec == null || nanoTime == null) throw new IllegalArgumentException("HTTP transport codec and clock are required");
 		this.identity = identity; this.authority = authority; this.onEnvelope = onEnvelope;
-		this.onAcknowledged = onAcknowledged; this.wireCodec = wireCodec; this.nanoTime = nanoTime;
+		this.onAcknowledged = onAcknowledged;
+		this.wireCodec = HttpEnvelopeWireCodec.serialized(wireCodec);
+		this.nanoTime = nanoTime;
 		durableOutgoing = outgoingDirectory == null ? null : new DurableOutgoingQueue(outgoingDirectory);
 		HttpsServer createdServer = null;
 		ThreadPoolExecutor createdListener = null, createdHandler = null;

SimpleAPI/src/test/java/com/bencodez/simpleapi/servercomm/http/HttpEnvelopeWireCodecTest.java

@@ -13,14 +13,69 @@
 import java.util.LinkedHashMap;
 import java.util.UUID;
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.io.TempDir;
 
 class HttpEnvelopeWireCodecTest {
 	@TempDir Path directory;
 
+	@Test
+	void serializedWrappersCoordinateConcurrentAccessToOneCodec() throws Exception {
+		CountDownLatch firstEntered = new CountDownLatch(1);
+		CountDownLatch releaseFirst = new CountDownLatch(1);
+		AtomicInteger active = new AtomicInteger();
+		AtomicInteger maximumActive = new AtomicInteger();
+		HttpEnvelopeWireCodec delegate = new HttpEnvelopeWireCodec() {
+			private JsonEnvelope apply(JsonEnvelope envelope) {
+				int current = active.incrementAndGet();
+				maximumActive.accumulateAndGet(current, Math::max);
+				try {
+					if (current == 1) {
+						firstEntered.countDown();
+						assertTrue(releaseFirst.await(5, TimeUnit.SECONDS));
+					}
+					return envelope;
+				} catch (InterruptedException interrupted) {
+					Thread.currentThread().interrupt();
+					throw new IllegalStateException(interrupted);
+				} finally {
+					active.decrementAndGet();
+				}
+			}
+
+			@Override public JsonEnvelope encode(JsonEnvelope envelope) { return apply(envelope); }
+			@Override public JsonEnvelope decode(JsonEnvelope envelope) { return apply(envelope); }
+		};
+		HttpEnvelopeWireCodec first = HttpEnvelopeWireCodec.serialized(delegate);
+		HttpEnvelopeWireCodec second = HttpEnvelopeWireCodec.serialized(delegate);
+		JsonEnvelope envelope = JsonEnvelope.builder("test").build();
+		var executor = Executors.newFixedThreadPool(2);
+		try {
+			var firstCall = executor.submit(() -> first.encode(envelope));
+			assertTrue(firstEntered.await(5, TimeUnit.SECONDS));
+			CountDownLatch secondAttempted = new CountDownLatch(1);
+			var secondCall = executor.submit(() -> {
+				secondAttempted.countDown();
+				return second.decode(envelope);
+			});
+			assertTrue(secondAttempted.await(5, TimeUnit.SECONDS));
+			assertThrows(java.util.concurrent.TimeoutException.class,
+					() -> secondCall.get(100, TimeUnit.MILLISECONDS));
+			assertEquals(1, maximumActive.get());
+			releaseFirst.countDown();
+			assertEquals(envelope, firstCall.get(5, TimeUnit.SECONDS));
+			assertEquals(envelope, secondCall.get(5, TimeUnit.SECONDS));
+			assertEquals(1, maximumActive.get());
+		} finally {
+			releaseFirst.countDown();
+			executor.shutdownNow();
+		}
+	}
+
 	@Test
 	void durableProxyQueueKeepsSemanticEnvelopeAcrossCodecChange() throws Exception {
 		Path outgoing = directory.resolve("outgoing");