| /* |
| 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.plc4x.java.utils.connectionpool; |
| |
| import org.apache.commons.lang3.reflect.FieldUtils; |
| import org.apache.commons.pool2.impl.GenericKeyedObjectPool; |
| import org.apache.commons.pool2.impl.GenericKeyedObjectPoolConfig; |
| import org.apache.plc4x.java.api.PlcConnection; |
| import org.apache.plc4x.java.api.authentication.PlcAuthentication; |
| import org.apache.plc4x.java.api.authentication.PlcUsernamePasswordAuthentication; |
| import org.apache.plc4x.java.api.exceptions.PlcConnectionException; |
| import org.apache.plc4x.java.api.exceptions.PlcUnsupportedOperationException; |
| import org.apache.plc4x.java.api.messages.PlcReadRequest; |
| import org.apache.plc4x.java.api.messages.PlcSubscriptionRequest; |
| import org.apache.plc4x.java.api.messages.PlcUnsubscriptionRequest; |
| import org.apache.plc4x.java.api.messages.PlcWriteRequest; |
| import org.apache.plc4x.java.api.metadata.PlcConnectionMetadata; |
| import org.apache.plc4x.java.spi.PlcDriver; |
| import org.assertj.core.api.WithAssertions; |
| import org.junit.jupiter.api.AfterEach; |
| import org.junit.jupiter.api.BeforeEach; |
| import org.junit.jupiter.api.Test; |
| import org.junit.jupiter.api.extension.ExtendWith; |
| import org.mockito.Answers; |
| import org.mockito.Mock; |
| import org.mockito.junit.jupiter.MockitoExtension; |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| |
| import java.util.LinkedList; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.concurrent.*; |
| import java.util.concurrent.atomic.AtomicBoolean; |
| import java.util.stream.IntStream; |
| |
| import static org.mockito.ArgumentMatchers.any; |
| import static org.mockito.ArgumentMatchers.anyString; |
| import static org.mockito.Mockito.*; |
| |
| @ExtendWith(MockitoExtension.class) |
| class PooledPlcDriverManagerTest implements WithAssertions { |
| |
| private static Logger LOGGER = LoggerFactory.getLogger(PooledPlcDriverManagerTest.class); |
| |
| private PooledPlcDriverManager SUT = new PooledPlcDriverManager(pooledPlcConnectionFactory -> { |
| GenericKeyedObjectPoolConfig<PlcConnection> config = new GenericKeyedObjectPoolConfig<>(); |
| config.setMinIdlePerKey(1); |
| config.setTestOnBorrow(true); |
| config.setTestOnReturn(true); |
| return new GenericKeyedObjectPool<>(pooledPlcConnectionFactory, config); |
| }); |
| |
| @Mock(answer = Answers.RETURNS_DEEP_STUBS) |
| PlcDriver plcDriver; |
| |
| private ExecutorService executorService; |
| |
| @SuppressWarnings("unchecked") |
| @BeforeEach |
| void setUp() throws Exception { |
| Map<String, PlcDriver> driverMap = (Map) FieldUtils.getField(PooledPlcDriverManager.class, "driverMap", true).get(SUT); |
| driverMap.put("dummydummy", plcDriver); |
| executorService = Executors.newFixedThreadPool(100); |
| |
| assertThat(SUT.getStatistics()).containsOnly( |
| entry("pools.count", 0), |
| entry("numActive", 0), |
| entry("numIdle", 0) |
| ); |
| } |
| |
| @AfterEach |
| void tearDown() { |
| executorService.shutdown(); |
| } |
| |
| @Test |
| void getConnection() throws Exception { |
| when(plcDriver.connect(anyString())).then(invocationOnMock -> new DummyPlcConnection(invocationOnMock.getArgument(0))); |
| |
| LinkedList<Callable<PlcConnection>> callables = new LinkedList<>(); |
| |
| // This: should result in one open connection |
| IntStream.range(0, 8).forEach(i -> callables.add(() -> { |
| try { |
| return SUT.getConnection("dummydummy:single/socket1/socket2?fancyOption=true"); |
| } catch (PlcConnectionException e) { |
| throw new RuntimeException(e); |
| } |
| })); |
| |
| // This should result in five open connections |
| IntStream.range(0, 5).forEach(i -> callables.add(() -> { |
| try { |
| return SUT.getConnection("dummydummy:multi-" + i + "/socket1/socket2?fancyOption=true"); |
| } catch (PlcConnectionException e) { |
| throw new RuntimeException(e); |
| } |
| })); |
| |
| List<Future<PlcConnection>> futures = executorService.invokeAll(callables); |
| |
| // Wait for existing connections |
| futures.forEach(plcConnectionFuture1 -> { |
| try { |
| plcConnectionFuture1.get(); |
| } catch (InterruptedException | ExecutionException e) { |
| throw new RuntimeException(e); |
| } |
| }); |
| LOGGER.info("Statistics after execution {}", SUT.getStatistics()); |
| |
| // As we have a pool size of 8 we should have only 8 + 5 calls for the separate pools |
| verify(plcDriver, times(13)).connect(anyString()); |
| |
| assertThat(SUT.getStatistics()).contains( |
| entry("PoolKey{url='dummydummy:single/socket1/socket2?fancyOption=true'}.numActive", 8) |
| ); |
| |
| futures.forEach(plcConnectionFuture -> { |
| try { |
| plcConnectionFuture.get().close(); |
| } catch (Exception e) { |
| throw new RuntimeException(e); |
| } |
| }); |
| |
| assertThat(SUT.getStatistics()).contains( |
| entry("PoolKey{url='dummydummy:single/socket1/socket2?fancyOption=true'}.numActive", 0) |
| ); |
| } |
| |
| @Test |
| void getConnectionWithAuth() throws Exception { |
| when(plcDriver.connect(anyString(), any())).then(invocationOnMock -> new DummyPlcConnection(invocationOnMock.getArgument(0), invocationOnMock.getArgument(1))); |
| |
| LinkedList<Callable<PlcConnection>> callables = new LinkedList<>(); |
| |
| // This: should result in one open connection |
| IntStream.range(0, 8).forEach(i -> callables.add(() -> { |
| try { |
| return SUT.getConnection("dummydummy:single/socket1/socket2?fancyOption=true", new PlcUsernamePasswordAuthentication("user", "passwordp954368564098ß")); |
| } catch (PlcConnectionException e) { |
| throw new RuntimeException(e); |
| } |
| })); |
| |
| // This should result in five open connections |
| IntStream.range(0, 5).forEach(i -> callables.add(() -> { |
| try { |
| return SUT.getConnection("dummydummy:multi-" + i + "/socket1/socket2?fancyOption=true", new PlcUsernamePasswordAuthentication("user", "passwordp954368564098ß")); |
| } catch (PlcConnectionException e) { |
| throw new RuntimeException(e); |
| } |
| })); |
| |
| List<Future<PlcConnection>> futures = executorService.invokeAll(callables); |
| |
| futures.forEach(plcConnectionFuture1 -> { |
| try { |
| plcConnectionFuture1.get(); |
| } catch (InterruptedException | ExecutionException e) { |
| throw new RuntimeException(e); |
| } |
| }); |
| |
| LOGGER.info("Statistics after execution {}", SUT.getStatistics()); |
| |
| // As we have a pool size of 8 we should have only 8 + 5 calls for the separate pools |
| verify(plcDriver, times(13)).connect(anyString(), any()); |
| |
| assertThat(SUT.getStatistics()).contains( |
| entry("PoolKey{url='dummydummy:single/socket1/socket2?fancyOption=true', plcAuthentication=PlcUsernamePasswordAuthentication{username='user', password='*****************'}}.numActive", 8) |
| ); |
| |
| futures.forEach(plcConnectionFuture -> { |
| try { |
| plcConnectionFuture.get().close(); |
| } catch (Exception e) { |
| throw new RuntimeException(e); |
| } |
| }); |
| |
| assertThat(SUT.getStatistics()).contains( |
| entry("PoolKey{url='dummydummy:single/socket1/socket2?fancyOption=true', plcAuthentication=PlcUsernamePasswordAuthentication{username='user', password='*****************'}}.numActive", 0) |
| ); |
| } |
| |
| @Test |
| void connectionInvalidation() throws Exception { |
| when(plcDriver.connect(anyString())).then(invocationOnMock -> new DummyPlcConnection(invocationOnMock.getArgument(0))); |
| |
| PlcConnection connection = SUT.getConnection("dummydummy:single/socket1/socket2?fancyOption=true"); |
| assertThat(connection.isConnected()).isEqualTo(true); |
| assertThat(connection.getMetadata().canRead()).isEqualTo(false); |
| assertThat(connection.getMetadata().canWrite()).isEqualTo(false); |
| assertThat(connection.getMetadata().canSubscribe()).isEqualTo(false); |
| |
| connection.close(); |
| assertThatThrownBy(connection::connect).isInstanceOf(IllegalStateException.class).hasMessage("Proxy not valid anymore"); |
| assertThatThrownBy(connection::isConnected).isInstanceOf(IllegalStateException.class).hasMessage("Proxy not valid anymore"); |
| assertThatThrownBy(connection::close).isInstanceOf(IllegalStateException.class).hasMessage("Proxy not valid anymore"); |
| assertThatThrownBy(connection::getMetadata).isInstanceOf(IllegalStateException.class).hasMessage("Proxy not valid anymore"); |
| assertThatThrownBy(connection::readRequestBuilder).isInstanceOf(IllegalStateException.class).hasMessage("Proxy not valid anymore"); |
| assertThatThrownBy(connection::writeRequestBuilder).isInstanceOf(IllegalStateException.class).hasMessage("Proxy not valid anymore"); |
| assertThatThrownBy(connection::subscriptionRequestBuilder).isInstanceOf(IllegalStateException.class).hasMessage("Proxy not valid anymore"); |
| assertThatThrownBy(connection::unsubscriptionRequestBuilder).isInstanceOf(IllegalStateException.class).hasMessage("Proxy not valid anymore"); |
| } |
| |
| @Test |
| void cleanupOfBrokenConnections() throws Exception { |
| AtomicBoolean failNow = new AtomicBoolean(false); |
| when(plcDriver.connect(anyString())).then(invocationOnMock -> { |
| DummyPlcConnection dummyPlcConnection = spy(new DummyPlcConnection(invocationOnMock.getArgument(0))); |
| // we fake an connection which breaks at this call |
| doAnswer(invocation -> { |
| if (failNow.get()) { |
| throw new PlcConnectionException("blub"); |
| } |
| return invocation.callRealMethod(); |
| }).when(dummyPlcConnection).connect(); |
| return dummyPlcConnection; |
| }); |
| |
| assertThat(SUT.getStatistics()).containsOnly( |
| entry("pools.count", 0), |
| entry("numActive", 0), |
| entry("numIdle", 0) |
| ); |
| PlcConnection connection = SUT.getConnection("dummydummy:breakIt"); |
| assertThat(SUT.getStatistics()).containsOnly( |
| entry("pools.count", 1), |
| entry("numActive", 1), |
| entry("numIdle", 0), |
| entry("PoolKey{url='dummydummy:breakIt'}.numActive", 1) |
| ); |
| failNow.set(true); |
| try { |
| connection.connect(); |
| fail("This should throw an exception"); |
| } catch (Exception e) { |
| // TODO: currently UndeclaredThrowableException is the top one which should be InvocationTargetException |
| //assertThat(e).isInstanceOf(InvocationTargetException.class); |
| assertThat(e).hasRootCauseInstanceOf(PlcConnectionException.class); |
| } |
| // Faulty connection should have been discarded |
| assertThat(SUT.getStatistics()).containsOnly( |
| entry("pools.count", 0), |
| entry("numActive", 0), |
| entry("numIdle", 0) |
| ); |
| } |
| |
| @Test |
| public void testOtherConstructors() { |
| assertThat(new PooledPlcDriverManager()).isNotNull(); |
| assertThat(new PooledPlcDriverManager(new PoolKeyFactory())).isNotNull(); |
| assertThat(new PooledPlcDriverManager(PooledPlcDriverManager.class.getClassLoader())).isNotNull(); |
| assertThat(new PooledPlcDriverManager( |
| PooledPlcDriverManager.class.getClassLoader(), new PoolKeyFactory())).isNotNull(); |
| } |
| |
| class DummyPlcConnection implements PlcConnection, PlcConnectionMetadata { |
| |
| private final String url; |
| |
| private final PlcAuthentication plcAuthentication; |
| |
| boolean connected = false; |
| |
| public DummyPlcConnection(String url) { |
| this(url, null); |
| } |
| |
| public DummyPlcConnection(String url, PlcAuthentication plcAuthentication) { |
| this.url = url; |
| this.plcAuthentication = plcAuthentication; |
| } |
| |
| @Override |
| public void connect() { |
| connected = true; |
| } |
| |
| @Override |
| public CompletableFuture<Void> ping() { |
| CompletableFuture<Void> future = new CompletableFuture<>(); |
| future.completeExceptionally(new UnsupportedOperationException()); |
| return future; |
| } |
| |
| @Override |
| public boolean isConnected() { |
| return connected; |
| } |
| |
| @Override |
| public PlcConnectionMetadata getMetadata() { |
| return this; |
| } |
| |
| @Override |
| public boolean canRead() { |
| return false; |
| } |
| |
| @Override |
| public boolean canWrite() { |
| return false; |
| } |
| |
| @Override |
| public boolean canSubscribe() { |
| return false; |
| } |
| |
| @Override |
| public void close() { |
| connected = false; |
| } |
| |
| @Override |
| public PlcReadRequest.Builder readRequestBuilder() { |
| throw new PlcUnsupportedOperationException("The connection does not support reading"); |
| } |
| |
| @Override |
| public PlcWriteRequest.Builder writeRequestBuilder() { |
| throw new PlcUnsupportedOperationException("The connection does not support writing"); |
| } |
| |
| @Override |
| public PlcSubscriptionRequest.Builder subscriptionRequestBuilder() { |
| throw new PlcUnsupportedOperationException("The connection does not support subscription"); |
| } |
| |
| @Override |
| public PlcUnsubscriptionRequest.Builder unsubscriptionRequestBuilder() { |
| throw new PlcUnsupportedOperationException("The connection does not support subscription"); |
| } |
| |
| @Override |
| public String toString() { |
| return "DummyPlcConnection{" + |
| "url='" + url + '\'' + |
| ", plcAuthentication=" + plcAuthentication + |
| ", connected=" + connected + |
| '}'; |
| } |
| } |
| } |
