PoolWatchThread implements Runnable { public void run() { int maxNewConnections; while (!this.signalled){ maxNewConnections=0; try{ if (this.lazyInit){ this.partition.getPoolWatchThreadSignalQueue().take(); } maxNewConnections = this.partition.getMaxConnections()-this.partition.getCreatedConnections(); while (maxNewConnections == 0 || (this.partition.getAvailableConnections() *100/this.partition.getMaxConnections() > this.poolAvailabilityThreshold)){ if (maxNewConnections == 0){ this.partition.setUnableToCreateMoreTransactions(true); } this.partition.getPoolWatchThreadSignalQueue().take(); maxNewConnections = this.partition.getMaxConnections()-this.partition.getCreatedConnections(); } if (maxNewConnections > 0 && !this.pool.poolShuttingDown){ fillConnections(Math.min(maxNewConnections, this.partition.getAcquireIncrement())); if (this.partition.getCreatedConnections() < this.partition.getMinConnections()){ fillConnections(this.partition.getMinConnections() - this.partition.getCreatedConnections() ); } } if (this.pool.poolShuttingDown){ return; } } catch (InterruptedException e) { logger.debug("Terminating pool watch thread"); return; } } } PoolWatchThread(ConnectionPartition connectionPartition, BoneCP pool); void run();  }
@SuppressWarnings({ "unchecked", "rawtypes" }) @Test public void testRunFullConnections() throws InterruptedException{ BlockingQueue<Object> bq = new ArrayBlockingQueue<Object>(1); bq.add(new Object()); expect(mockPartition.getPoolWatchThreadSignalQueue()).andReturn(bq); expect(mockPartition.getMaxConnections()).andReturn(5).once(); expect(mockPartition.getCreatedConnections()).andReturn(5).once(); mockPartition.setUnableToCreateMoreTransactions(true); expectLastCall().once(); BlockingQueue<?> mockQueue = EasyMock.createNiceMock(BlockingQueue.class); expect(mockPartition.getPoolWatchThreadSignalQueue()).andReturn((BlockingQueue) mockQueue); expect(mockQueue.take()).andThrow(new InterruptedException()); replay(mockPartition, mockPool, mockLogger, mockQueue); testClass.run(); EasyMock.verify(mockPartition); }
@SuppressWarnings({ "unchecked", "rawtypes" }) @Test public void testRunLazyInit() throws InterruptedException{ BlockingQueue<Object> bq = new ArrayBlockingQueue<Object>(1); bq.add(new Object()); expect(mockPartition.getPoolWatchThreadSignalQueue()).andReturn(bq); expect(mockPartition.getMaxConnections()).andReturn(5).once(); expect(mockPartition.getCreatedConnections()).andReturn(5).once(); mockPartition.setUnableToCreateMoreTransactions(true); expectLastCall().once(); BlockingQueue<?> mockQueue = EasyMock.createNiceMock(BlockingQueue.class); expect(mockPartition.getPoolWatchThreadSignalQueue()).andReturn((BlockingQueue) mockQueue); expect(mockQueue.take()).andThrow(new InterruptedException()); replay(mockPartition, mockPool, mockLogger, mockQueue); testClass.lazyInit = true; testClass.run(); EasyMock.verify(mockPartition); }
@Test public void testRunCreateConnections() throws InterruptedException, SecurityException, NoSuchFieldException, IllegalArgumentException, IllegalAccessException, SQLException{ expect(mockLogger.isDebugEnabled()).andReturn(true).anyTimes(); LinkedBlockingQueue<ConnectionHandle> fakeConnections = new LinkedBlockingQueue<ConnectionHandle>(100); expect(mockPartition.getMaxConnections()).andAnswer(new IAnswer<Integer>() { public Integer answer() throws Throwable { if (first) { first=false; return 4; } Field field = testClass.getClass().getDeclaredField("signalled"); field.setAccessible(true); field.setBoolean(testClass, true); return 4; } }).anyTimes(); expect(mockPartition.getFreeConnections()).andReturn(fakeConnections).anyTimes(); expect(mockPartition.getAcquireIncrement()).andReturn(1); expect(mockPartition.getUrl()).andReturn(CommonTestUtils.url).anyTimes(); expect(mockPartition.getPassword()).andReturn(CommonTestUtils.password).anyTimes(); expect(mockPartition.getUsername()).andReturn(CommonTestUtils.username).anyTimes(); expect(mockPartition.getAvailableConnections()).andReturn(fakeConnections.size()).anyTimes(); mockPartition.addFreeConnection((ConnectionHandle)anyObject()); expectLastCall().once(); expect(mockPool.obtainInternalConnection((ConnectionHandle)anyObject())).andReturn(EasyMock.createNiceMock(Connection.class)).anyTimes(); expect(mockPool.getDbIsDown()).andReturn(new AtomicBoolean()).anyTimes(); expect(mockPool.getConfig()).andReturn(mockConfig).anyTimes(); replay(mockPool, mockPartition, mockLogger); testClass.run(); EasyMock.verify(mockPartition); reset(mockPartition, mockPool, mockLogger); resetSignalled(); first = true; replay(mockPartition, mockLogger); try{ testClass.run(); Assert.fail("Exception should have been thrown"); } catch (RuntimeException e){ } EasyMock.verify(mockPartition); reset(mockPool, mockLogger, mockConfig); reset(mockPartition); resetSignalled(); first = true; expect(mockPartition.getFreeConnections()).andReturn(fakeConnections).anyTimes(); expect(mockPartition.getMaxConnections()).andAnswer(new IAnswer<Integer>() { public Integer answer() throws Throwable { if (first) { first=false; return 4; } Field field = testClass.getClass().getDeclaredField("signalled"); field.setAccessible(true); field.setBoolean(testClass, true); return 4; } }).anyTimes(); expect(mockConfig.getStatementsCacheSize()).andAnswer(new IAnswer<Integer>() { public Integer answer() throws Throwable { throw new SQLException(); } }).once(); expect(mockPartition.getAcquireIncrement()).andReturn(1).anyTimes(); expect(mockPartition.getAvailableConnections()).andReturn(fakeConnections.size()).anyTimes(); expect(mockPool.getConfig()).andReturn(mockConfig).anyTimes(); expect(mockConfig.getAcquireRetryAttempts()).andReturn(0).anyTimes(); mockLogger.error((String)anyObject(), anyObject()); expectLastCall(); expect(mockPool.getDbIsDown()).andReturn(new AtomicBoolean()).anyTimes(); replay(mockPartition, mockPool, mockLogger, mockConfig); testClass.run(); EasyMock.verify(mockPartition); }