|
36 | 36 | import java.util.concurrent.atomic.AtomicReference; |
37 | 37 | import java.util.function.Function; |
38 | 38 |
|
| 39 | +import static java.util.concurrent.TimeUnit.MILLISECONDS; |
| 40 | +import static java.util.concurrent.TimeUnit.SECONDS; |
| 41 | + |
39 | 42 | @RunWith(VertxUnitRunner.class) |
40 | 43 | public abstract class MetricsTestBase { |
41 | 44 |
|
@@ -81,6 +84,8 @@ protected Pool getPool() { |
81 | 84 | return pool; |
82 | 85 | } |
83 | 86 |
|
| 87 | + protected abstract SqlConnectOptions connectOptions(); |
| 88 | + |
84 | 89 | protected abstract ClientBuilder<Pool> poolBuilder(); |
85 | 90 |
|
86 | 91 | protected Pool createPool(Vertx vertx) { |
@@ -122,6 +127,15 @@ public void close() { |
122 | 127 |
|
123 | 128 | @Test |
124 | 129 | public void testQueuing(TestContext ctx) throws Exception { |
| 130 | + testQueuing(ctx, false); |
| 131 | + } |
| 132 | + |
| 133 | + @Test |
| 134 | + public void testQueuingTimeout(TestContext ctx) throws Exception { |
| 135 | + testQueuing(ctx, true); |
| 136 | + } |
| 137 | + |
| 138 | + private void testQueuing(TestContext ctx, boolean timeout) throws Exception { |
125 | 139 | AtomicInteger queueSize = new AtomicInteger(); |
126 | 140 | AtomicInteger inUse = new AtomicInteger(); |
127 | 141 | List<Object> enqueueMetrics = Collections.synchronizedList(new ArrayList<>()); |
@@ -156,28 +170,106 @@ public void end(Object inUseMetric, boolean succeeded) { |
156 | 170 | endMetrics.add(inUseMetric); |
157 | 171 | } |
158 | 172 | }; |
159 | | - Pool pool = createPool(vertx, new PoolOptions().setMaxSize(1).setName("the-pool")); |
160 | | - SqlConnection conn = pool.getConnection().toCompletionStage().toCompletableFuture().get(20, TimeUnit.SECONDS); |
| 173 | + PoolOptions poolOptions = new PoolOptions().setMaxSize(1).setName("the-pool"); |
| 174 | + if (timeout) { |
| 175 | + poolOptions.setConnectionTimeout(2).setConnectionTimeoutUnit(SECONDS); |
| 176 | + } |
| 177 | + Pool pool = createPool(vertx, poolOptions); |
| 178 | + SqlConnection conn = Future.await(pool.getConnection(), 20, SECONDS); |
161 | 179 | int num = 16; |
162 | 180 | List<Future<?>> futures = new ArrayList<>(); |
163 | 181 | for (int i = 0;i < num;i++) { |
164 | | - futures.add(pool.query("SELECT * FROM immutable WHERE id=1").execute()); |
165 | | - } |
166 | | - long now = System.currentTimeMillis(); |
167 | | - while (queueSize.get() != num) { |
168 | | - ctx.assertTrue(System.currentTimeMillis() - now < 20_000); |
169 | | - Thread.sleep(100); |
| 182 | + futures.add(pool.withConnection(sqlConn -> sqlConn.query("SELECT * FROM immutable WHERE id=1").execute())); |
170 | 183 | } |
171 | | - conn.close(); |
172 | | - Future.join(futures).toCompletionStage().toCompletableFuture().get(20, TimeUnit.SECONDS); |
| 184 | + awaitQueueSize(ctx, queueSize, timeout ? 0 : num); |
| 185 | + Future.await(conn.close(), 20, SECONDS); |
| 186 | + Future.await(Future.join(futures).otherwiseEmpty(), 20, SECONDS); |
173 | 187 | ctx.assertEquals(0, queueSize.get()); |
174 | 188 | ctx.assertEquals(0, inUse.get()); |
175 | | - ctx.assertEquals(enqueueMetrics, dequeueMetrics); |
| 189 | + if (timeout) { |
| 190 | + ctx.assertTrue(enqueueMetrics.containsAll(dequeueMetrics) && dequeueMetrics.containsAll(enqueueMetrics)); |
| 191 | + } else { |
| 192 | + ctx.assertEquals(enqueueMetrics, dequeueMetrics); |
| 193 | + } |
176 | 194 | ctx.assertEquals(beginMetrics, endMetrics); |
177 | 195 | ctx.assertEquals("sql", poolType); |
178 | 196 | ctx.assertEquals("the-pool", poolName); |
179 | 197 | } |
180 | 198 |
|
| 199 | + private void awaitQueueSize(TestContext ctx, AtomicInteger queueSize, int num) throws InterruptedException { |
| 200 | + long now = System.currentTimeMillis(); |
| 201 | + for (; ; ) { |
| 202 | + if (queueSize.get() != num) { |
| 203 | + if (System.currentTimeMillis() - now >= 20_000) { |
| 204 | + ctx.fail("Timeout waiting for queue size " + queueSize.get() + " to be equal to " + num); |
| 205 | + } else { |
| 206 | + MILLISECONDS.sleep(500); |
| 207 | + } |
| 208 | + } else { |
| 209 | + break; |
| 210 | + } |
| 211 | + } |
| 212 | + } |
| 213 | + |
| 214 | + @Test |
| 215 | + public void testConnectionLost(TestContext ctx) throws Exception { |
| 216 | + SqlConnectOptions connectOptions = connectOptions(); |
| 217 | + ProxyServer proxy = ProxyServer.create(vertx, connectOptions.getPort(), connectOptions.getHost()); |
| 218 | + AtomicReference<ProxyServer.Connection> firstConnection = new AtomicReference<>(); |
| 219 | + proxy.proxyHandler(proxiedConn -> { |
| 220 | + if (firstConnection.compareAndSet(null, proxiedConn)) { |
| 221 | + proxiedConn.connect(); |
| 222 | + } |
| 223 | + }); |
| 224 | + // Start proxy |
| 225 | + Async listenLatch = ctx.async(); |
| 226 | + proxy.listen(8080, "localhost", ctx.asyncAssertSuccess(res -> listenLatch.complete())); |
| 227 | + listenLatch.awaitSuccess(20_000); |
| 228 | + |
| 229 | + |
| 230 | + AtomicInteger queueSize = new AtomicInteger(); |
| 231 | + poolMetrics = new PoolMetrics() { |
| 232 | + @Override |
| 233 | + public Object submitted() { |
| 234 | + queueSize.incrementAndGet(); |
| 235 | + return null; |
| 236 | + } |
| 237 | + |
| 238 | + @Override |
| 239 | + public Object begin(Object o) { |
| 240 | + queueSize.decrementAndGet(); |
| 241 | + return null; |
| 242 | + } |
| 243 | + |
| 244 | + @Override |
| 245 | + public void rejected(Object o) { |
| 246 | + queueSize.decrementAndGet(); |
| 247 | + } |
| 248 | + }; |
| 249 | + PoolOptions poolOptions = new PoolOptions() |
| 250 | + .setConnectionTimeout(500) |
| 251 | + .setConnectionTimeoutUnit(MILLISECONDS) |
| 252 | + .setMaxSize(1) |
| 253 | + .setName("the-pool"); |
| 254 | + Pool pool = poolBuilder() |
| 255 | + .with(poolOptions) |
| 256 | + .using(vertx) |
| 257 | + .connectingTo(connectOptions.setHost("localhost").setPort(8080)) |
| 258 | + .build(); |
| 259 | + SqlConnection conn = Future.await(pool.getConnection(), 20, SECONDS); |
| 260 | + int num = 16; |
| 261 | + Async async = ctx.async(num + 1); |
| 262 | + for (int i = 0; i < num; i++) { |
| 263 | + pool.withConnection(sqlConn -> sqlConn.query("SELECT * FROM immutable WHERE id=1").execute()) |
| 264 | + .onComplete(ctx.asyncAssertFailure(t -> async.countDown())); |
| 265 | + } |
| 266 | + conn.closeHandler(v -> async.countDown()); |
| 267 | + awaitQueueSize(ctx, queueSize, 16); |
| 268 | + firstConnection.get().clientSocket().close(); |
| 269 | + async.await(20_000); |
| 270 | + ctx.assertEquals(0, queueSize.get()); |
| 271 | + } |
| 272 | + |
181 | 273 | @Test |
182 | 274 | public void testSimpleQuery(TestContext ctx) { |
183 | 275 | Function<SqlConnection, Future<?>> fn = conn -> conn.query("SELECT * FROM immutable WHERE id=1").execute(); |
|
0 commit comments