[CAMEL-22557] AS2: Fix configuration isolation and enable port sharing for inbound routes

components/camel-as2/camel-as2-api/src/main/java/org/apache/camel/component/as2/api/AS2ServerConnection.java

@@ -64,57 +64,88 @@ public class AS2ServerConnection {
     private static final String REQUEST_LISTENER_THREAD_NAME_PREFIX = "AS2Svr-";
     private static final String REQUEST_HANDLER_THREAD_NAME_PREFIX = "AS2Hdlr-";
 
-    class RequestListenerThread extends Thread {
+    private ServerSocket serversocket;
+    private RequestListenerService listenerService;
+    private RequestAcceptorThread acceptorThread;
+    private final Lock lock = new ReentrantLock();
+
+    private final String as2Version;
+    private final String originServer;
+    private final String serverFqdn;
+    private final Certificate[] signingCertificateChain;
+    private final PrivateKey signingPrivateKey;
+    private final PrivateKey decryptingPrivateKey;
+    private final Certificate[] validateSigningCertificateChain;
+    private final AS2SignatureAlgorithm signingAlgorithm;
+
+    class RequestListenerService {
 
-        private final ServerSocket serversocket;
         private final HttpService httpService;
         private final RequestHandlerRegistry registry;
-        private final HttpServerRequestHandler handler;
-
-        public RequestListenerThread(String as2Version,
-                                     String originServer,
-                                     String serverFqdn,
-                                     int port,
-                                     AS2SignatureAlgorithm signatureAlgorithm,
-                                     Certificate[] signingCertificateChain,
-                                     PrivateKey signingPrivateKey,
-                                     PrivateKey decryptingPrivateKey,
-                                     String mdnMessageTemplate,
-                                     Certificate[] validateSigningCertificateChain,
-                                     SSLContext sslContext)
-                                                            throws IOException {
-            setName(REQUEST_LISTENER_THREAD_NAME_PREFIX + port);
 
-            if (sslContext == null) {
-                serversocket = new ServerSocket(port);
-            } else {
-                SSLServerSocketFactory factory = sslContext.getServerSocketFactory();
-                serversocket = factory.createServerSocket(port);
-            }
+        public RequestListenerService(String as2Version,
+                                      String originServer,
+                                      String serverFqdn,
+                                      String mdnMessageTemplate,
+                                      Certificate[] validateSigningCertificateChain)
+                                                                                     throws IOException {
 
             // Set up HTTP protocol processor for incoming connections
-            final HttpProcessor inhttpproc = initProtocolProcessor(as2Version, originServer, serverFqdn,
-                    signatureAlgorithm, signingCertificateChain, signingPrivateKey, decryptingPrivateKey, mdnMessageTemplate,
+            final HttpProcessor inhttpproc = initProtocolProcessor(
+                    as2Version, originServer, serverFqdn,
+                    AS2ServerConnection.this.signingAlgorithm,
+                    AS2ServerConnection.this.signingCertificateChain,
+                    AS2ServerConnection.this.signingPrivateKey,
+                    AS2ServerConnection.this.decryptingPrivateKey,
+                    mdnMessageTemplate,
                     validateSigningCertificateChain);
 
             registry = new RequestHandlerRegistry<>();
-            handler = new BasicHttpServerRequestHandler(registry);
+            HttpServerRequestHandler handler = new BasicHttpServerRequestHandler(registry);
 
             // Set up the HTTP service
             httpService = new HttpService(inhttpproc, handler);
         }
 
+        void registerHandler(String requestUriPattern, HttpRequestHandler httpRequestHandler) {
+            registry.register(null, requestUriPattern, httpRequestHandler);
+        }
+
+        void unregisterHandler(String requestUriPattern) {
+            // we cannot remove from http registry, but we can replace with a not found to simulate 404
+            registry.register(null, requestUriPattern, new NotFoundHttpRequestHandler());
+        }
+    }
+
+    class RequestAcceptorThread extends Thread {
+
+        private final RequestListenerService service;
+
+        public RequestAcceptorThread(int port, SSLContext sslContext, RequestListenerService service)
+                                                                                                      throws IOException {
+            setName(REQUEST_LISTENER_THREAD_NAME_PREFIX + port);
+            this.service = service;
+
+            // 2. BIND THE PORT HERE! This happens only once.
+            if (sslContext == null) {
+                serversocket = new ServerSocket(port);
+            } else {
+                SSLServerSocketFactory factory = sslContext.getServerSocketFactory();
+                serversocket = factory.createServerSocket(port);
+            }
+        }
+
         @Override
         public void run() {
-            LOG.info("Listening on port {}", this.serversocket.getLocalPort());
+            // serversocket is now a field of the outer AS2ServerConnection class
+            LOG.info("Listening on port {}", serversocket.getLocalPort());
             while (!Thread.interrupted()) {
                 try {
-
                     // Set up incoming HTTP connection
-                    final Socket inSocket = this.serversocket.accept();
+                    final Socket inSocket = serversocket.accept();
 
-                    // Start worker thread
-                    final Thread t = new RequestHandlerThread(this.httpService, inSocket);
+                    // Start worker thread, using the service's HttpService
+                    final Thread t = new RequestHandlerThread(this.service.httpService, inSocket);
                     t.setDaemon(true);
                     t.start();
                 } catch (final InterruptedIOException | SocketException ex) {
@@ -126,15 +157,6 @@ public void run() {
                 }
             }
         }
-
-        void registerHandler(String requestUriPattern, HttpRequestHandler httpRequestHandler) {
-            registry.register(null, requestUriPattern, httpRequestHandler);
-        }
-
-        void unregisterHandler(String requestUriPattern) {
-            // we cannot remove from http registry, but we can replace with a not found to simulate 404
-            registry.register(null, requestUriPattern, new NotFoundHttpRequestHandler());
-        }
     }
 
     class RequestHandlerThread extends Thread {
@@ -175,12 +197,18 @@ public void run() {
                                 AS2AsynchronousMDNManager.ASYNCHRONOUS_MDN,
                                 DispositionNotificationMultipartReportEntity.class);
                         AS2AsynchronousMDNManager asynchronousMDNManager = new AS2AsynchronousMDNManager(
-                                as2Version,
-                                originServer, serverFqdn, signingCertificateChain, signingPrivateKey);
+                                AS2ServerConnection.this.as2Version,
+                                AS2ServerConnection.this.originServer,
+                                AS2ServerConnection.this.serverFqdn,
+                                AS2ServerConnection.this.signingCertificateChain,
+                                AS2ServerConnection.this.signingPrivateKey);
 
                         HttpRequest request = coreContext.getAttribute(HttpCoreContext.HTTP_REQUEST, HttpRequest.class);
                         AS2SignedDataGenerator gen = ResponseMDN.createSigningGenerator(
-                                request, signingAlgorithm, signingCertificateChain, signingPrivateKey);
+                                request,
+                                AS2ServerConnection.this.signingAlgorithm,
+                                AS2ServerConnection.this.signingCertificateChain,
+                                AS2ServerConnection.this.signingPrivateKey);
 
                         MultipartMimeEntity asyncReceipt = multipartReportEntity;
                         if (gen != null) {
@@ -219,17 +247,6 @@ public void run() {
 
     }
 
-    private RequestListenerThread listenerThread;
-    private final Lock lock = new ReentrantLock();
-    private final String as2Version;
-    private final String originServer;
-    private final String serverFqdn;
-    private final Certificate[] signingCertificateChain;
-    private final PrivateKey signingPrivateKey;
-    private final PrivateKey decryptingPrivateKey;
-    private final Certificate[] validateSigningCertificateChain;
-    private final AS2SignatureAlgorithm signingAlgorithm;
-
     public AS2ServerConnection(String as2Version,
                                String originServer,
                                String serverFqdn,
@@ -250,14 +267,18 @@ public AS2ServerConnection(String as2Version,
         this.signingPrivateKey = signingPrivateKey;
         this.decryptingPrivateKey = decryptingPrivateKey;
         this.validateSigningCertificateChain = validateSigningCertificateChain;
-
         this.signingAlgorithm = signingAlgorithm;
-        listenerThread = new RequestListenerThread(
-                this.as2Version, this.originServer, this.serverFqdn,
-                parserServerPortNumber, signingAlgorithm, this.signingCertificateChain, this.signingPrivateKey,
-                this.decryptingPrivateKey, mdnMessageTemplate, validateSigningCertificateChain, sslContext);
-        listenerThread.setDaemon(true);
-        listenerThread.start();
+
+        listenerService = new RequestListenerService(
+                this.as2Version,
+                this.originServer,
+                this.serverFqdn,
+                mdnMessageTemplate,
+                validateSigningCertificateChain);
+
+        acceptorThread = new RequestAcceptorThread(parserServerPortNumber, sslContext, listenerService);
+        acceptorThread.setDaemon(true);
+        acceptorThread.start();
     }
 
     public Certificate[] getValidateSigningCertificateChain() {
@@ -273,15 +294,19 @@ public PrivateKey getDecryptingPrivateKey() {
     }
 
     public void close() {
-        if (listenerThread != null) {
+        if (acceptorThread != null) {
             lock.lock();
             try {
                 try {
-                    listenerThread.serversocket.close();
+                    // 3. Close the shared ServerSocket
+                    if (serversocket != null) {
+                        serversocket.close();
+                    }
                 } catch (IOException e) {
                     LOG.debug(e.getMessage(), e);
                 } finally {
-                    listenerThread = null;
+                    acceptorThread = null;
+                    listenerService = null;
                 }
             } finally {
                 lock.unlock();
@@ -290,21 +315,21 @@ public void close() {
     }
 
     public void listen(String requestUri, HttpRequestHandler handler) {
-        if (listenerThread != null) {
+        if (listenerService != null) {
             lock.lock();
             try {
-                listenerThread.registerHandler(requestUri, handler);
+                listenerService.registerHandler(requestUri, handler);
             } finally {
                 lock.unlock();
             }
         }
     }
 
     public void unlisten(String requestUri) {
-        if (listenerThread != null) {
+        if (listenerService != null) {
             lock.lock();
             try {
-                listenerThread.unregisterHandler(requestUri);
+                listenerService.unregisterHandler(requestUri);
             } finally {
                 lock.unlock();
             }

components/camel-as2/camel-as2-component/src/main/java/org/apache/camel/component/as2/AS2Consumer.java

@@ -34,7 +34,6 @@
 import org.apache.camel.support.component.AbstractApiConsumer;
 import org.apache.camel.support.component.ApiConsumerHelper;
 import org.apache.camel.support.component.ApiMethod;
-import org.apache.camel.support.component.ApiMethodHelper;
 import org.apache.hc.core5.http.ClassicHttpRequest;
 import org.apache.hc.core5.http.ClassicHttpResponse;
 import org.apache.hc.core5.http.HttpEntityContainer;
@@ -98,18 +97,23 @@ protected void doStart() throws Exception {
         as2ServerConnection = getEndpoint().getAS2ServerConnection();
         apiProxy = new AS2ServerManager(as2ServerConnection);
 
-        // invoke the API method to start listening
-        ApiMethodHelper.invokeMethod(apiProxy, apiMethod, properties);
+        String uri = properties.computeIfAbsent("requestUriPattern", param -> "/").toString();
+
+        as2ServerConnection.listen(uri, this);
     }
 
     @Override
     protected void doStop() throws Exception {
-        super.doStop();
 
-        if (apiProxy != null) {
-            String uri = properties.get("requestUriPattern").toString();
-            apiProxy.unlisten(uri);
+        if (as2ServerConnection != null) {
+            // Resolve the unique URI pattern for this consumer
+            String uri = properties.computeIfAbsent("requestUriPattern", param -> "/").toString();
+
+            // Unregister this consumer from the shared AS2ServerConnection
+            as2ServerConnection.unlisten(uri);
         }
+
+        super.doStop();
     }
 
     @Override
@@ -125,8 +129,8 @@ public void handle(ClassicHttpRequest request, ClassicHttpResponse response, Htt
             ApplicationEntity ediEntity
                     = HttpMessageUtils.extractEdiPayload(request,
                             new HttpMessageUtils.DecrpytingAndSigningInfo(
-                                    as2ServerConnection.getValidateSigningCertificateChain(),
-                                    as2ServerConnection.getDecryptingPrivateKey()));
+                                    getEndpoint().getValidateSigningCertificateChain(),
+                                    getEndpoint().getDecryptingPrivateKey()));
 
             // Set AS2 Interchange property and EDI message into body of input message.
             Exchange exchange = createExchange(false);

components/camel-as2/camel-as2-component/src/main/java/org/apache/camel/component/as2/AS2Endpoint.java

@@ -198,6 +198,22 @@ public void setSigningCertificateChain(Certificate[] signingCertificateChain) {
         configuration.setSigningCertificateChain(signingCertificateChain);
     }
 
+    public Certificate[] getValidateSigningCertificateChain() {
+        return configuration.getValidateSigningCertificateChain();
+    }
+
+    public void setValidateSigningCertificateChain(Certificate[] validateSigningCertificateChain) {
+        configuration.setValidateSigningCertificateChain(validateSigningCertificateChain);
+    }
+
+    public PrivateKey getDecryptingPrivateKey() {
+        return configuration.getDecryptingPrivateKey();
+    }
+
+    public void setDecryptingPrivateKey(PrivateKey decryptingPrivateKey) {
+        configuration.setDecryptingPrivateKey(decryptingPrivateKey);
+    }
+
     public PrivateKey getSigningPrivateKey() {
         return configuration.getSigningPrivateKey();
     }

components/camel-as2/camel-as2-component/src/main/java/org/apache/camel/component/as2/internal/AS2ConnectionHelper.java

@@ -103,9 +103,12 @@ public static AS2ServerConnection createAS2ServerConnection(AS2Configuration con
                                     configuration.getAs2Version(), configuration.getServer(),
                                     configuration.getServerFqdn(), configuration.getServerPortNumber(),
                                     configuration.getSigningAlgorithm(),
-                                    configuration.getSigningCertificateChain(), configuration.getSigningPrivateKey(),
-                                    configuration.getDecryptingPrivateKey(), configuration.getMdnMessageTemplate(),
-                                    configuration.getValidateSigningCertificateChain(), configuration.getSslContext());
+                                    configuration.getSigningCertificateChain(),
+                                    configuration.getSigningPrivateKey(),
+                                    configuration.getDecryptingPrivateKey(),
+                                    configuration.getMdnMessageTemplate(),
+                                    configuration.getValidateSigningCertificateChain(),
+                                    configuration.getSslContext());
                         } catch (IOException e) {
                             throw new UncheckedIOException(e);
                         }