@@ -81,14 +98,40 @@ public OpcUaClient getOpcUaClient() { } protected void init() { - this.manager = new SubscriptionManager(this.configuration, Stack.sharedScheduledExecutor(), 10_000); + ObjectHelper.notNull(this.camelContext, "camelContext"); + + // A failed request disconnects the client, which can wait for a reconnect attempt that Milo schedules on its shared + // scheduled executor, so this connection handles its requests and reconnects on an executor of its own. + final ThreadPoolProfile profile = new ThreadPoolProfile("MiloClient"); + // as many threads as that executor; with one processor or with virtual threads this is one thread, so a route + // that continues on it after an awaited request must not wait for another request of this connection + profile.setPoolSize(Runtime.getRuntime().availableProcessors()); + // not bounded, like that executor, because a rejected reconnect would end the reconnects of this connection + profile.setMaxQueueSize(-1); + // a request that completes after the connection is closed fails instead of being dropped + profile.setRejectedPolicy(ThreadPoolRejectedPolicy.Abort); + this.executor = this.camelContext.getExecutorServiceManager().newScheduledThreadPool(this, "MiloClient", profile); + + this.manager = new SubscriptionManager(this.configuration, this.executor, 10_000); } @Override public void close() throws Exception { - if (this.manager != null) { - this.manager.dispose(); - this.manager = null; + lock.lock(); + try { + try { + if (this.manager != null) { + this.manager.dispose(); + this.manager = null; + } + } finally { + if (this.executor != null) { + this.camelContext.getExecutorServiceManager().shutdown(this.executor); + this.executor = null; + } + } + } finally { + lock.unlock(); } } diff --git a/components/camel-milo/src/main/java/org/apache/camel/component/milo/client/MiloClientEndpoint.java b/components/camel-milo/src/main/java/org/apache/camel/component/milo/client/MiloClientEndpoint.java index afc7cc34ec2fe..254b612f2f6b6 100644 --- a/components/camel-milo/src/main/java/org/apache/camel/component/milo/client/MiloClientEndpoint.java +++ b/components/camel-milo/src/main/java/org/apache/camel/component/milo/client/MiloClientEndpoint.java @@ -145,7 +145,10 @@ public Consumer createConsumer(final Processor processor) throws Exception { } public MiloClientConnection createConnection() { - return this.connectionManager.createConnection(configuration, monitorFilterConfiguration); + final MiloClientConnection connection + = this.connectionManager.createConnection(configuration, monitorFilterConfiguration); + connection.setCamelContext(getCamelContext()); + return connection; } public void releaseConnection(MiloClientConnection connection) { diff --git a/components/camel-milo/src/main/java/org/apache/camel/component/milo/client/internal/SubscriptionManager.java b/components/camel-milo/src/main/java/org/apache/camel/component/milo/client/internal/SubscriptionManager.java index e8dab0b933369..314a55b4625db 100644 --- a/components/camel-milo/src/main/java/org/apache/camel/component/milo/client/internal/SubscriptionManager.java +++ b/components/camel-milo/src/main/java/org/apache/camel/component/milo/client/internal/SubscriptionManager.java @@ -834,8 +834,8 @@ private void recoverSubscription(final OpcUaSubscription subscription, final Sub * The executor that the client of a subscription delivers its values on, which handles the outcome of a check or a * delete. Creating the subscription again and rebuilding the connection block until the server answers, and a * request sent while the connection is lost waits for the channel until a reconnect attempt restores it or fails. - * The client schedules those attempts on the shared scheduled executor of Milo, which {@link #executor} is as well - * and which has a single thread on a machine with one processor, so the wait must not hold one of its threads. + * The client schedules those attempts on the shared scheduled executor of Milo, which has a single thread on a + * machine with one processor, so the wait must not hold one of its threads. *
* Milo asks that blocking calls are not made on the threads of this executor either, because they complete the * responses. The executor of the transport is not set here, so this is the shared executor of Milo, a cached pool diff --git a/components/camel-milo/src/test/java/org/apache/camel/component/milo/MiloClientExecutorTest.java b/components/camel-milo/src/test/java/org/apache/camel/component/milo/MiloClientExecutorTest.java new file mode 100644 index 0000000000000..a4be2a00135be --- /dev/null +++ b/components/camel-milo/src/test/java/org/apache/camel/component/milo/MiloClientExecutorTest.java @@ -0,0 +1,127 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.camel.component.milo; + +import java.util.List; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ScheduledThreadPoolExecutor; +import java.util.concurrent.TimeUnit; + +import org.apache.camel.EndpointInject; +import org.apache.camel.RoutesBuilder; +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.component.milo.server.MiloServerComponent; +import org.apache.camel.component.mock.MockEndpoint; +import org.eclipse.milo.opcua.stack.core.Stack; +import org.eclipse.milo.opcua.stack.core.types.builtin.Variant; +import org.junit.jupiter.api.Test; + +import static org.apache.camel.component.milo.NodeIds.nodeValue; +import static org.awaitility.Awaitility.await; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Tests the executor that a client connection handles the outcome of its requests on. + */ +public class MiloClientExecutorTest extends AbstractMiloServerTest { + + private static final String MILO_CLIENT_ITEM + = "milo-client:opc.tcp://foo:bar@localhost:@@port@@?node=" + + nodeValue(MiloServerComponent.DEFAULT_NAMESPACE_URI, "myitem1") + "&overrideHost=true"; + + @EndpointInject("mock:item") + protected MockEndpoint item; + + @Override + protected RoutesBuilder createRouteBuilder() { + return new RouteBuilder() { + @Override + public void configure() { + from("milo-server:myitem1").to("mock:item"); + from("direct:write").routeId("write").to(resolve(MILO_CLIENT_ITEM)); + } + }; + } + + /** + * A failed request disconnects the client, and the disconnect can wait for a reconnect attempt that the client + * schedules on the shared scheduled executor of Milo, which has a single thread on a machine with one processor. + * The outcome of a request is therefore not handled on that executor: an awaited write completes while all of its + * threads are busy. + */ + @Test + void awaitedWriteCompletesWhileTheSharedScheduledExecutorOfMiloIsBusy() throws Exception { + item.expectedMessageCount(2); + testBody(item.message(0), assertGoodValue("Foo1")); + testBody(item.message(1), assertGoodValue("Foo2")); + + // connects + write("Foo1"); + + final ScheduledThreadPoolExecutor scheduler = (ScheduledThreadPoolExecutor) Stack.sharedScheduledExecutor(); + final CountDownLatch busy = new CountDownLatch(scheduler.getCorePoolSize()); + final CountDownLatch release = new CountDownLatch(1); + for (int i = 0; i < scheduler.getCorePoolSize(); i++) { + scheduler.execute(() -> { + busy.countDown(); + try { + release.await(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + }); + } + try { + assertTrue(busy.await(30, TimeUnit.SECONDS), "All the threads of the executor should be busy"); + + final CompletableFuture